swh_scheduler = <swh.scheduler.backend.SchedulerBackend object at 0x7f95ec3f2080> def test_journal_client_origin_visit_status_from_journal_last_failed(swh_scheduler): visit_statuses = [ { "origin": "foo", "visit": 1, "status": "partial", "date": utcnow(), "type": "git", "snapshot": None, }, { "origin": "bar", "visit": 1, "status": "full", "date": DATE1, "type": "git", "snapshot": None, }, { "origin": "bar", "visit": 2, "status": "full", "date": DATE2, "type": "git", "snapshot": None, }, { "origin": "bar", "visit": 3, "status": "full", "date": DATE3, "type": "git", "snapshot": None, }, ] process_journal_objects( {"origin_visit_status": visit_statuses}, scheduler=swh_scheduler ) actual_origin_visit_stats = swh_scheduler.origin_visit_stats_get([("bar", "git")]) assert_visit_stats_ok( actual_origin_visit_stats, [ OriginVisitStats( url="bar", visit_type="git", last_eventful=None, last_uneventful=None, last_failed=DATE3, last_notfound=None, last_snapshot=None, next_position_offset=7, > successive_visits=3, ) ], ) .tox/py3/lib/python3.7/site-packages/swh/scheduler/tests/test_journal_client.py:281: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ actual_visit_stats = [] expected_visit_stats = [OriginVisitStats(url='bar', visit_type='git', last_eventful=None, last_uneventful=None, last_failed=datetime.datetime... last_scheduled=None, last_snapshot=None, next_visit_queue_position=None, next_position_offset=7, successive_visits=3)] compare_successive_visits = True def assert_visit_stats_ok( actual_visit_stats, expected_visit_stats, compare_successive_visits=True ): """Utility test function to ensure visits stats read from the backend are in the right shape. The comparison on the next_visit_queue_position will be dealt with in dedicated tests so it's not tested in tests that are calling this function. """ > assert len(actual_visit_stats) == len(expected_visit_stats) E assert 0 == 1 E +0 E -1 .tox/py3/lib/python3.7/site-packages/swh/scheduler/tests/test_journal_client.py:144: AssertionError