test_caches_crypto_cache_entry.py

# Copyright 2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"). You
# may not use this file except in compliance with the License. A copy of
# the License is located at
#
# http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file is
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
"""Unit test suite for CryptoMaterialsCacheEntry and CryptoMaterialsCacheEntryHints"""
from mock import MagicMock
import pytest
from pytest_mock import mocker  # noqa pylint: disable=unused-import

import aws_encryption_sdk.caches
from aws_encryption_sdk.caches import (
    CryptoMaterialsCacheEntry, CryptoMaterialsCacheEntryHints
)
from aws_encryption_sdk.exceptions import NotSupportedError
from aws_encryption_sdk.materials_managers import DecryptionMaterials, EncryptionMaterials

pytestmark = [pytest.mark.unit, pytest.mark.local]

_VALID_KWARGS = {
    'CryptoMaterialsCacheEntryHints': dict(lifetime=5.0),
    'CryptoMaterialsCacheEntry': dict(
        cache_key=b'this is a cache key',
        value=MagicMock(__class__=EncryptionMaterials),
        hints=MagicMock(__class__=CryptoMaterialsCacheEntryHints)
    )
}


@pytest.yield_fixture
def patch_time(mocker):
    mocker.patch.object(aws_encryption_sdk.caches.time, 'time')
    aws_encryption_sdk.caches.time.time.side_effect = (3.0, 10.0)
    yield 7.0


@pytest.mark.parametrize('attr_class, invalid_kwargs', (
    (CryptoMaterialsCacheEntryHints, dict(lifetime=4)),
    (CryptoMaterialsCacheEntry, dict(cache_key=None)),
    (CryptoMaterialsCacheEntry, dict(value=None))
))
def test_attributes_fails(attr_class, invalid_kwargs):
    kwargs = _VALID_KWARGS[attr_class.__name__].copy()
    kwargs.update(invalid_kwargs)
    with pytest.raises(TypeError):
        attr_class(**kwargs)


@pytest.mark.parametrize('attr_class', (
    CryptoMaterialsCacheEntry,
    CryptoMaterialsCacheEntryHints
))
def test_attributes_valid(attr_class):
    attr_class(**_VALID_KWARGS[attr_class.__name__])


@pytest.mark.parametrize('valid_kwargs_overrides', (
    dict(value=MagicMock(__class__=EncryptionMaterials)),
    dict(value=MagicMock(__class__=DecryptionMaterials))
))
def test_crypto_cache_entry_valid_attributes(valid_kwargs_overrides):
    kwargs = _VALID_KWARGS['CryptoMaterialsCacheEntry'].copy()
    kwargs.update(valid_kwargs_overrides)
    CryptoMaterialsCacheEntry(**kwargs)


def test_crypto_cache_entry_init(patch_time):
    entry = CryptoMaterialsCacheEntry(
        cache_key=b'ex_cache_key',
        value=MagicMock(__class__=EncryptionMaterials),
        hints=CryptoMaterialsCacheEntryHints(lifetime=10.0)
    )

    assert entry.creation_time == 3.0
    assert entry.bytes_encrypted == 0
    assert entry.messages_encrypted == 0
    assert entry.valid
    # Because Lock is a helper function that returns whatever native primitive is best,
    #  we cannot test for type.  Instead, testing for interface.
    assert hasattr(entry._lock, 'acquire')
    assert callable(entry._lock.acquire)
    assert hasattr(entry._lock, 'release')
    assert callable(entry._lock.release)


def test_crypto_cache_entry_defaults():
    entry = CryptoMaterialsCacheEntry(
        cache_key=b'ex_cache_key',
        value=MagicMock(__class__=EncryptionMaterials)
    )

    assert entry.hints == CryptoMaterialsCacheEntryHints()


def test_crypto_cache_entry_setattr():
    entry = CryptoMaterialsCacheEntry(**_VALID_KWARGS['CryptoMaterialsCacheEntry'])

    with pytest.raises(NotSupportedError) as excinfo:
        entry.bytes_encrypted = 0

    excinfo.match(r'Attributes may not be set on CryptoMaterialsCacheEntry objects')


def test_crypto_cache_entry_age(patch_time):
    entry = CryptoMaterialsCacheEntry(**_VALID_KWARGS['CryptoMaterialsCacheEntry'])

    assert entry.age == patch_time


def test_crypto_cache_entry_is_too_old_no_lifetime_hint(patch_time):
    kwargs = _VALID_KWARGS['CryptoMaterialsCacheEntry'].copy()
    kwargs['hints'] = MagicMock(__class__=CryptoMaterialsCacheEntryHints, lifetime=None)
    entry = CryptoMaterialsCacheEntry(**kwargs)

    assert not entry.is_too_old()


@pytest.mark.parametrize('age_modifier, result', (
    (-1.0, True),
    (0.0, False),
    (1.0, False)
))
def test_crypto_cache_entry_is_too_old(patch_time, age_modifier, result):
    lifetime = patch_time + age_modifier
    kwargs = _VALID_KWARGS['CryptoMaterialsCacheEntry'].copy()
    kwargs['hints'] = MagicMock(__class__=CryptoMaterialsCacheEntryHints, lifetime=lifetime)
    entry = CryptoMaterialsCacheEntry(**kwargs)

    if result:
        assert entry.is_too_old()
    else:
        assert not entry.is_too_old()


def test_crypto_cache_entry_update_with_message_bytes_encrypted():
    entry = CryptoMaterialsCacheEntry(**_VALID_KWARGS['CryptoMaterialsCacheEntry'])
    super(CryptoMaterialsCacheEntry, entry).__setattr__('messages_encrypted', 10)
    super(CryptoMaterialsCacheEntry, entry).__setattr__('bytes_encrypted', 10)

    entry._update_with_message_bytes_encrypted(50)

    assert entry.messages_encrypted == 11
    assert entry.bytes_encrypted == 60


def test_crypto_cache_entry_invalidate():
    entry = CryptoMaterialsCacheEntry(**_VALID_KWARGS['CryptoMaterialsCacheEntry'])

    assert entry.valid

    entry.invalidate()

    assert not entry.valid