Get existing tests passing.

This commit is contained in:
Dan Helfman
2025-02-12 22:49:16 -08:00
parent 8745b9939d
commit d6cf48544a
4 changed files with 39 additions and 21 deletions
+1 -1
View File
@@ -42,5 +42,5 @@ def resolve_credential(value):
raise ValueError(f'Cannot load credential with invalid syntax "{value}"')
return borgmatic.hooks.dispatch.call_hook(
'load_credential', {}, hook_name, credential_parameters
'load_credential', {}, hook_name, tuple(credential_parameters)
)
+17 -11
View File
@@ -5,29 +5,35 @@ import re
logger = logging.getLogger(__name__)
SECRET_NAME_PATTERN = re.compile(r'^\w+$')
SECRETS_DIRECTORY = '/run/secrets'
CREDENTIAL_NAME_PATTERN = re.compile(r'^\w+$')
def load_credential(hook_config, config, credential_parameters):
'''
Given the hook configuration dict, the configuration dict, and a credential parameters tuple
containing a secret name to load, read the secret from the corresponding container secrets file
and return it.
containing a credential name to load, read the credential from the corresponding systemd
credential file and return it.
Raise ValueError if the credential parameters is not one element, the secret name is invalid, or
the secret file cannot be read.
Raise ValueError if the systemd CREDENTIALS_DIRECTORY environment variable is not set, the
credential name is invalid, or the credential file cannot be read.
'''
try:
(secert_name,) = credential_parameters
(credential_name,) = credential_parameters
except ValueError:
raise ValueError(f'Cannot load invalid secret name: "{' '.join(credential_parameters)}"')
raise ValueError(f'Cannot load invalid credential name: "{' '.join(credential_parameters)}"')
if not SECRET_NAME_PATTERN.match(SECRET_NAME):
raise ValueError(f'Cannot load invalid secret name: "{credential_name}"')
credentials_directory = os.environ.get('CREDENTIALS_DIRECTORY')
if not credentials_directory:
raise ValueError(
f'Cannot load credential "{credential_name}" because the systemd CREDENTIALS_DIRECTORY environment variable is not set'
)
if not CREDENTIAL_NAME_PATTERN.match(credential_name):
raise ValueError(f'Cannot load invalid credential name "{credential_name}"')
try:
with open(os.path.join(SECRETS_DIRECTORY, credential_name)) as credential_file:
with open(os.path.join(credentials_directory, credential_name)) as credential_file:
return credential_file.read().rstrip(os.linesep)
except (FileNotFoundError, OSError) as error:
logger.warning(error)
+17 -5
View File
@@ -4,7 +4,7 @@ from flexmock import flexmock
from borgmatic.hooks.credential import parse as module
def test_resolve_credential_passes_through_string_without_credential_tag():
def test_resolve_credential_passes_through_string_without_credential():
module.resolve_credential.cache_clear()
flexmock(module.borgmatic.hooks.dispatch).should_receive('call_hook').never()
@@ -19,7 +19,7 @@ def test_resolve_credential_passes_through_none():
@pytest.mark.parametrize('invalid_value', ('{credential}', '{credential }', '{credential systemd}'))
def test_resolve_credential_with_invalid_credential_tag_raises(invalid_value):
def test_resolve_credential_with_invalid_credential_raises(invalid_value):
module.resolve_credential.cache_clear()
flexmock(module.borgmatic.hooks.dispatch).should_receive('call_hook').never()
@@ -27,25 +27,37 @@ def test_resolve_credential_with_invalid_credential_tag_raises(invalid_value):
module.resolve_credential(invalid_value)
def test_resolve_credential_with_valid_credential_tag_loads_credential():
def test_resolve_credential_with_valid_credential_loads_credential():
module.resolve_credential.cache_clear()
flexmock(module.borgmatic.hooks.dispatch).should_receive('call_hook').with_args(
'load_credential',
{},
'systemd',
'mycredential',
('mycredential',),
).and_return('result').once()
assert module.resolve_credential('{credential systemd mycredential}') == 'result'
def test_resolve_credential_with_valid_credential_and_quoted_parameters_loads_credential():
module.resolve_credential.cache_clear()
flexmock(module.borgmatic.hooks.dispatch).should_receive('call_hook').with_args(
'load_credential',
{},
'systemd',
('my credential',),
).and_return('result').once()
assert module.resolve_credential('{credential systemd "my credential"}') == 'result'
def test_resolve_credential_caches_credential_after_first_call():
module.resolve_credential.cache_clear()
flexmock(module.borgmatic.hooks.dispatch).should_receive('call_hook').with_args(
'load_credential',
{},
'systemd',
'mycredential',
('mycredential',),
).and_return('result').once()
assert module.resolve_credential('{credential systemd mycredential}') == 'result'
+4 -4
View File
@@ -13,7 +13,7 @@ def test_load_credential_without_credentials_directory_raises():
)
with pytest.raises(ValueError):
module.load_credential(hook_config={}, config={}, credential_name='mycredential')
module.load_credential(hook_config={}, config={}, credential_parameters=('mycredential',))
def test_load_credential_with_invalid_credential_name_raises():
@@ -22,7 +22,7 @@ def test_load_credential_with_invalid_credential_name_raises():
)
with pytest.raises(ValueError):
module.load_credential(hook_config={}, config={}, credential_name='../../my!@#$credential')
module.load_credential(hook_config={}, config={}, credential_parameters=('../../my!@#$credential',))
def test_load_credential_reads_named_credential_from_file():
@@ -35,7 +35,7 @@ def test_load_credential_reads_named_credential_from_file():
builtins.should_receive('open').with_args('/var/mycredential').and_return(credential_stream)
assert (
module.load_credential(hook_config={}, config={}, credential_name='mycredential')
module.load_credential(hook_config={}, config={}, credential_parameters=('mycredential',))
== 'password'
)
@@ -48,4 +48,4 @@ def test_load_credential_with_file_not_found_error_raises():
builtins.should_receive('open').with_args('/var/mycredential').and_raise(FileNotFoundError)
with pytest.raises(ValueError):
module.load_credential(hook_config={}, config={}, credential_name='mycredential')
module.load_credential(hook_config={}, config={}, credential_parameters=('mycredential',))