Differential D6569 Diff 24088 swh/storage/tests/test_cli.py

Changeset 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

logger = logging.getLogger(__name__)		logger = logging.getLogger(__name__)


CLI_CONFIG = {		CLI_CONFIG = {
"storage": {"cls": "memory",},		"storage": {"cls": "memory",},
}		}

▲ Show 20 Lines • Show All 75 Lines • ▼ Show 20 Lines	):
expected = r"Done.\n"		expected = r"Done.\n"
assert result.exit_code == 0, result.output		assert result.exit_code == 0, result.output
assert re.fullmatch(expected, result.output, re.MULTILINE), result.output		assert re.fullmatch(expected, result.output, re.MULTILINE), result.output

assert swh_storage.snapshot_get(snapshot.id) == {		assert swh_storage.snapshot_get(snapshot.id) == {
**snapshot_dict,		**snapshot_dict,
"next_branch": None,		"next_branch": None,
}		}


		def test_replay_type_list():
		result = invoke("replay", "--help",)
		assert result.exit_code == 0, result.output
		types_in_help = re.findall("--type [[]([a-z_\|]+)[]]", result.output)
		assert len(types_in_help) == 1
		types = types_in_help[0].split("\|")

		assert sorted(types) == sorted(list(object_converter_fn.keys())), (
		"Make sure the list of accepted types in cli.py "
		"matches implementation in replay.py"
		)