Changeset View
Changeset View
Standalone View
Standalone View
swh/scheduler/tests/test_cli.py
| Show All 12 Lines | |||||
| from click.testing import CliRunner | from click.testing import CliRunner | ||||
| import pytest | import pytest | ||||
| from swh.core.api.classes import stream_results | from swh.core.api.classes import stream_results | ||||
| from swh.model.model import Origin | from swh.model.model import Origin | ||||
| from swh.scheduler.cli import cli | from swh.scheduler.cli import cli | ||||
| from swh.scheduler.utils import create_task_dict, utcnow | from swh.scheduler.utils import create_task_dict, utcnow | ||||
| from swh.storage import get_storage | |||||
| CLI_CONFIG = """ | CLI_CONFIG = """ | ||||
| scheduler: | scheduler: | ||||
| cls: foo | cls: foo | ||||
| args: {} | args: {} | ||||
| """ | """ | ||||
| ▲ Show 20 Lines • Show All 562 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 | @pytest.fixture | ||||
| def storage(): | def storage(swh_storage): | ||||
| """An instance of in-memory storage that gets injected | """An instance of in-memory storage that gets injected | ||||
| into the CLI functions.""" | into the CLI functions.""" | ||||
| storage = get_storage(cls="memory") | |||||
| with patch("swh.storage.get_storage") as get_storage_mock: | with patch("swh.storage.get_storage") as get_storage_mock: | ||||
| get_storage_mock.return_value = storage | get_storage_mock.return_value = swh_storage | ||||
| yield 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) | ||||
| ▲ Show 20 Lines • Show All 244 Lines • Show Last 20 Lines | |||||