Changeset View
Changeset View
Standalone View
Standalone View
swh/journal/tests/test_cli.py
Show All 35 Lines | with patch('swh.journal.cli.get_storage') as get_storage_mock: | ||||
yield storage | yield storage | ||||
def invoke(catch_exceptions, args): | def invoke(catch_exceptions, args): | ||||
runner = CliRunner() | runner = CliRunner() | ||||
with tempfile.NamedTemporaryFile('a', suffix='.yml') as config_fd: | with tempfile.NamedTemporaryFile('a', suffix='.yml') as config_fd: | ||||
config_fd.write(CLI_CONFIG) | config_fd.write(CLI_CONFIG) | ||||
config_fd.seek(0) | config_fd.seek(0) | ||||
args = ['-C' + config_fd.name, '-l', 'DEBUG'] + args | args = ['-C' + config_fd.name] + args | ||||
result = runner.invoke(cli, args) | result = runner.invoke(cli, args) | ||||
if not catch_exceptions and result.exception: | if not catch_exceptions and result.exception: | ||||
print(result.output) | print(result.output) | ||||
raise result.exception | raise result.exception | ||||
return result | return result | ||||
def test_replay( | def test_replay( | ||||
Show All 17 Lines | snapshot = {'id': b'foo', 'branches': { | ||||
} | } | ||||
}} | }} | ||||
producer.send( | producer.send( | ||||
topic=kafka_prefix+'.snapshot', key=snapshot['id'], value=snapshot) | topic=kafka_prefix+'.snapshot', key=snapshot['id'], value=snapshot) | ||||
result = invoke(False, [ | result = invoke(False, [ | ||||
'replay', | 'replay', | ||||
'--broker', 'localhost:%d' % port, | '--broker', 'localhost:%d' % port, | ||||
'--consumer-id', 'test-cli-consumer', | '--group-id', 'test-cli-consumer', | ||||
'--prefix', kafka_prefix, | '--prefix', kafka_prefix, | ||||
'--max-messages', '1', | '--max-messages', '1', | ||||
]) | ]) | ||||
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 storage.snapshot_get(snapshot['id']) == { | assert storage.snapshot_get(snapshot['id']) == { | ||||
**snapshot, 'next_branch': None} | **snapshot, 'next_branch': None} |