Differential D4855 Diff 17186 swh/scheduler/tests/test_scheduler.py

Changeset View

Standalone View

swh/scheduler/tests/test_scheduler.py

Show First 20 Lines • Show All 721 Lines • ▼ Show 20 Lines	class TestScheduler:
def test_get_listed_origins_all(self, swh_scheduler, listed_origins) -> None:		def test_get_listed_origins_all(self, swh_scheduler, listed_origins) -> None:
swh_scheduler.record_listed_origins(listed_origins)		swh_scheduler.record_listed_origins(listed_origins)

ret = swh_scheduler.get_listed_origins(limit=len(listed_origins) + 1)		ret = swh_scheduler.get_listed_origins(limit=len(listed_origins) + 1)
assert ret.next_page_token is None		assert ret.next_page_token is None
assert len(ret.origins) == len(listed_origins)		assert len(ret.origins) == len(listed_origins)

@pytest.mark.parametrize("policy", ["oldest_scheduled_first"])		@pytest.mark.parametrize("policy", ["oldest_scheduled_first"])
def test_grab_next_visits(self, swh_scheduler, listed_origins, policy):		def test_grab_next_visits(self, swh_scheduler, listed_origins_by_type, policy):
NUM_RESULTS = 5		NUM_RESULTS = 5
# Strict inequality to check that grab_next_visits doesn't return more		# Strict inequality to check that grab_next_visits doesn't return more
# results than requested		# results than requested
assert len(listed_origins) > NUM_RESULTS		visit_type = next(iter(listed_origins_by_type))
		assert len(listed_origins_by_type[visit_type]) > NUM_RESULTS

swh_scheduler.record_listed_origins(listed_origins)		for origins in listed_origins_by_type.values():
		swh_scheduler.record_listed_origins(origins)

before = utcnow()		before = utcnow()
ret = swh_scheduler.grab_next_visits(NUM_RESULTS, policy=policy)		ret = swh_scheduler.grab_next_visits(visit_type, NUM_RESULTS, policy=policy)
after = utcnow()		after = utcnow()

assert len(ret) == NUM_RESULTS		assert len(ret) == NUM_RESULTS
for origin in ret:		for origin in ret:
assert before <= origin.last_scheduled <= after		assert before <= origin.last_scheduled <= after

@pytest.mark.parametrize("policy", ["oldest_scheduled_first"])		@pytest.mark.parametrize("policy", ["oldest_scheduled_first"])
def test_grab_next_visits_underflow(self, swh_scheduler, listed_origins, policy):		def test_grab_next_visits_underflow(
		self, swh_scheduler, listed_origins_by_type, policy
		):
NUM_RESULTS = 5		NUM_RESULTS = 5
assert len(listed_origins) >= NUM_RESULTS		visit_type = next(iter(listed_origins_by_type))
		assert len(listed_origins_by_type[visit_type]) > NUM_RESULTS

swh_scheduler.record_listed_origins(listed_origins[:NUM_RESULTS])		swh_scheduler.record_listed_origins(
		listed_origins_by_type[visit_type][:NUM_RESULTS]
		)

ret = swh_scheduler.grab_next_visits(NUM_RESULTS + 2, policy=policy)		ret = swh_scheduler.grab_next_visits(visit_type, NUM_RESULTS + 2, policy=policy)

assert len(ret) == NUM_RESULTS		assert len(ret) == NUM_RESULTS

def test_grab_next_visits_unknown_policy(self, swh_scheduler):		def test_grab_next_visits_unknown_policy(self, swh_scheduler):
NUM_RESULTS = 5		NUM_RESULTS = 5
with pytest.raises(UnknownPolicy, match="non_existing_policy"):		with pytest.raises(UnknownPolicy, match="non_existing_policy"):
swh_scheduler.grab_next_visits(NUM_RESULTS, policy="non_existing_policy")		swh_scheduler.grab_next_visits(
		"type", NUM_RESULTS, policy="non_existing_policy"
		)

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)

def test_origin_visit_stats_upsert(self, swh_scheduler) -> None:		def test_origin_visit_stats_upsert(self, swh_scheduler) -> None:
eventful_date = utcnow()		eventful_date = utcnow()
url = "https://github.com/test"		url = "https://github.com/test"
▲ Show 20 Lines • Show All 57 Lines • Show Last 20 Lines