Nucypher
/
nucypher
mirror of https://github.com/nucypher/nucypher.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #2033 from fjarri/mock_stdin 

IO test fixtures change: mock stdin directly instead of `click` functions, use `capsys` for stdout
pull/2064/head
K Prasch 2020-05-28 17:41:51 -07:00 committed by GitHub
parent 0eb3fba5eb 8dbaf2b206
commit 26b07fc4e9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 394 additions and 241 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
tests/acceptance/cli/ursula/test_federated_ursula.py
View File

@ -28,7 +28,7 @@ from nucypher.config.constants import APP_DIR, DEFAULT_CONFIG_ROOT, NUCYPHER_ENV
from tests.constants import (
FAKE_PASSWORD_CONFIRMED, INSECURE_DEVELOPMENT_PASSWORD,
MOCK_CUSTOM_INSTALLATION_PATH,
MOCK_IP_ADDRESS, YES)
MOCK_IP_ADDRESS, YES_ENTER)
from tests.utils.ursula import MOCK_URSULA_STARTING_PORT
@ -41,7 +41,7 @@ def test_initialize_ursula_defaults(click_runner, mocker):
# Use default ursula init args
init_args = ('ursula', 'init', '--network', TEMPORARY_DOMAIN, '--federated-only')
user_input = YES + FAKE_PASSWORD_CONFIRMED
user_input = YES_ENTER + FAKE_PASSWORD_CONFIRMED
result = click_runner.invoke(nucypher_cli, init_args, input=user_input, catch_exceptions=False)
assert result.exit_code == 0

4
tests/acceptance/cli/ursula/test_run_ursula.py
View File

@ -37,7 +37,7 @@ from tests.constants import (
INSECURE_DEVELOPMENT_PASSWORD,
MOCK_IP_ADDRESS,
TEST_PROVIDER_URI,
YES
YES_ENTER
)
from tests.utils.ursula import MOCK_URSULA_STARTING_PORT, start_pytest_ursula_services
@ -57,7 +57,7 @@ def test_ursula_rest_host_determination(click_runner, mocker):
mocker.patch.object(UrsulaConfiguration, 'to_configuration_file', return_value=None)
args = ('ursula', 'init', '--federated-only', '--network', TEMPORARY_DOMAIN)
user_input = YES + FAKE_PASSWORD_CONFIRMED
user_input = YES_ENTER + FAKE_PASSWORD_CONFIRMED
result = click_runner.invoke(nucypher_cli, args, catch_exceptions=False, input=user_input)
assert result.exit_code == 0
assert MOCK_IP_ADDRESS in result.output

6
tests/constants.py
View File

@ -139,9 +139,11 @@ PYEVM_GAS_LIMIT = TEST_GAS_LIMIT # TODO: move elsewhere (used to set pyevm gas
# CLI
#
YES = 'Y\n'
YES = 'Y'
YES_ENTER = YES + '\n'
NO = 'N\n'
NO = 'N'
NO_ENTER = NO + '\n'
FAKE_PASSWORD_CONFIRMED = '{password}\n{password}\n'.format(password=INSECURE_DEVELOPMENT_PASSWORD)

15
tests/fixtures.py
View File

@ -972,18 +972,11 @@ def highperf_mocked_bob(fleet_of_highperf_mocked_ursulas):
#
@pytest.fixture(scope='function', autouse=True)
def stdout_trap(mocker):
trap = StringIO()
mocker.patch('sys.stdout', new=trap)
trap.truncate(0)
yield trap
trap.truncate(0)
@pytest.fixture(scope='function')
def test_emitter(mocker, stdout_trap):
mocker.patch('sys.stdout', new=stdout_trap)
def test_emitter(mocker):
# Note that this fixture does not capture console output.
# Whether the output is captured or not is controlled by
# the usage of the (built-in) `capsys` fixture or global PyTest run settings.
return StdoutEmitter()

88
tests/integration/cli/actions/test_auth_actions.py
View File

@ -39,79 +39,68 @@ from tests.constants import INSECURE_DEVELOPMENT_PASSWORD
@pytest.mark.parametrize('confirm', (True, False))
def test_get_password_from_prompt_cli_action(mock_click_prompt, mock_click_confirm, confirm):
def test_get_password_from_prompt_cli_action(mocker, mock_stdin, confirm, capsys):
# Setup
mock_click_prompt.return_value = INSECURE_DEVELOPMENT_PASSWORD
mock_click_confirm.return_value = True
# Test
result = get_password_from_prompt(confirm=confirm)
assert result
assert result is not NO_PASSWORD
assert result == INSECURE_DEVELOPMENT_PASSWORD
mock_click_prompt.assert_called_once_with(GENERIC_PASSWORD_PROMPT,
confirmation_prompt=confirm,
hide_input=True)
@pytest.mark.parametrize('confirm', (True, False))
def test_get_password_from_prompt_cli_action(mocker, mock_click_prompt, mock_click_confirm, confirm):
# Setup
mock_click_prompt.return_value = INSECURE_DEVELOPMENT_PASSWORD
mock_click_confirm.return_value = True
test_envavr = 'NUCYPHER_TEST_ENVVAR'
mock_stdin.password(INSECURE_DEVELOPMENT_PASSWORD, confirm=confirm)
test_envvar = 'NUCYPHER_TEST_ENVVAR'
another_password = 'th1s-iS-n0t-secur3'
mocker.patch.dict(os.environ, {test_envavr: another_password})
mocker.patch.dict(os.environ, {test_envvar: another_password})
result = get_password_from_prompt(confirm=confirm)
assert result == INSECURE_DEVELOPMENT_PASSWORD
mock_click_prompt.assert_called_once_with(GENERIC_PASSWORD_PROMPT,
confirmation_prompt=confirm,
hide_input=True)
mock_click_prompt.reset()
assert mock_stdin.empty()
captured = capsys.readouterr()
assert GENERIC_PASSWORD_PROMPT in captured.out
if confirm:
assert "Repeat for confirmation:" in captured.out
# From env var
mocker.patch.dict(os.environ, {test_envavr: another_password})
result = get_password_from_prompt(confirm=confirm, envvar=test_envavr)
mocker.patch.dict(os.environ, {test_envvar: another_password})
result = get_password_from_prompt(confirm=confirm, envvar=test_envvar)
assert result is not NO_PASSWORD
assert result != INSECURE_DEVELOPMENT_PASSWORD
assert result == another_password
mock_click_prompt.assert_called_once_with(GENERIC_PASSWORD_PROMPT,
confirmation_prompt=confirm,
hide_input=True)
assert mock_stdin.empty()
captured = capsys.readouterr()
assert not captured.out
assert not captured.err
def test_get_client_password_with_invalid_address(mock_click_prompt, mock_account):
def test_get_client_password_with_invalid_address(mock_stdin):
# `mock_stdin` used to assert the user was not prompted
bad_address = '0xdeadbeef'
with pytest.raises(InvalidChecksumAddress):
get_client_password(checksum_address=bad_address)
@pytest.mark.parametrize('confirm', (True, False))
def test_get_client_password(mock_click_prompt, mock_account, confirm):
mock_click_prompt.return_value = INSECURE_DEVELOPMENT_PASSWORD
def test_get_client_password(mock_stdin, mock_account, confirm, capsys):
mock_stdin.password(INSECURE_DEVELOPMENT_PASSWORD, confirm=confirm)
result = get_client_password(checksum_address=mock_account.address, confirm=confirm)
assert result == INSECURE_DEVELOPMENT_PASSWORD
assert mock_stdin.empty()
message = COLLECT_ETH_PASSWORD.format(checksum_address=mock_account.address)
mock_click_prompt.assert_called_once_with(message, confirmation_prompt=confirm, hide_input=True)
captured = capsys.readouterr()
assert message in captured.out
if confirm:
assert "Repeat for confirmation:" in captured.out
@pytest.mark.parametrize('confirm', (True, False))
def test_get_nucypher_password(mock_click_prompt, mock_account, confirm):
mock_click_prompt.return_value = INSECURE_DEVELOPMENT_PASSWORD
def test_get_nucypher_password(mock_stdin, mock_account, confirm, capsys):
mock_stdin.password(INSECURE_DEVELOPMENT_PASSWORD, confirm=confirm)
result = get_nucypher_password(confirm=confirm)
assert result == INSECURE_DEVELOPMENT_PASSWORD
prompt = COLLECT_NUCYPHER_PASSWORD
assert mock_stdin.empty()
captured = capsys.readouterr()
assert COLLECT_NUCYPHER_PASSWORD in captured.out
if confirm:
prompt += f" ({NucypherKeyring.MINIMUM_PASSWORD_LENGTH} character minimum)"
mock_click_prompt.assert_called_once_with(prompt, confirmation_prompt=confirm, hide_input=True)
prompt = COLLECT_NUCYPHER_PASSWORD + f" ({NucypherKeyring.MINIMUM_PASSWORD_LENGTH} character minimum)"
assert prompt in captured.out
def test_unlock_nucypher_keyring_invalid_password(mocker, test_emitter, stdout_trap, alice_blockchain_test_config):
def test_unlock_nucypher_keyring_invalid_password(mocker, test_emitter, alice_blockchain_test_config, capsys):
# Setup
keyring_attach_spy = mocker.spy(CharacterConfiguration, 'attach_keyring')
@ -128,8 +117,11 @@ def test_unlock_nucypher_keyring_invalid_password(mocker, test_emitter, stdout_t
character_configuration=alice_blockchain_test_config)
keyring_attach_spy.assert_called_once()
captured = capsys.readouterr()
assert DECRYPTING_CHARACTER_KEYRING.format(name='alice') in captured.out
def test_unlock_nucypher_keyring_dev_mode(mocker, test_emitter, stdout_trap, alice_blockchain_test_config):
def test_unlock_nucypher_keyring_dev_mode(mocker, test_emitter, capsys, alice_blockchain_test_config):
# Setup
unlock_spy = mocker.spy(NucypherKeyring, 'unlock')
@ -144,7 +136,7 @@ def test_unlock_nucypher_keyring_dev_mode(mocker, test_emitter, stdout_trap, ali
character_configuration=alice_blockchain_test_config)
assert result
output = stdout_trap.getvalue()
output = capsys.readouterr().out
message = DECRYPTING_CHARACTER_KEYRING.format(name=alice_blockchain_test_config.NAME)
assert message in output
@ -154,7 +146,7 @@ def test_unlock_nucypher_keyring_dev_mode(mocker, test_emitter, stdout_trap, ali
def test_unlock_nucypher_keyring(mocker,
test_emitter,
stdout_trap,
capsys,
alice_blockchain_test_config,
patch_keystore,
tmpdir):
@ -173,9 +165,9 @@ def test_unlock_nucypher_keyring(mocker,
character_configuration=alice_blockchain_test_config)
assert result
output = stdout_trap.getvalue()
captured = capsys.readouterr()
message = DECRYPTING_CHARACTER_KEYRING.format(name=alice_blockchain_test_config.NAME)
assert message in output
assert message in captured.out
unlock_spy.assert_called_once_with(password=INSECURE_DEVELOPMENT_PASSWORD)
attach_spy.assert_called_once()

47
tests/integration/cli/actions/test_config_actions.py
View File

@ -31,7 +31,10 @@ from nucypher.cli.literature import (
INVALID_CONFIGURATION_FILE_WARNING,
INVALID_JSON_IN_CONFIGURATION_WARNING,
MISSING_CONFIGURATION_FILE,
SUCCESSFUL_DESTRUCTION
SUCCESSFUL_DESTRUCTION,
SUCCESSFUL_UPDATE_CONFIGURATION_VALUES,
CONFIRM_FORGET_NODES,
SUCCESSFUL_FORGET_NODES,
)
from nucypher.config.node import CharacterConfiguration
from tests.constants import YES
@ -79,25 +82,30 @@ def config(request, mocker):
mocker.resetall() # dont carry over context between functions
def test_forget_cli_action(alice_blockchain_test_config, test_emitter, stdout_trap, mock_click_confirm, mocker):
def test_forget_cli_action(alice_blockchain_test_config, test_emitter, mock_stdin, mocker, capsys):
mock_forget = mocker.patch.object(CharacterConfiguration, 'forget_nodes')
mock_click_confirm.return_value = YES
mock_stdin.line(YES)
forget(emitter=test_emitter, configuration=alice_blockchain_test_config)
mock_forget.assert_called_once()
assert mock_stdin.empty()
captured = capsys.readouterr()
assert CONFIRM_FORGET_NODES in captured.out
assert SUCCESSFUL_FORGET_NODES in captured.out
def test_update_configuration_cli_action(config, test_emitter, stdout_trap, test_registry_source_manager):
def test_update_configuration_cli_action(config, test_emitter, test_registry_source_manager, capsys):
config_class, config_file = config.__class__, config.filepath
updates = dict(federated_only=True)
get_or_update_configuration(emitter=test_emitter, config_class=config_class, filepath=config_file, updates=updates)
config.update.assert_called_once_with(**updates)
configure.handle_invalid_configuration_file.assert_not_called()
configure.handle_missing_configuration_file.assert_not_called()
captured = capsys.readouterr()
assert SUCCESSFUL_UPDATE_CONFIGURATION_VALUES.format(fields='federated_only') in captured.out
def test_handle_update_missing_configuration_file_cli_action(config,
test_emitter,
stdout_trap,
test_registry_source_manager,
mocker):
config_class, config_file = config.__class__, config.filepath
@ -115,9 +123,9 @@ def test_handle_update_missing_configuration_file_cli_action(config,
def test_handle_update_invalid_configuration_file_cli_action(config,
test_emitter,
stdout_trap,
test_registry_source_manager,
mocker):
mocker,
capsys):
config_class = config.__class__
config_file = config.filepath
mocker.patch.object(config_class, '_read_configuration_file', side_effect=config_class.ConfigurationError)
@ -130,16 +138,19 @@ def test_handle_update_invalid_configuration_file_cli_action(config,
configure.handle_missing_configuration_file.assert_not_called()
config._write_configuration_file.assert_not_called()
configure.handle_invalid_configuration_file.assert_called()
captured = capsys.readouterr()
assert INVALID_CONFIGURATION_FILE_WARNING.format(filepath=config_file) in captured.out
def test_destroy_configuration_cli_action(config, test_emitter, stdout_trap, mocker, mock_click_confirm):
def test_destroy_configuration_cli_action(config, test_emitter, capsys, mocker, mock_stdin):
config_class = config.__class__
mock_config_destroy = mocker.patch.object(config_class, 'destroy')
mock_click_confirm.return_value = YES
mock_stdin.line(YES)
destroy_configuration(emitter=test_emitter, character_config=config)
mock_config_destroy.assert_called_once()
output = stdout_trap.getvalue()
assert SUCCESSFUL_DESTRUCTION in output
captured = capsys.readouterr()
assert SUCCESSFUL_DESTRUCTION in captured.out
assert mock_stdin.empty()
def test_handle_missing_configuration_file_cli_action(config):
@ -155,7 +166,7 @@ def test_handle_missing_configuration_file_cli_action(config):
@pytest.mark.parametrize('bad_config_payload', BAD_CONFIG_FILE_CONTENTS)
def test_handle_invalid_configuration_file_cli_action(mocker, config, test_emitter, stdout_trap, bad_config_payload):
def test_handle_invalid_configuration_file_cli_action(mocker, config, test_emitter, capsys, bad_config_payload):
config_class = config.__class__
config_file = Path(config.filepath)
mocker.patch.object(config_class, '_read_configuration_file', return_value=bad_config_payload)
@ -163,13 +174,13 @@ def test_handle_invalid_configuration_file_cli_action(mocker, config, test_emitt
handle_invalid_configuration_file(emitter=test_emitter,
config_class=config_class,
filepath=config_file)
output = stdout_trap.getvalue()
captured = capsys.readouterr()
message_1 = INVALID_CONFIGURATION_FILE_WARNING.format(filepath=config_file)
assert message_1 in output
assert message_1 in captured.out
@pytest.mark.parametrize('side_effect', (TypeError,))
def test_handle_corrupted_configuration_file_cli_action(mocker, config, test_emitter, stdout_trap, side_effect):
def test_handle_corrupted_configuration_file_cli_action(mocker, config, test_emitter, capsys, side_effect):
config_class = config.__class__
config_file = Path(config.filepath)
mocker.patch('__main__.open', return_value=b'AAAAAAAAAAAAA')
@ -178,8 +189,8 @@ def test_handle_corrupted_configuration_file_cli_action(mocker, config, test_emi
handle_invalid_configuration_file(emitter=test_emitter,
config_class=config_class,
filepath=config_file)
output = stdout_trap.getvalue()
captured = capsys.readouterr()
message_1 = INVALID_CONFIGURATION_FILE_WARNING.format(filepath=config_file)
message_2 = INVALID_JSON_IN_CONFIGURATION_WARNING.format(filepath=config_file)
assert message_1 in output
assert message_2 in output
assert message_1 in captured.out
assert message_2 in captured.out

179
tests/integration/cli/actions/test_confirm_actions.py
View File

@ -23,26 +23,26 @@ from nucypher.blockchain.eth.token import NU
from nucypher.cli.actions.confirm import (confirm_deployment, confirm_enable_restaking, confirm_enable_restaking_lock,
confirm_enable_winding_down, confirm_large_stake, confirm_staged_stake)
from nucypher.cli.literature import (ABORT_DEPLOYMENT, RESTAKING_AGREEMENT, RESTAKING_LOCK_AGREEMENT,
WINDING_DOWN_AGREEMENT)
WINDING_DOWN_AGREEMENT, CONFIRM_STAGED_STAKE,
CONFIRM_LARGE_STAKE_VALUE, CONFIRM_LARGE_STAKE_DURATION)
from tests.constants import YES, NO
def test_confirm_deployment_cli_action(mocker, mock_click_prompt, test_emitter, stdout_trap, mock_testerchain):
mock_click_prompt.return_value = False
def test_confirm_deployment_cli_action(mocker, mock_stdin, test_emitter, capsys, mock_testerchain):
mock_stdin.line('foo') # anything different from `deployer_interface.client.chain_name.upper()`
with pytest.raises(click.Abort):
confirm_deployment(emitter=test_emitter, deployer_interface=mock_testerchain)
output = stdout_trap.getvalue()
assert ABORT_DEPLOYMENT in output
captured = capsys.readouterr()
assert ABORT_DEPLOYMENT in captured.out
assert mock_stdin.empty()
stdout_trap.truncate(0) # clear
mock_click_prompt.return_value = 'DEPLOY' # say the magic word
mock_stdin.line('DEPLOY') # say the magic word
result = confirm_deployment(emitter=test_emitter, deployer_interface=mock_testerchain)
assert result
output = stdout_trap.getvalue()
assert not output
stdout_trap.truncate(0) # clear
captured = capsys.readouterr()
assert "Type 'DEPLOY' to continue: " in captured.out
assert mock_stdin.empty()
# Mimick a known chain name
llamanet, llamanet_chain_id = 'llamanet', 1123589012901209
@ -59,141 +59,184 @@ def test_confirm_deployment_cli_action(mocker, mock_click_prompt, test_emitter,
new_callable=mocker.PropertyMock)
mock_testerchain.client.is_local = False
mock_click_prompt.return_value = 'DEPLOY' # say the (wrong) magic word
mock_stdin.line('DEPLOY') # say the (wrong) magic word
with pytest.raises(click.Abort):
confirm_deployment(emitter=test_emitter, deployer_interface=mock_testerchain)
assert mock_stdin.empty()
captured = capsys.readouterr()
assert f"Type '{llamanet.upper()}' to continue: " in captured.out
assert ABORT_DEPLOYMENT in captured.out
mock_click_prompt.return_value = llamanet # say the (almost correct) magic word
mock_stdin.line(llamanet) # say the (almost correct) magic word
with pytest.raises(click.Abort):
confirm_deployment(emitter=test_emitter, deployer_interface=mock_testerchain)
assert mock_stdin.empty()
captured = capsys.readouterr()
assert f"Type '{llamanet.upper()}' to continue: " in captured.out
assert ABORT_DEPLOYMENT in captured.out
mock_click_prompt.return_value = llamanet.upper() # say the (correct, uppercase) network name
mock_stdin.line(llamanet.upper()) # say the (correct, uppercase) network name
result = confirm_deployment(emitter=test_emitter, deployer_interface=mock_testerchain)
assert result
assert mock_stdin.empty()
captured = capsys.readouterr()
assert f"Type '{llamanet.upper()}' to continue: " in captured.out
def test_confirm_enable_restaking_lock_cli_action(mock_click_confirm, test_emitter, stdout_trap):
def test_confirm_enable_restaking_lock_cli_action(mock_stdin, test_emitter, capsys):
# Test data
staking_address, release_period = '0xdeadbeef', 1
# Positive Case
mock_click_confirm.return_value = True
mock_stdin.line(YES)
result = confirm_enable_restaking_lock(emitter=test_emitter,
release_period=release_period,
staking_address=staking_address)
assert result
output = stdout_trap.getvalue()
captured = capsys.readouterr()
assert mock_stdin.empty()
restake_agreement = RESTAKING_LOCK_AGREEMENT.format(staking_address=staking_address, release_period=release_period)
assert restake_agreement in output
stdout_trap.truncate(0) # clear
assert restake_agreement in captured.out
# Negative case
mock_click_confirm.side_effect = click.Abort
mock_stdin.line(NO)
with pytest.raises(click.Abort):
confirm_enable_restaking_lock(emitter=test_emitter,
release_period=release_period,
staking_address=staking_address)
output = stdout_trap.getvalue()
captured = capsys.readouterr()
assert mock_stdin.empty()
restake_agreement = RESTAKING_LOCK_AGREEMENT.format(staking_address=staking_address,
release_period=release_period)
assert restake_agreement in output
assert restake_agreement in captured.out
def test_confirm_enable_restaking_cli_action(test_emitter, mock_click_confirm, stdout_trap):
def test_confirm_enable_restaking_cli_action(test_emitter, mock_stdin, capsys):
# Positive Case
mock_click_confirm.return_value = True
mock_stdin.line(YES)
staking_address = '0xdeadbeef'
result = confirm_enable_restaking(emitter=test_emitter, staking_address=staking_address)
assert result
assert mock_stdin.empty()
output = stdout_trap.getvalue()
captured = capsys.readouterr()
restake_agreement = RESTAKING_AGREEMENT.format(staking_address=staking_address)
assert restake_agreement in output
assert restake_agreement in captured.out
# Negative case
stdout_trap.truncate(0) # clear
mock_click_confirm.side_effect = click.Abort
mock_stdin.line(NO)
with pytest.raises(click.Abort):
confirm_enable_restaking(emitter=test_emitter, staking_address=staking_address)
output = stdout_trap.getvalue()
captured = capsys.readouterr()
assert mock_stdin.empty()
restake_agreement = RESTAKING_AGREEMENT.format(staking_address=staking_address)
assert restake_agreement in output
assert restake_agreement in captured.out
def test_confirm_enable_winding_down_cli_action(test_emitter, mock_click_confirm, stdout_trap):
def test_confirm_enable_winding_down_cli_action(test_emitter, mock_stdin, capsys):
# Positive Case
mock_click_confirm.return_value = True
mock_stdin.line(YES)
staking_address = '0xdeadbeef'
result = confirm_enable_winding_down(emitter=test_emitter, staking_address=staking_address)
assert result
assert mock_stdin.empty()
output = stdout_trap.getvalue()
assert WINDING_DOWN_AGREEMENT in output
captured = capsys.readouterr()
assert WINDING_DOWN_AGREEMENT in captured.out
# Negative case
stdout_trap.truncate(0) # clear
mock_click_confirm.side_effect = click.Abort
mock_stdin.line(NO)
with pytest.raises(click.Abort):
confirm_enable_winding_down(emitter=test_emitter, staking_address=staking_address)
output = stdout_trap.getvalue()
assert WINDING_DOWN_AGREEMENT in output
captured = capsys.readouterr()
assert mock_stdin.empty()
assert WINDING_DOWN_AGREEMENT in captured.out
def test_confirm_staged_stake_cli_action(test_emitter, mock_click_confirm, stdout_trap):
def test_confirm_staged_stake_cli_action(test_emitter, mock_stdin, capsys):
staking_address, value, lock_periods = '0xdeadbeef', NU.from_tokens(1), 1
confirmation = CONFIRM_STAGED_STAKE.format(staker_address=staking_address,
lock_periods=lock_periods,
tokens=value,
nunits=value.to_nunits())
# Positive Case
mock_click_confirm.return_value = True
staking_address, value, lock_periods = '0xdeadbeef', NU.from_tokens(1), 1
mock_stdin.line(YES)
result = confirm_staged_stake(staker_address=staking_address,
value=value,
lock_periods=lock_periods)
assert result
assert mock_stdin.empty()
output = stdout_trap.getvalue()
assert not output
captured = capsys.readouterr()
assert confirmation in captured.out
# Negative case
stdout_trap.truncate(0) # clear
mock_click_confirm.side_effect = click.Abort
mock_stdin.line(NO)
with pytest.raises(click.Abort):
confirm_staged_stake(staker_address=staking_address,
value=value,
lock_periods=lock_periods)
output = stdout_trap.getvalue()
assert not output
captured = capsys.readouterr()
assert confirmation in captured.out
assert mock_stdin.empty()
@pytest.mark.parametrize('value,duration,prompt_indicated', (
(NU.from_tokens(1), 1, False),
(NU.from_tokens(1), 31, False),
(NU.from_tokens(15), 31, False),
(NU.from_tokens(150001), 31, True),
(NU.from_tokens(150000), 366, True),
(NU.from_tokens(150001), 366, True),
@pytest.mark.parametrize('value,duration,must_confirm_value,must_confirm_duration', (
(NU.from_tokens(1), 1, False, False),
(NU.from_tokens(1), 31, False, False),
(NU.from_tokens(15), 31, False, False),
(NU.from_tokens(150001), 31, True, False),
(NU.from_tokens(150000), 366, False, True),
(NU.from_tokens(150001), 366, True, True),
))
def test_confirm_large_stake_cli_action(test_emitter, mock_click_confirm, stdout_trap, value, duration, prompt_indicated):
def test_confirm_large_stake_cli_action(test_emitter,
mock_stdin,
capsys,
value,
duration,
must_confirm_value,
must_confirm_duration):
# Positive Cases
mock_click_confirm.return_value = True
asked_about_value = lambda output: CONFIRM_LARGE_STAKE_VALUE.format(value=value) in output
asked_about_duration = lambda output: CONFIRM_LARGE_STAKE_DURATION.format(lock_periods=duration) in output
# Positive Cases - either do not need to confirm anything, or say yes
if must_confirm_value:
mock_stdin.line(YES)
if must_confirm_duration:
mock_stdin.line(YES)
result = confirm_large_stake(value=value, lock_periods=duration)
assert result
output = stdout_trap.getvalue()
assert not output
stdout_trap.truncate(0) # clear
captured = capsys.readouterr()
assert must_confirm_value == asked_about_value(captured.out)
assert must_confirm_duration == asked_about_duration(captured.out)
assert mock_stdin.empty()
if must_confirm_value or must_confirm_duration:
# Negative cases - must confirm something and say no
if must_confirm_value and must_confirm_duration:
# yes to the former but not to the latter
mock_stdin.line(YES)
mock_stdin.line(NO)
else:
# no to whatever one we are asked about
mock_stdin.line(NO)
if prompt_indicated:
# Negative cases
mock_click_confirm.side_effect = click.Abort
with pytest.raises(click.Abort):
confirm_large_stake(value=value, lock_periods=duration)
output = stdout_trap.getvalue()
assert not output
captured = capsys.readouterr()
assert must_confirm_value == asked_about_value(captured.out)
assert must_confirm_duration == asked_about_duration(captured.out)
assert mock_stdin.empty()

69
tests/integration/cli/actions/test_select_client_account.py
View File

@ -27,36 +27,44 @@ from nucypher.blockchain.eth.interfaces import BlockchainInterfaceFactory
from nucypher.blockchain.eth.signers import KeystoreSigner
from nucypher.blockchain.eth.token import NU
from nucypher.cli.actions.select import select_client_account
from nucypher.cli.literature import NO_ETH_ACCOUNTS
from nucypher.cli.literature import (
NO_ETH_ACCOUNTS,
GENERIC_SELECT_ACCOUNT,
)
from nucypher.config.constants import TEMPORARY_DOMAIN
from tests.constants import MOCK_PROVIDER_URI, MOCK_SIGNER_URI, NUMBER_OF_ETH_TEST_ACCOUNTS
@pytest.mark.parametrize('selection', range(NUMBER_OF_ETH_TEST_ACCOUNTS))
def test_select_client_account(mock_click_prompt, test_emitter, mock_testerchain, selection):
def test_select_client_account(mock_stdin, test_emitter, mock_testerchain, selection, capsys):
"""Fine-grained assertions about the return value of interactive client account selection"""
mock_click_prompt.return_value = selection
mock_stdin.line(str(selection))
expected_account = mock_testerchain.client.accounts[selection]
selected_account = select_client_account(emitter=test_emitter, provider_uri=MOCK_PROVIDER_URI)
assert selected_account, "Account selection returned Falsy instead of an address"
assert isinstance(selected_account, str), "Selection is not a str"
assert is_checksum_address(selected_account), "Selection is not a valid checksum address"
assert selected_account == expected_account, "Selection returned the wrong address"
assert mock_stdin.empty()
captured = capsys.readouterr()
assert GENERIC_SELECT_ACCOUNT in captured.out
def test_select_client_account_with_no_accounts(mocker,
mock_click_prompt,
mock_stdin, # used to assert the user was not prompted
test_emitter,
mock_testerchain,
stdout_trap):
capsys):
mocker.patch.object(EthereumClient, 'accounts', return_value=[])
with pytest.raises(click.Abort):
select_client_account(emitter=test_emitter, provider_uri=MOCK_PROVIDER_URI)
output = stdout_trap.getvalue()
assert NO_ETH_ACCOUNTS in output
captured = capsys.readouterr()
assert NO_ETH_ACCOUNTS in captured.out
def test_select_client_account_ambiguous_source(mock_click_prompt, test_emitter, mock_testerchain):
def test_select_client_account_ambiguous_source(mock_stdin, # used to assert the user was not prompted
test_emitter,
mock_testerchain):
#
# Implicit wallet
@ -93,40 +101,54 @@ def test_select_client_account_ambiguous_source(mock_click_prompt, test_emitter,
@pytest.mark.parametrize('selection', range(NUMBER_OF_ETH_TEST_ACCOUNTS))
def test_select_client_account_valid_sources(mocker,
mock_click_prompt,
mock_stdin,
test_emitter,
mock_testerchain,
patch_keystore,
mock_accounts,
selection):
# Setup
mock_click_prompt.return_value = selection
selection,
capsys):
# From External Signer
mock_stdin.line(str(selection))
mock_signer = mocker.patch.object(KeystoreSigner, 'from_signer_uri')
selected_account = select_client_account(emitter=test_emitter, signer_uri=MOCK_SIGNER_URI)
expected_account = mock_testerchain.client.accounts[selection]
assert selected_account == expected_account
mock_signer.assert_called_once_with(uri=MOCK_SIGNER_URI)
assert mock_stdin.empty()
captured = capsys.readouterr()
assert GENERIC_SELECT_ACCOUNT in captured.out and f"Selected {selection}" in captured.out
# From Wallet
mock_stdin.line(str(selection))
expected_account = mock_testerchain.client.accounts[selection]
wallet = Wallet(provider_uri=MOCK_PROVIDER_URI)
selected_account = select_client_account(emitter=test_emitter, wallet=wallet)
assert selected_account == expected_account
assert mock_stdin.empty()
captured = capsys.readouterr()
assert GENERIC_SELECT_ACCOUNT in captured.out and f"Selected {selection}" in captured.out
# From pre-initialized Provider
mock_stdin.line(str(selection))
expected_account = mock_testerchain.client.accounts[selection]
selected_account = select_client_account(emitter=test_emitter, provider_uri=MOCK_PROVIDER_URI)
assert selected_account == expected_account
assert mock_stdin.empty()
captured = capsys.readouterr()
assert GENERIC_SELECT_ACCOUNT in captured.out and f"Selected {selection}" in captured.out
# From uninitialized Provider
mock_stdin.line(str(selection))
mocker.patch.object(BlockchainInterfaceFactory, 'is_interface_initialized', return_value=False)
mocker.patch.object(BlockchainInterfaceFactory, '_interfaces', return_value={})
mocker.patch.object(BlockchainInterfaceFactory, 'get_interface', return_value=mock_testerchain)
selected_account = select_client_account(emitter=test_emitter, provider_uri=MOCK_PROVIDER_URI)
assert selected_account == expected_account
assert mock_stdin.empty()
captured = capsys.readouterr()
assert GENERIC_SELECT_ACCOUNT in captured.out and f"Selected {selection}" in captured.out
@pytest.mark.parametrize('selection,show_staking,show_eth,show_tokens,stake_info', (
@ -140,10 +162,10 @@ def test_select_client_account_valid_sources(mocker,
(0, False, False, True, []),
(0, False, False, False, []),
))
def test_select_client_account_with_balance_display(mock_click_prompt,
def test_select_client_account_with_balance_display(mock_stdin,
test_emitter,
mock_testerchain,
stdout_trap,
capsys,
test_registry_source_manager,
mock_staking_agent,
mock_token_agent,
@ -154,7 +176,6 @@ def test_select_client_account_with_balance_display(mock_click_prompt,
stake_info):
# Setup
mock_click_prompt.return_value = selection
mock_staking_agent.get_all_stakes.return_value = stake_info
# Missing network kwarg with balance display active
@ -168,6 +189,7 @@ def test_select_client_account_with_balance_display(mock_click_prompt,
provider_uri=MOCK_PROVIDER_URI)
# Good selection
mock_stdin.line(str(selection))
selected_account = select_client_account(emitter=test_emitter,
network=TEMPORARY_DOMAIN,
show_eth_balance=show_eth,
@ -177,6 +199,7 @@ def test_select_client_account_with_balance_display(mock_click_prompt,
# check for accurate selection consistency with client index
assert selected_account == mock_testerchain.client.accounts[selection]
assert mock_stdin.empty()
# Display account info
headers = ['Account']
@ -187,23 +210,23 @@ def test_select_client_account_with_balance_display(mock_click_prompt,
if show_tokens:
headers.append('NU')
output = stdout_trap.getvalue()
captured = capsys.readouterr()
for column_name in headers:
assert column_name in output, f'"{column_name}" column was not displayed'
assert column_name in captured.out, f'"{column_name}" column was not displayed'
for account in mock_testerchain.client.accounts:
assert account in output
assert account in captured.out
if show_tokens:
balance = mock_token_agent.get_balance(address=account)
assert str(NU.from_nunits(balance)) in output
assert str(NU.from_nunits(balance)) in captured.out
if show_eth:
balance = mock_testerchain.client.get_balance(account=account)
assert str(Web3.fromWei(balance, 'ether')) in output
assert str(Web3.fromWei(balance, 'ether')) in captured.out
if show_staking:
if len(stake_info) == 0:
assert "No" in output
assert "No" in captured.out
else:
assert 'Yes' in output
assert 'Yes' in captured.out

15
tests/integration/cli/actions/test_select_client_account_for_staking.py
View File

@ -25,10 +25,9 @@ from tests.constants import YES
def test_select_client_account_for_staking_cli_action(test_emitter,
test_registry,
test_registry_source_manager,
mock_click_prompt,
mock_click_confirm,
mock_stdin,
mock_testerchain,
stdout_trap,
capsys,
mocker):
"""Fine-grained assertions about the return value of interactive client account selection"""
force = False
@ -45,18 +44,19 @@ def test_select_client_account_for_staking_cli_action(test_emitter,
force=force)
assert client_account == staking_address == selected_account
mock_click_prompt.return_value = selected_index
mock_stdin.line(str(selected_index))
client_account, staking_address = select_client_account_for_staking(emitter=test_emitter,
stakeholder=stakeholder,
staking_address=None,
individual_allocation=None,
force=force)
assert client_account == staking_address == selected_account
assert mock_stdin.empty()
staking_contract_address = '0xFABADA'
mock_individual_allocation = mocker.Mock(beneficiary_address=selected_account,
contract_address=staking_contract_address)
mock_click_confirm.return_value = YES
mock_stdin.line(YES)
client_account, staking_address = select_client_account_for_staking(emitter=test_emitter,
stakeholder=stakeholder,
individual_allocation=mock_individual_allocation,
@ -65,8 +65,9 @@ def test_select_client_account_for_staking_cli_action(test_emitter,
assert client_account == selected_account
assert staking_address == staking_contract_address
assert mock_stdin.empty()
output = stdout_trap.getvalue()
captured = capsys.readouterr()
message = PREALLOCATION_STAKE_ADVISORY.format(client_account=selected_account,
staking_address=staking_contract_address)
assert message in output
assert message in captured.out

29
tests/integration/cli/actions/test_select_config_file.py
View File

@ -27,7 +27,7 @@ from nucypher.cli.literature import NO_CONFIGURATIONS_ON_DISK
def test_select_config_file_with_no_config_files(test_emitter,
stdout_trap,
capsys,
alice_blockchain_test_config,
tmpdir):
@ -42,17 +42,17 @@ def test_select_config_file_with_no_config_files(test_emitter,
config_root=tmpdir)
# Ensure we notified the user accurately.
output = stdout_trap.getvalue()
captured = capsys.readouterr()
message = NO_CONFIGURATIONS_ON_DISK.format(name=config_class.NAME.capitalize(),
command=config_class.NAME)
assert message in output
assert message in captured.out
def test_auto_select_config_file(test_emitter,
stdout_trap,
capsys,
alice_blockchain_test_config,
tmpdir,
mock_click_prompt):
mock_stdin):
"""Only one configuration was found, so it was chosen automatically"""
config_class = alice_blockchain_test_config
@ -70,18 +70,18 @@ def test_auto_select_config_file(test_emitter,
assert result == str(config_path)
# ... the user was *not* prompted
mock_click_prompt.assert_not_called()
# If they were, `mock_stdin` would complain.
# ...nothing was displayed
output = stdout_trap.getvalue()
assert not output
captured = capsys.readouterr()
assert not captured.out
def test_interactive_select_config_file(test_emitter,
stdout_trap,
capsys,
alice_blockchain_test_config,
tmpdir,
mock_click_prompt,
mock_stdin,
mock_accounts,
patch_keystore):
@ -109,16 +109,17 @@ def test_interactive_select_config_file(test_emitter,
filenames[path] = account.address
assert config_path.exists()
mock_click_prompt.return_value = user_input
mock_stdin.line(str(user_input))
result = select_config_file(emitter=test_emitter,
config_class=config_class,
config_root=tmpdir)
output = stdout_trap.getvalue()
captured = capsys.readouterr()
for filename, account in accounts:
assert account.address in output
assert account.address in captured.out
assert mock_stdin.empty()
table_data = output.split('\n')
table_data = captured.out.split('\n')
table_addresses = [row.split()[1] for row in table_data[2:-2]]
# TODO: Finish this test

9
tests/integration/cli/actions/test_select_network.py
View File

@ -25,11 +25,12 @@ __NETWORKS = NetworksInventory.NETWORKS
@pytest.mark.parametrize('user_input', range(0, len(__NETWORKS)-1))
def test_select_network_cli_action(test_emitter, stdout_trap, mock_click_prompt, user_input):
mock_click_prompt.return_value = user_input
def test_select_network_cli_action(test_emitter, capsys, mock_stdin, user_input):
mock_stdin.line(str(user_input))
selection = __NETWORKS[user_input]
result = select_network(emitter=test_emitter)
assert result == selection
output = stdout_trap.getvalue()
captured = capsys.readouterr()
for name in __NETWORKS:
assert name in output
assert name in captured.out
assert mock_stdin.empty()

52
tests/integration/cli/actions/test_select_stake.py
View File

@ -88,8 +88,8 @@ def test_handle_select_stake_with_no_stakes(test_emitter,
mock_staking_agent,
test_registry,
mock_testerchain,
mock_click_prompt,
stdout_trap):
mock_stdin, # used to assert user hasn't been prompted
capsys):
# Setup
mock_stakes = []
@ -101,9 +101,9 @@ def test_handle_select_stake_with_no_stakes(test_emitter,
select_stake(emitter=test_emitter, stakeholder=stakeholder)
# Examine
output = stdout_trap.getvalue()
assert NO_STAKES_FOUND in output
assert_stake_table_not_painted(output=output)
captured = capsys.readouterr()
assert NO_STAKES_FOUND in captured.out
assert_stake_table_not_painted(output=captured.out)
def test_select_non_divisible_stake(test_emitter,
@ -111,8 +111,8 @@ def test_select_non_divisible_stake(test_emitter,
mock_staking_agent,
test_registry,
mock_testerchain,
mock_click_prompt,
stdout_trap,
mock_stdin,
capsys,
non_divisible_stakes,
stakeholder_with_no_divisible_stakes):
@ -123,7 +123,7 @@ def test_select_non_divisible_stake(test_emitter,
economics=token_economics)
# User's selection
mock_click_prompt.return_value = SELECTION
mock_stdin.line(str(SELECTION))
selected_stake = select_stake(emitter=test_emitter,
divisible=False,
stakeholder=stakeholder_with_no_divisible_stakes)
@ -133,10 +133,11 @@ def test_select_non_divisible_stake(test_emitter,
assert selected_stake == expected_stake
# Examine the output
output = stdout_trap.getvalue()
assert NO_STAKES_FOUND not in output
assert ONLY_DISPLAYING_DIVISIBLE_STAKES_NOTE not in output
assert_stake_table_painted(output=output)
captured = capsys.readouterr()
assert NO_STAKES_FOUND not in captured.out
assert ONLY_DISPLAYING_DIVISIBLE_STAKES_NOTE not in captured.out
assert_stake_table_painted(output=captured.out)
assert mock_stdin.empty()
def test_handle_selection_with_no_divisible_stakes(test_emitter,
@ -144,8 +145,8 @@ def test_handle_selection_with_no_divisible_stakes(test_emitter,
mock_staking_agent,
test_registry,
mock_testerchain,
mock_click_prompt,
stdout_trap,
mock_stdin, # used to assert the user hasn't been prompted
capsys,
non_divisible_stakes):
# Setup
@ -163,10 +164,10 @@ def test_handle_selection_with_no_divisible_stakes(test_emitter,
# Divisible warning was displayed, but having
# no divisible stakes cases an expected failure
output = stdout_trap.getvalue()
assert NO_STAKES_FOUND not in output
assert ONLY_DISPLAYING_DIVISIBLE_STAKES_NOTE in output
assert_stake_table_not_painted(output=output)
captured = capsys.readouterr()
assert NO_STAKES_FOUND not in captured.out
assert ONLY_DISPLAYING_DIVISIBLE_STAKES_NOTE in captured.out
assert_stake_table_not_painted(output=captured.out)
def test_select_divisible_stake(test_emitter,
@ -174,8 +175,8 @@ def test_select_divisible_stake(test_emitter,
mock_staking_agent,
test_registry,
mock_testerchain,
mock_click_prompt,
stdout_trap,
mock_stdin,
capsys,
divisible_stakes,
stakeholder_with_divisible_stakes):
@ -186,7 +187,7 @@ def test_select_divisible_stake(test_emitter,
economics=token_economics)
# SUCCESS: Display all divisible-only stakes and make a selection
mock_click_prompt.return_value = SELECTION
mock_stdin.line(str(SELECTION))
selected_stake = select_stake(emitter=test_emitter,
divisible=True,
@ -196,7 +197,8 @@ def test_select_divisible_stake(test_emitter,
assert selected_stake == expected_stake
# Examine the output
output = stdout_trap.getvalue()
assert NO_STAKES_FOUND not in output
assert ONLY_DISPLAYING_DIVISIBLE_STAKES_NOTE in output
assert_stake_table_painted(output=output)
captured = capsys.readouterr()
assert NO_STAKES_FOUND not in captured.out
assert ONLY_DISPLAYING_DIVISIBLE_STAKES_NOTE in captured.out
assert_stake_table_painted(output=captured.out)
assert mock_stdin.empty()

18
tests/integration/cli/test_worklock_cli_functionality.py
View File

@ -27,7 +27,7 @@ from nucypher.blockchain.eth.interfaces import BlockchainInterface
from nucypher.blockchain.eth.token import NU
from nucypher.cli.commands.worklock import worklock
from nucypher.config.constants import TEMPORARY_DOMAIN
from tests.constants import CLI_TEST_ENV, MOCK_PROVIDER_URI, YES, NO
from tests.constants import CLI_TEST_ENV, MOCK_PROVIDER_URI, YES_ENTER, NO_ENTER
from tests.mock.agents import MockContractAgent
from nucypher.cli.literature import CONFIRM_BID_VERIFICATION
@ -81,7 +81,7 @@ def test_bid_too_soon(click_runner,
a_month_too_soon = now-(3600*30)
mocker.patch.object(BlockchainInterface, 'get_blocktime', return_value=a_month_too_soon)
with pytest.raises(Bidder.BiddingIsClosed):
result = click_runner.invoke(worklock, bidding_command, catch_exceptions=False, input=YES, env=CLI_TEST_ENV)
result = click_runner.invoke(worklock, bidding_command, catch_exceptions=False, input=YES_ENTER, env=CLI_TEST_ENV)
assert result.exit_code != 0
@ -99,7 +99,7 @@ def test_bid_too_late(click_runner,
a_month_too_late = now+(3600*30)
mocker.patch.object(BlockchainInterface, 'get_blocktime', return_value=a_month_too_late)
with pytest.raises(Bidder.BiddingIsClosed):
result = click_runner.invoke(worklock, bidding_command, catch_exceptions=False, input=YES, env=CLI_TEST_ENV)
result = click_runner.invoke(worklock, bidding_command, catch_exceptions=False, input=YES_ENTER, env=CLI_TEST_ENV)
assert result.exit_code != 0
@ -130,7 +130,7 @@ def test_valid_bid(click_runner,
'--network', TEMPORARY_DOMAIN,
'--force')
result = click_runner.invoke(worklock, command, catch_exceptions=False, input=YES, env=CLI_TEST_ENV)
result = click_runner.invoke(worklock, command, catch_exceptions=False, input=YES_ENTER, env=CLI_TEST_ENV)
assert result.exit_code == 0
# OK - Let's see what happened
@ -166,7 +166,7 @@ def test_cancel_bid(click_runner,
'--provider', MOCK_PROVIDER_URI,
'--network', TEMPORARY_DOMAIN,
'--force')
result = click_runner.invoke(worklock, command, input=YES, env=CLI_TEST_ENV, catch_exceptions=False)
result = click_runner.invoke(worklock, command, input=YES_ENTER, env=CLI_TEST_ENV, catch_exceptions=False)
assert result.exit_code == 0
# Bidder
@ -234,7 +234,7 @@ def test_enable_claiming(click_runner,
gas_limit_1 = 200000
gas_limit_2 = 300000
user_input = YES + YES + str(gas_limit_1) + '\n' + NO + str(gas_limit_2) + '\n' + YES
user_input = YES_ENTER + YES_ENTER + str(gas_limit_1) + '\n' + NO_ENTER + str(gas_limit_2) + '\n' + YES_ENTER
result = click_runner.invoke(worklock, command, input=user_input, env=CLI_TEST_ENV, catch_exceptions=False)
assert result.exit_code == 0
confirmation = CONFIRM_BID_VERIFICATION.format(bidder_address=surrogate_bidder.checksum_address,
@ -305,7 +305,7 @@ def test_initial_claim(click_runner,
'--network', TEMPORARY_DOMAIN,
'--force')
result = click_runner.invoke(worklock, command, input=YES, env=CLI_TEST_ENV, catch_exceptions=False)
result = click_runner.invoke(worklock, command, input=YES_ENTER, env=CLI_TEST_ENV, catch_exceptions=False)
assert result.exit_code == 0
mock_worklock_agent.claim.assert_called_once_with(checksum_address=surrogate_bidder.checksum_address)
@ -352,7 +352,7 @@ def test_already_claimed(click_runner,
'--network', TEMPORARY_DOMAIN,
'--force')
result = click_runner.invoke(worklock, command, input=YES, env=CLI_TEST_ENV, catch_exceptions=False)
result = click_runner.invoke(worklock, command, input=YES_ENTER, env=CLI_TEST_ENV, catch_exceptions=False)
assert result.exit_code == 1 # TODO: Decide if this case should error (like now) or simply do nothing
# Bidder
@ -408,7 +408,7 @@ def test_refund(click_runner,
'--network', TEMPORARY_DOMAIN,
'--force')
result = click_runner.invoke(worklock, command, input=YES, env=CLI_TEST_ENV, catch_exceptions=False)
result = click_runner.invoke(worklock, command, input=YES_ENTER, env=CLI_TEST_ENV, catch_exceptions=False)
assert result.exit_code == 0
# Bidder

6
tests/integration/config/test_character_configuration.py
View File

@ -177,7 +177,7 @@ def test_ursula_development_configuration(federated_only=True):
@pytest.mark.skip("See #2016")
def test_destroy_configuration(config,
test_emitter,
stdout_trap,
capsys,
mocker):
# Setup
config_class = config.__class__
@ -193,8 +193,8 @@ def test_destroy_configuration(config,
destroy_configuration(emitter=test_emitter, character_config=config)
mock_config_destroy.assert_called_once()
output = stdout_trap.getvalue()
assert SUCCESSFUL_DESTRUCTION in output
captured = capsys.readouterr()
assert SUCCESSFUL_DESTRUCTION in captured.out
spy_keyring_attached.assert_called_once()
spy_keyring_destroy.assert_called_once()

17
tests/integration/conftest.py
View File

@ -45,6 +45,7 @@ from tests.constants import (
from tests.fixtures import _make_testerchain, make_token_economics
from tests.mock.agents import MockContractAgency, MockContractAgent
from tests.mock.interfaces import MockBlockchain, mock_registry_source_manager
from tests.mock.io import MockStdinWrapper
from tests.utils.config import (
make_alice_test_configuration,
make_bob_test_configuration,
@ -111,13 +112,19 @@ def mock_worklock_agent(mock_testerchain, token_economics, mock_contract_agency)
@pytest.fixture(scope='function')
def mock_click_prompt(mocker):
return mocker.patch.object(click, 'prompt')
def mock_stdin(mocker):
mock = MockStdinWrapper()
@pytest.fixture(scope='function')
def mock_click_confirm(mocker):
return mocker.patch.object(click, 'confirm')
mocker.patch('sys.stdin', new=mock.mock_stdin)
mocker.patch('getpass.getpass', new=mock.mock_getpass)
yield mock
# Sanity check.
# The user is encouraged to `assert mock_stdin.empty()` explicitly in the test
# right after the input-consuming function call.
assert mock.empty(), "Stdin mock was not empty on teardown - some unclaimed input remained"
@pytest.fixture(scope='module', autouse=True)

77
tests/mock/io.py Normal file
View File

@ -0,0 +1,77 @@
"""
This file is part of nucypher.
nucypher is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
nucypher is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with nucypher. If not, see <https://www.gnu.org/licenses/>.
"""
import io
class MockStdinWrapper:
def __init__(self):
self.mock_stdin = MockStdin()
self.mock_getpass = MockGetpass()
def line(self, s):
self.mock_stdin.line(s)
def password(self, s, confirm=False):
self.mock_getpass.line(s)
if confirm:
self.mock_getpass.line(s)
def empty(self):
return self.mock_stdin.empty() and self.mock_getpass.empty()
class MockStdinBase:
def __init__(self):
self.stream = io.StringIO()
self.lines = 0
def line(self, s):
pos = self.stream.tell() # preserve the current read pointer
self.stream.seek(0, io.SEEK_END)
self.stream.write(s + '\n')
self.stream.seek(pos)
self.lines += 1
def _readline(self):
assert self.lines > 0, "Stdin was queried, but the list of mock inputs is empty"
self.lines -= 1
return self.stream.readline()
def empty(self):
return self.lines == 0
class MockGetpass(MockStdinBase):
"""
Mocks `getpass.getpass()`
"""
def __call__(self, prompt):
print(prompt, end='')
s = self._readline()
return s[:-1] # remove the final line break
class MockStdin(MockStdinBase):
"""
Mocks `sys.stdin`
"""
def readline(self):
return self._readline()