Changeset View
Changeset View
Standalone View
Standalone View
swh/scheduler/tests/test_cli.py
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 |