Changeset View

Standalone View

	Show First 20 Lines • Show All 589 Lines • ▼ Show 20 Lines


	def _fill_storage_with_origins(storage, nb_origins):			def _fill_storage_with_origins(storage, nb_origins):
	origins = [Origin(url=f"http://example.com/{i}") for i in range(nb_origins)]			origins = [Origin(url=f"http://example.com/{i}") for i in range(nb_origins)]
	storage.origin_add(origins)			storage.origin_add(origins)
	return origins			return origins


	@pytest.fixture
	def storage(swh_storage):
	"""An instance of in-memory storage that gets injected
	into the CLI functions."""
	with patch("swh.storage.get_storage") as get_storage_mock:
	get_storage_mock.return_value = swh_storage
	yield swh_storage


	@patch("swh.scheduler.cli.utils.TASK_BATCH_SIZE", 3)			@patch("swh.scheduler.cli.utils.TASK_BATCH_SIZE", 3)
	def test_task_schedule_origins_dry_run(swh_scheduler, storage):			def test_task_schedule_origins_dry_run(swh_scheduler, storage):
	"""Tests the scheduling when origin_batch_size*task_batch_size is a			"""Tests the scheduling when origin_batch_size*task_batch_size is a
	divisor of nb_origins."""			divisor of nb_origins."""
	_fill_storage_with_origins(storage, 90)			_fill_storage_with_origins(storage, 90)

	result = invoke(			result = invoke(
	swh_scheduler,			swh_scheduler,
	▲ Show 20 Lines • Show All 242 Lines • Show Last 20 Lines