Changeset View
Changeset View
Standalone View
Standalone View
swh/scheduler/tests/test_scheduler.py
Show First 20 Lines • Show All 639 Lines • ▼ Show 20 Lines | class TestScheduler: | ||||
def test_update_lister_stale(self, swh_scheduler, stored_lister): | def test_update_lister_stale(self, swh_scheduler, stored_lister): | ||||
swh_scheduler.update_lister(stored_lister) | swh_scheduler.update_lister(stored_lister) | ||||
with pytest.raises(StaleData) as exc: | with pytest.raises(StaleData) as exc: | ||||
swh_scheduler.update_lister(stored_lister) | swh_scheduler.update_lister(stored_lister) | ||||
assert "state not updated" in exc.value.args[0] | assert "state not updated" in exc.value.args[0] | ||||
def test_record_listed_origins(self, swh_scheduler, listed_origins): | |||||
ret = swh_scheduler.record_listed_origins(listed_origins) | |||||
assert set(returned.url for returned in ret) == set( | |||||
origin.url for origin in listed_origins | |||||
) | |||||
assert all(origin.first_seen == origin.last_seen for origin in ret) | |||||
def test_record_listed_origins_upsert(self, swh_scheduler, listed_origins): | |||||
# First, insert `cutoff` origins | |||||
cutoff = 100 | |||||
assert cutoff < len(listed_origins) | |||||
ret = swh_scheduler.record_listed_origins(listed_origins[:cutoff]) | |||||
assert len(ret) == cutoff | |||||
# Then, insert all origins, including the `cutoff` first. | |||||
ret = swh_scheduler.record_listed_origins(listed_origins) | |||||
assert len(ret) == len(listed_origins) | |||||
# Two different "first seen" values | |||||
assert len(set(origin.first_seen for origin in ret)) == 2 | |||||
# But a single "last seen" value | |||||
assert len(set(origin.last_seen for origin in ret)) == 1 | |||||
def _create_task_types(self, scheduler): | def _create_task_types(self, scheduler): | ||||
for tt in TASK_TYPES.values(): | for tt in TASK_TYPES.values(): | ||||
scheduler.create_task_type(tt) | scheduler.create_task_type(tt) |