⚒ sindri

sindri / master / tests / test_events.py

test_events.py

            
                from sindri.events import EventBus, PushEvent, IssueOpenedEvent, PROpenedEvent, CIFailedEvent


                def test_push_event_type():
                    e = PushEvent(repo="myrepo", branch="main", commits=[], pusher="alex")
                    assert e.event_type == "push"


                def test_issue_opened_event_type():
                    e = IssueOpenedEvent(repo="myrepo", issue_id=1, title="Fix bug", body="crashes")
                    assert e.event_type == "issue.opened"


                def test_eventbus_dispatches_to_handler(db):
                    received = []
                    bus = EventBus(db=db)
                    bus.on("push", lambda e: received.append(e))
                    bus.emit(PushEvent(repo="myrepo", branch="main", commits=[], pusher="alex"))
                    assert len(received) == 1
                    assert received[0].repo == "myrepo"


                def test_eventbus_stores_event_in_db(db):
                    bus = EventBus(db=db)
                    bus.emit(IssueOpenedEvent(repo="myrepo", issue_id=1, title="Fix bug", body=""))
                    stored = db.list_events("myrepo")
                    assert len(stored) == 1
                    assert stored[0].type == "issue.opened"