Changeset View
Changeset View
Standalone View
Standalone View
swh/scheduler/tests/test_scheduler.py
| Show First 20 Lines • Show All 624 Lines • ▼ Show 20 Lines | def test_get_or_create_lister(self, swh_scheduler): | ||||
| assert lister.instance_name == lister_args.get("instance_name", "") | assert lister.instance_name == lister_args.get("instance_name", "") | ||||
| lister_get_again = swh_scheduler.get_or_create_lister( | lister_get_again = swh_scheduler.get_or_create_lister( | ||||
| lister.name, lister.instance_name | lister.name, lister.instance_name | ||||
| ) | ) | ||||
| assert lister == lister_get_again | assert lister == lister_get_again | ||||
| def test_update_lister(self, swh_scheduler): | def test_update_lister(self, swh_scheduler, stored_lister): | ||||
| lister = swh_scheduler.get_or_create_lister(**LISTERS[0]) | lister = attr.evolve(stored_lister, current_state={"updated": "now"}) | ||||
| lister.current_state = {"updated": "now"} | |||||
| updated_lister = swh_scheduler.update_lister(lister) | updated_lister = swh_scheduler.update_lister(lister) | ||||
| assert updated_lister.updated > lister.updated | assert updated_lister.updated > lister.updated | ||||
| assert updated_lister == attr.evolve(lister, updated=updated_lister.updated) | assert updated_lister == attr.evolve(lister, updated=updated_lister.updated) | ||||
| def test_update_lister_stale(self, swh_scheduler): | def test_update_lister_stale(self, swh_scheduler, stored_lister): | ||||
| lister = swh_scheduler.get_or_create_lister(**LISTERS[0]) | swh_scheduler.update_lister(stored_lister) | ||||
| swh_scheduler.update_lister(lister) | |||||
| with pytest.raises(StaleData) as exc: | with pytest.raises(StaleData) as exc: | ||||
| swh_scheduler.update_lister(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 _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) | ||||