Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_cli.py
| Show All 12 Lines | |||||
| from confluent_kafka import Producer | from confluent_kafka import Producer | ||||
| import pytest | import pytest | ||||
| import yaml | import yaml | ||||
| from swh.journal.serializers import key_to_kafka, value_to_kafka | from swh.journal.serializers import key_to_kafka, value_to_kafka | ||||
| from swh.model.model import Snapshot, SnapshotBranch, TargetType | from swh.model.model import Snapshot, SnapshotBranch, TargetType | ||||
| from swh.storage import get_storage | from swh.storage import get_storage | ||||
| from swh.storage.cli import storage as cli | from swh.storage.cli import storage as cli | ||||
| from swh.storage.replay import object_converter_fn | from swh.storage.replay import OBJECT_CONVERTERS | ||||
| logger = logging.getLogger(__name__) | logger = logging.getLogger(__name__) | ||||
| CLI_CONFIG = { | CLI_CONFIG = { | ||||
| "storage": {"cls": "memory",}, | "storage": {"cls": "memory",}, | ||||
| } | } | ||||
| ▲ Show 20 Lines • Show All 84 Lines • ▼ Show 20 Lines | |||||
| def test_replay_type_list(): | def test_replay_type_list(): | ||||
| result = invoke("replay", "--help",) | result = invoke("replay", "--help",) | ||||
| assert result.exit_code == 0, result.output | assert result.exit_code == 0, result.output | ||||
| types_in_help = re.findall("--type [[]([a-z_|]+)[]]", result.output) | types_in_help = re.findall("--type [[]([a-z_|]+)[]]", result.output) | ||||
| assert len(types_in_help) == 1 | assert len(types_in_help) == 1 | ||||
| types = types_in_help[0].split("|") | types = types_in_help[0].split("|") | ||||
| assert sorted(types) == sorted(list(object_converter_fn.keys())), ( | assert sorted(types) == sorted(list(OBJECT_CONVERTERS.keys())), ( | ||||
| "Make sure the list of accepted types in cli.py " | "Make sure the list of accepted types in cli.py " | ||||
| "matches implementation in replay.py" | "matches implementation in replay.py" | ||||
| ) | ) | ||||