keep-it-secret/tests/ext/vault/test_VaultKV2Field.py
2024-02-08 20:52:08 +00:00

213 lines
4.9 KiB
Python

# -*- coding: utf-8 -*-
# type: ignore
from __future__ import annotations
from unittest import mock
import hvac
import pytest
from pytest_mock import MockerFixture
from keep_it_secret import Secrets
from keep_it_secret.ext import vault
@pytest.fixture
def field() -> vault.VaultKV2Field:
return vault.VaultKV2Field('keep_it_secret/', 'tests/spam')
@pytest.fixture
def as_type() -> mock.Mock:
return mock.Mock()
@pytest.fixture
def field_with_as_type(as_type: mock.Mock) -> vault.VaultKV2Field:
return vault.VaultKV2Field('keep_it_secret/', 'tests/spam', as_type=as_type)
@pytest.fixture
def hvac_kv_v2_client(mocker: MockerFixture) -> mock.Mock:
result = mock.Mock()
result.secrets.kv.v2 = mock.Mock(spec=['read_secret_version'])
return result
def test_init():
# When
result = vault.VaultKV2Field('keep_it_secret/', 'tests/spam')
# Then
assert result.mount_point == 'keep_it_secret/'
assert result.path == 'tests/spam'
assert result.version is None
assert result.default is None
assert result.cast_to is None
assert result.as_type is None
def test_init_with_version():
# When
result = vault.VaultKV2Field(
'keep_it_secret/', 'tests/spam', version='1',
)
# Then
assert result.version == '1'
def test_init_with_default():
# When
result = vault.VaultKV2Field(
'keep_it_secret/', 'tests/spam', default='eggs',
)
# Then
assert result.default == 'eggs'
def test_init_with_as_type():
# When
result = vault.VaultKV2Field(
'keep_it_secret/', 'tests/spam', as_type=dict,
)
# Then
assert result.as_type == result.cast
assert result.cast_to == dict
def test_init_with_field_options():
# When
result = vault.VaultKV2Field(
'keep_it_secret/', 'tests/spam', required=False, description='eggs',
)
# Then
assert result.required is False
assert result.description == 'eggs'
def test_new(mocker: MockerFixture):
# Given
mock_init = mocker.patch.object(
vault.VaultKV2Field, '__init__', return_value=None,
)
# When
_ = vault.VaultKV2Field.new(
'keep_it_secret/',
'tests/spam',
version='1',
default=dict,
as_type=dict,
required=False,
description='eggs',
)
# Then
mock_init.assert_called_once_with(
'keep_it_secret/',
'tests/spam',
version='1',
default=dict,
as_type=dict,
required=False,
description='eggs',
)
def test_cast_not_dict(field: vault.VaultKV2Field):
# When
result = field.cast('spam')
# Then
assert result == 'spam'
def test_cast(field_with_as_type: vault.VaultKV2Field, as_type: mock.Mock):
# Given
as_type.return_value = 'spam'
# When
result = field_with_as_type.cast({
'spam': True,
'eggs': False,
})
# Then
assert result == 'spam'
as_type.assert_called_once_with(
spam=True,
eggs=False,
)
def test_get_value_vault_dependency_missing(field: vault.VaultKV2Field,
testing_secrets: Secrets):
# Given
with pytest.raises(field.DependencyMissing) as exception_info:
# When
_ = field(testing_secrets)
# Then
assert exception_info.value.args[0] == 'vault'
def test_get_value_required_value_not_found(testing_vault_secrets: Secrets,
hvac_kv_v2_client: mock.Mock,
field: vault.VaultKV2Field):
# Given
testing_vault_secrets.vault.client = hvac_kv_v2_client
hvac_kv_v2_client.secrets.kv.v2.read_secret_version.side_effect = hvac.exceptions.InvalidPath()
with pytest.raises(field.RequiredValueMissing) as exception_info:
# When
_ = field(testing_vault_secrets)
# Then
assert exception_info.value.args[0] == 'keep_it_secret/tests/spam'
def test_get_value_not_found_not_required(testing_vault_secrets: Secrets,
hvac_kv_v2_client: mock.Mock,
field: vault.VaultKV2Field):
# Given
testing_vault_secrets.vault.client = hvac_kv_v2_client
hvac_kv_v2_client.secrets.kv.v2.read_secret_version.side_effect = hvac.exceptions.InvalidPath()
field = vault.VaultKV2Field(
'keep_it_secret/', 'tests/spam', required=False,
)
result = field(testing_vault_secrets)
# Then
assert result is None
def test_get_value(testing_vault_secrets: Secrets,
hvac_kv_v2_client: mock.Mock,
field: vault.VaultKV2Field):
# Given
testing_vault_secrets.vault.client = hvac_kv_v2_client
hvac_kv_v2_client.secrets.kv.v2.read_secret_version.return_value = {
'data': {
'data': {
'spam': True,
},
},
}
# When
result = field(testing_vault_secrets)
# Then
assert result == {'spam': True}