# -*- coding: utf-8 -*-

# Copyright (c) 2017 Ansible, Inc.
# All Rights Reserved.

from contextlib import contextmanager
import codecs
from uuid import uuid4
import time

from django.conf import LazySettings
from django.core.cache.backends.locmem import LocMemCache
from django.core.exceptions import ImproperlyConfigured
from django.utils.translation import ugettext_lazy as _
import pytest

from awx.conf import models, fields
from awx.conf.settings import SettingsWrapper, EncryptedCacheProxy, SETTING_CACHE_NOTSET
from awx.conf.registry import SettingsRegistry

from awx.main.utils import encrypt_field, decrypt_field


@contextmanager
def apply_patches(_patches):
    [p.start() for p in _patches]
    yield
    [p.stop() for p in _patches]


@pytest.fixture()
def settings(request):
    """
    This fixture initializes a Django settings object that wraps our
    `awx.conf.settings.SettingsWrapper` and passes it as an argument into the
    test function.

    This mimics the work done by `awx.conf.settings.SettingsWrapper.initialize`
    on `django.conf.settings`.
    """
    cache = LocMemCache(str(uuid4()), {})  # make a new random cache each time
    settings = LazySettings()
    registry = SettingsRegistry(settings)
    defaults = {}

    # @pytest.mark.defined_in_file can be used to mark specific setting values
    # as "defined in a settings file".  This is analogous to manually
    # specifying a setting on the filesystem (e.g., in a local_settings.py in
    # development, or in /etc/tower/conf.d/<something>.py)
    for marker in request.node.own_markers:
        if marker.name == 'defined_in_file':
            defaults = marker.kwargs

    defaults['DEFAULTS_SNAPSHOT'] = {}
    settings.configure(**defaults)
    settings._wrapped = SettingsWrapper(settings._wrapped,
                                        cache,
                                        registry)
    return settings


@pytest.mark.defined_in_file(DEBUG=True)
def test_unregistered_setting(settings):
    "native Django settings are not stored in DB, and aren't cached"
    assert settings.DEBUG is True
    assert settings.cache.get('DEBUG') is None


def test_read_only_setting(settings):
    settings.registry.register(
        'AWX_READ_ONLY',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system',
        default='NO-EDITS',
        read_only=True
    )
    assert settings.AWX_READ_ONLY == 'NO-EDITS'
    assert len(settings.registry.get_registered_settings(read_only=False)) == 0
    settings = settings.registry.get_registered_settings(read_only=True)
    assert settings == ['AWX_READ_ONLY']


@pytest.mark.defined_in_file(AWX_SOME_SETTING='DEFAULT')
@pytest.mark.parametrize('read_only', [True, False])
def test_setting_defined_in_file(settings, read_only):
    kwargs = {'read_only': True} if read_only else {}
    settings.registry.register(
        'AWX_SOME_SETTING',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system',
        **kwargs
    )
    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    assert len(settings.registry.get_registered_settings(read_only=False)) == 0
    settings = settings.registry.get_registered_settings(read_only=True)
    assert settings == ['AWX_SOME_SETTING']


@pytest.mark.defined_in_file(AWX_SOME_SETTING='DEFAULT')
def test_setting_defined_in_file_with_empty_default(settings):
    settings.registry.register(
        'AWX_SOME_SETTING',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system',
        default='',
    )
    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    assert len(settings.registry.get_registered_settings(read_only=False)) == 0
    settings = settings.registry.get_registered_settings(read_only=True)
    assert settings == ['AWX_SOME_SETTING']


@pytest.mark.defined_in_file(AWX_SOME_SETTING='DEFAULT')
def test_setting_defined_in_file_with_specific_default(settings):
    settings.registry.register(
        'AWX_SOME_SETTING',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system',
        default=123
    )
    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    assert len(settings.registry.get_registered_settings(read_only=False)) == 0
    settings = settings.registry.get_registered_settings(read_only=True)
    assert settings == ['AWX_SOME_SETTING']


@pytest.mark.defined_in_file(AWX_SOME_SETTING='DEFAULT')
def test_read_only_defaults_are_cached(settings):
    "read-only settings are stored in the cache"
    settings.registry.register(
        'AWX_SOME_SETTING',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system'
    )
    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    assert settings.cache.get('AWX_SOME_SETTING') == 'DEFAULT'


@pytest.mark.defined_in_file(AWX_SOME_SETTING='DEFAULT')
def test_cache_respects_timeout(settings):
    "only preload the cache every SETTING_CACHE_TIMEOUT settings"
    settings.registry.register(
        'AWX_SOME_SETTING',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system'
    )

    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    cache_expiration = settings.cache.get('_awx_conf_preload_expires')
    assert cache_expiration > time.time()

    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    assert settings.cache.get('_awx_conf_preload_expires') == cache_expiration


def test_default_setting(settings, mocker):
    "settings that specify a default are inserted into the cache"
    settings.registry.register(
        'AWX_SOME_SETTING',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system',
        default='DEFAULT'
    )

    settings_to_cache = mocker.Mock(**{'order_by.return_value': []})
    with mocker.patch('awx.conf.models.Setting.objects.filter', return_value=settings_to_cache):
        assert settings.AWX_SOME_SETTING == 'DEFAULT'
        assert settings.cache.get('AWX_SOME_SETTING') == 'DEFAULT'


@pytest.mark.defined_in_file(AWX_SOME_SETTING='DEFAULT')
def test_setting_is_from_setting_file(settings, mocker):
    settings.registry.register(
        'AWX_SOME_SETTING',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system'
    )
    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    assert settings.registry.get_setting_field('AWX_SOME_SETTING').defined_in_file is True


def test_setting_is_not_from_setting_file(settings, mocker):
    settings.registry.register(
        'AWX_SOME_SETTING',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system',
        default='DEFAULT'
    )

    settings_to_cache = mocker.Mock(**{'order_by.return_value': []})
    with mocker.patch('awx.conf.models.Setting.objects.filter', return_value=settings_to_cache):
        assert settings.AWX_SOME_SETTING == 'DEFAULT'
        assert settings.registry.get_setting_field('AWX_SOME_SETTING').defined_in_file is False


def test_empty_setting(settings, mocker):
    "settings with no default and no defined value are not valid"
    settings.registry.register(
        'AWX_SOME_SETTING',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system'
    )

    mocks = mocker.Mock(**{
        'order_by.return_value': mocker.Mock(**{
            '__iter__': lambda self: iter([]),
            'first.return_value': None
        }),
    })
    with mocker.patch('awx.conf.models.Setting.objects.filter', return_value=mocks):
        with pytest.raises(AttributeError):
            settings.AWX_SOME_SETTING
        assert settings.cache.get('AWX_SOME_SETTING') == SETTING_CACHE_NOTSET


def test_setting_from_db(settings, mocker):
    "settings can be loaded from the database"
    settings.registry.register(
        'AWX_SOME_SETTING',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system',
        default='DEFAULT'
    )

    setting_from_db = mocker.Mock(key='AWX_SOME_SETTING', value='FROM_DB')
    mocks = mocker.Mock(**{
        'order_by.return_value': mocker.Mock(**{
            '__iter__': lambda self: iter([setting_from_db]),
            'first.return_value': setting_from_db
        }),
    })
    with mocker.patch('awx.conf.models.Setting.objects.filter', return_value=mocks):
        assert settings.AWX_SOME_SETTING == 'FROM_DB'
        assert settings.cache.get('AWX_SOME_SETTING') == 'FROM_DB'


@pytest.mark.defined_in_file(AWX_SOME_SETTING='DEFAULT')
def test_read_only_setting_assignment(settings):
    "read-only settings cannot be overwritten"
    settings.registry.register(
        'AWX_SOME_SETTING',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system'
    )
    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    with pytest.raises(ImproperlyConfigured):
        settings.AWX_SOME_SETTING = 'CHANGED'
    assert settings.AWX_SOME_SETTING == 'DEFAULT'


def test_db_setting_create(settings, mocker):
    "settings are stored in the database when set for the first time"
    settings.registry.register(
        'AWX_SOME_SETTING',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system'
    )

    setting_list = mocker.Mock(**{'order_by.return_value.first.return_value': None})
    with apply_patches([
        mocker.patch('awx.conf.models.Setting.objects.filter',
                     return_value=setting_list),
        mocker.patch('awx.conf.models.Setting.objects.create', mocker.Mock())
    ]):
        settings.AWX_SOME_SETTING = 'NEW-VALUE'

    models.Setting.objects.create.assert_called_with(
        key='AWX_SOME_SETTING',
        user=None,
        value='NEW-VALUE'
    )


def test_db_setting_update(settings, mocker):
    "settings are updated in the database when their value changes"
    settings.registry.register(
        'AWX_SOME_SETTING',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system'
    )

    existing_setting = mocker.Mock(key='AWX_SOME_SETTING', value='FROM_DB')
    setting_list = mocker.Mock(**{
        'order_by.return_value.first.return_value': existing_setting
    })
    with mocker.patch('awx.conf.models.Setting.objects.filter', return_value=setting_list):
        settings.AWX_SOME_SETTING = 'NEW-VALUE'

    assert existing_setting.value == 'NEW-VALUE'
    existing_setting.save.assert_called_with(update_fields=['value'])


def test_db_setting_deletion(settings, mocker):
    "settings are auto-deleted from the database"
    settings.registry.register(
        'AWX_SOME_SETTING',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system'
    )

    existing_setting = mocker.Mock(key='AWX_SOME_SETTING', value='FROM_DB')
    with mocker.patch('awx.conf.models.Setting.objects.filter', return_value=[existing_setting]):
        del settings.AWX_SOME_SETTING

    assert existing_setting.delete.call_count == 1


@pytest.mark.defined_in_file(AWX_SOME_SETTING='DEFAULT')
def test_read_only_setting_deletion(settings):
    "read-only settings cannot be deleted"
    settings.registry.register(
        'AWX_SOME_SETTING',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system'
    )
    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    with pytest.raises(ImproperlyConfigured):
        del settings.AWX_SOME_SETTING
    assert settings.AWX_SOME_SETTING == 'DEFAULT'


def test_charfield_properly_sets_none(settings, mocker):
    "see: https://github.com/ansible/ansible-tower/issues/5322"
    settings.registry.register(
        'AWX_SOME_SETTING',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system',
        allow_null=True
    )

    setting_list = mocker.Mock(**{'order_by.return_value.first.return_value': None})
    with apply_patches([
        mocker.patch('awx.conf.models.Setting.objects.filter',
                     return_value=setting_list),
        mocker.patch('awx.conf.models.Setting.objects.create', mocker.Mock())
    ]):
        settings.AWX_SOME_SETTING = None

    models.Setting.objects.create.assert_called_with(
        key='AWX_SOME_SETTING',
        user=None,
        value=None
    )


def test_settings_use_cache(settings, mocker):
    settings.registry.register(
        'AWX_VAR',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system'
    )
    settings.cache.set('AWX_VAR', 'foobar')
    settings.cache.set('_awx_conf_preload_expires', 100)
    # Will fail test if database is used
    getattr(settings, 'AWX_VAR')


def test_settings_use_an_encrypted_cache(settings, mocker):
    settings.registry.register(
        'AWX_ENCRYPTED',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system',
        encrypted=True
    )
    assert isinstance(settings.cache, EncryptedCacheProxy)
    assert settings.cache.__dict__['encrypter'] == encrypt_field
    assert settings.cache.__dict__['decrypter'] == decrypt_field
    settings.cache.set('AWX_ENCRYPTED_ID', 402)
    settings.cache.set('AWX_ENCRYPTED', 'foobar')
    settings.cache.set('_awx_conf_preload_expires', 100)
    # Will fail test if database is used
    getattr(settings, 'AWX_ENCRYPTED')


def test_sensitive_cache_data_is_encrypted(settings, mocker):
    "fields marked as `encrypted` are stored in the cache with encryption"
    settings.registry.register(
        'AWX_ENCRYPTED',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system',
        encrypted=True
    )

    def rot13(obj, attribute):
        assert obj.pk == 123
        return codecs.encode(getattr(obj, attribute), 'rot_13')

    native_cache = LocMemCache(str(uuid4()), {})
    cache = EncryptedCacheProxy(
        native_cache,
        settings.registry,
        encrypter=rot13,
        decrypter=rot13
    )
    # Insert the setting value into the database; the encryption process will
    # use its primary key as part of the encryption key
    setting_from_db = mocker.Mock(pk=123, key='AWX_ENCRYPTED', value='SECRET!')
    mocks = mocker.Mock(**{
        'order_by.return_value': mocker.Mock(**{
            '__iter__': lambda self: iter([setting_from_db]),
            'first.return_value': setting_from_db
        }),
    })
    with mocker.patch('awx.conf.models.Setting.objects.filter', return_value=mocks):
        cache.set('AWX_ENCRYPTED', 'SECRET!')
        assert cache.get('AWX_ENCRYPTED') == 'SECRET!'
        assert native_cache.get('AWX_ENCRYPTED') == 'FRPERG!'


def test_readonly_sensitive_cache_data_is_encrypted(settings):
    "readonly fields marked as `encrypted` are stored in the cache with encryption"
    settings.registry.register(
        'AWX_ENCRYPTED',
        field_class=fields.CharField,
        category=_('System'),
        category_slug='system',
        read_only=True,
        encrypted=True
    )

    def rot13(obj, attribute):
        assert obj.pk is None
        return codecs.encode(getattr(obj, attribute), 'rot_13')

    native_cache = LocMemCache(str(uuid4()), {})
    cache = EncryptedCacheProxy(
        native_cache,
        settings.registry,
        encrypter=rot13,
        decrypter=rot13
    )
    cache.set('AWX_ENCRYPTED', 'SECRET!')
    assert cache.get('AWX_ENCRYPTED') == 'SECRET!'
    assert native_cache.get('AWX_ENCRYPTED') == 'FRPERG!'