keep-it-secret/tests/ext/vault/test_VaultKV2Field.py

# -*- 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}