diff --git a/awx/conf/apps.py b/awx/conf/apps.py index 6e09545236..9ae459fb35 100644 --- a/awx/conf/apps.py +++ b/awx/conf/apps.py @@ -22,4 +22,3 @@ class ConfConfig(AppConfig): if 'http_receiver' not in LOGGING_DICT['loggers']['awx']['handlers']: LOGGING_DICT['loggers']['awx']['handlers'] += ['http_receiver'] configure_logging(settings.LOGGING_CONFIG, LOGGING_DICT) - # checks.register(SettingsWrapper._check_settings) diff --git a/awx/conf/registry.py b/awx/conf/registry.py index 5a4334c181..dd4fdb7e6a 100644 --- a/awx/conf/registry.py +++ b/awx/conf/registry.py @@ -18,9 +18,18 @@ __all__ = ['settings_registry'] class SettingsRegistry(object): """Registry of all API-configurable settings and categories.""" - def __init__(self): + def __init__(self, settings=None): + """ + :param settings: a ``django.conf.LazySettings`` object used to lookup + file-based field values (e.g., ``local_settings.py`` + and ``/etc/tower/conf.d/example.py``). If unspecified, + defaults to ``django.conf.settings``. + """ + if settings is None: + from django.conf import settings self._registry = OrderedDict() self._dependent_settings = {} + self.settings = settings def register(self, setting, **kwargs): if setting in self._registry: @@ -94,7 +103,6 @@ class SettingsRegistry(object): return bool(self._registry.get(setting, {}).get('encrypted', False)) def get_setting_field(self, setting, mixin_class=None, for_user=False, **kwargs): - from django.conf import settings from rest_framework.fields import empty field_kwargs = {} field_kwargs.update(self._registry[setting]) @@ -124,12 +132,12 @@ class SettingsRegistry(object): original_field_instance = original_field_class(**field_kwargs) if category_slug == 'user' and for_user: try: - field_instance.default = original_field_instance.to_representation(getattr(settings, setting)) + field_instance.default = original_field_instance.to_representation(getattr(self.settings, setting)) except: logger.warning('Unable to retrieve default value for user setting "%s".', setting, exc_info=True) elif not field_instance.read_only or field_instance.default is empty or not field_instance.default: try: - field_instance.default = original_field_instance.to_representation(settings._awx_conf_settings._get_default(setting)) + field_instance.default = original_field_instance.to_representation(self.settings._awx_conf_settings._get_default(setting)) except AttributeError: pass except: diff --git a/awx/conf/settings.py b/awx/conf/settings.py index d5e379ba9f..076340afa9 100644 --- a/awx/conf/settings.py +++ b/awx/conf/settings.py @@ -7,8 +7,7 @@ import time # Django from django.conf import settings, UserSettingsHolder -from django.core.cache import cache -from django.core import checks +from django.core.cache import cache as django_cache from django.core.exceptions import ImproperlyConfigured from django.db import ProgrammingError, OperationalError @@ -65,35 +64,42 @@ def _log_database_error(): class SettingsWrapper(UserSettingsHolder): @classmethod - def initialize(cls): + def initialize(cls, cache=None, registry=None): + """ + Used to initialize and wrap the Django settings context. + + :param cache: the Django cache backend to use for caching setting + values. ``django.core.cache`` is used by default. + :param registry: the settings registry instance used. The global + ``awx.conf.settings_registry`` is used by default. + """ if not getattr(settings, '_awx_conf_settings', False): - settings_wrapper = cls(settings._wrapped) + settings_wrapper = cls( + settings._wrapped, + cache=cache or django_cache, + registry=registry or settings_registry + ) settings._wrapped = settings_wrapper - @classmethod - def _check_settings(cls, app_configs, **kwargs): - errors = [] - # FIXME: Warn if database not available! - for setting in Setting.objects.filter(key__in=settings_registry.get_registered_settings(), user__isnull=True): - field = settings_registry.get_setting_field(setting.key) - try: - field.to_internal_value(setting.value) - except Exception as e: - errors.append(checks.Error(str(e))) - return errors - - def __init__(self, default_settings): + def __init__(self, default_settings, cache, registry): + """ + This constructor is generally not called directly, but by + ``SettingsWrapper.initialize`` at app startup time when settings are + parsed. + """ self.__dict__['default_settings'] = default_settings self.__dict__['_awx_conf_settings'] = self self.__dict__['_awx_conf_preload_expires'] = None self.__dict__['_awx_conf_preload_lock'] = threading.RLock() self.__dict__['_awx_conf_init_readonly'] = False + self.__dict__['cache'] = cache + self.__dict__['registry'] = registry def _get_supported_settings(self): - return settings_registry.get_registered_settings() + return self.registry.get_registered_settings() def _get_writeable_settings(self): - return settings_registry.get_registered_settings(read_only=False) + return self.registry.get_registered_settings(read_only=False) def _get_cache_value(self, value): if value is None: @@ -124,11 +130,11 @@ class SettingsWrapper(UserSettingsHolder): file_default = None if file_default != init_default and file_default is not None: logger.warning('Setting %s has been marked read-only!', key) - settings_registry._registry[key]['read_only'] = True + self.registry._registry[key]['read_only'] = True self.__dict__['_awx_conf_init_readonly'] = True # If local preload timer has expired, check to see if another process # has already preloaded the cache and skip preloading if so. - if cache.get('_awx_conf_preload_expires', empty) is not empty: + if self.cache.get('_awx_conf_preload_expires', empty) is not empty: return # Initialize all database-configurable settings with a marker value so # to indicate from the cache that the setting is not configured without @@ -138,7 +144,7 @@ class SettingsWrapper(UserSettingsHolder): for setting in Setting.objects.filter(key__in=settings_to_cache.keys(), user__isnull=True).order_by('pk'): if settings_to_cache[setting.key] != SETTING_CACHE_NOTSET: continue - if settings_registry.is_setting_encrypted(setting.key): + if self.registry.is_setting_encrypted(setting.key): value = decrypt_field(setting, 'value') else: value = setting.value @@ -148,7 +154,7 @@ class SettingsWrapper(UserSettingsHolder): for key, value in settings_to_cache.items(): if value != SETTING_CACHE_NOTSET: continue - field = settings_registry.get_setting_field(key) + field = self.registry.get_setting_field(key) try: settings_to_cache[key] = self._get_cache_value(field.get_default()) except SkipField: @@ -157,13 +163,13 @@ class SettingsWrapper(UserSettingsHolder): settings_to_cache = dict([(Setting.get_cache_key(k), v) for k, v in settings_to_cache.items()]) settings_to_cache['_awx_conf_preload_expires'] = self._awx_conf_preload_expires logger.debug('cache set_many(%r, %r)', settings_to_cache, SETTING_CACHE_TIMEOUT) - cache.set_many(settings_to_cache, SETTING_CACHE_TIMEOUT) + self.cache.set_many(settings_to_cache, SETTING_CACHE_TIMEOUT) def _get_local(self, name): self._preload_cache() cache_key = Setting.get_cache_key(name) try: - cache_value = cache.get(cache_key, empty) + cache_value = self.cache.get(cache_key, empty) except ValueError: cache_value = empty logger.debug('cache get(%r, %r) -> %r', cache_key, empty, cache_value) @@ -177,7 +183,7 @@ class SettingsWrapper(UserSettingsHolder): value = {} else: value = cache_value - field = settings_registry.get_setting_field(name) + field = self.registry.get_setting_field(name) if value is empty: setting = None if not field.read_only: @@ -198,8 +204,10 @@ class SettingsWrapper(UserSettingsHolder): if value is None and SETTING_CACHE_NOTSET == SETTING_CACHE_NONE: value = SETTING_CACHE_NOTSET if cache_value != value: - logger.debug('cache set(%r, %r, %r)', cache_key, self._get_cache_value(value), SETTING_CACHE_TIMEOUT) - cache.set(cache_key, self._get_cache_value(value), SETTING_CACHE_TIMEOUT) + logger.debug('cache set(%r, %r, %r)', cache_key, + self._get_cache_value(value), + SETTING_CACHE_TIMEOUT) + self.cache.set(cache_key, self._get_cache_value(value), SETTING_CACHE_TIMEOUT) if value == SETTING_CACHE_NOTSET and not SETTING_CACHE_DEFAULTS: try: value = field.get_default() @@ -214,7 +222,9 @@ class SettingsWrapper(UserSettingsHolder): else: return field.run_validation(value) except: - logger.warning('The current value "%r" for setting "%s" is invalid.', value, name, exc_info=True) + logger.warning( + 'The current value "%r" for setting "%s" is invalid.', + value, name, exc_info=True) return empty def _get_default(self, name): @@ -234,7 +244,7 @@ class SettingsWrapper(UserSettingsHolder): return self._get_default(name) def _set_local(self, name, value): - field = settings_registry.get_setting_field(name) + field = self.registry.get_setting_field(name) if field.read_only: logger.warning('Attempt to set read only setting "%s".', name) raise ImproperlyConfigured('Setting "%s" is read only.'.format(name)) @@ -244,7 +254,8 @@ class SettingsWrapper(UserSettingsHolder): setting_value = field.run_validation(data) db_value = field.to_representation(setting_value) except Exception as e: - logger.exception('Unable to assign value "%r" to setting "%s".', value, name, exc_info=True) + logger.exception('Unable to assign value "%r" to setting "%s".', + value, name, exc_info=True) raise e setting = Setting.objects.filter(key=name, user__isnull=True).order_by('pk').first() @@ -264,7 +275,7 @@ class SettingsWrapper(UserSettingsHolder): setattr(self.default_settings, name, value) def _del_local(self, name): - field = settings_registry.get_setting_field(name) + field = self.registry.get_setting_field(name) if field.read_only: logger.warning('Attempt to delete read only setting "%s".', name) raise ImproperlyConfigured('Setting "%s" is read only.'.format(name)) @@ -282,7 +293,8 @@ class SettingsWrapper(UserSettingsHolder): def __dir__(self): keys = [] with _log_database_error(): - for setting in Setting.objects.filter(key__in=self._get_supported_settings(), user__isnull=True): + for setting in Setting.objects.filter( + key__in=self._get_supported_settings(), user__isnull=True): # Skip returning settings that have been overridden but are # considered to be "not set". if setting.value is None and SETTING_CACHE_NOTSET == SETTING_CACHE_NONE: diff --git a/awx/conf/tests/__init__.py b/awx/conf/tests/__init__.py new file mode 100644 index 0000000000..46176c348f --- /dev/null +++ b/awx/conf/tests/__init__.py @@ -0,0 +1,2 @@ +# Copyright (c) 2017 Ansible, Inc. +# All Rights Reserved. diff --git a/awx/conf/tests/unit/test_registry.py b/awx/conf/tests/unit/test_registry.py new file mode 100644 index 0000000000..1516c72605 --- /dev/null +++ b/awx/conf/tests/unit/test_registry.py @@ -0,0 +1,302 @@ +# Copyright (c) 2017 Ansible, Inc. +# All Rights Reserved. + +from uuid import uuid4 + +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 _ +from rest_framework.fields import empty +import pytest + +from awx.conf import fields +from awx.conf.settings import SettingsWrapper +from awx.conf.registry import SettingsRegistry + + +@pytest.fixture() +def reg(request): + cache = LocMemCache(str(uuid4()), {}) # make a new random cache each time + settings = LazySettings() + registry = SettingsRegistry(settings) + defaults = request.node.get_marker('defaults') + if defaults: + settings.configure(**defaults.kwargs) + settings._wrapped = SettingsWrapper(settings._wrapped, + cache, + registry) + return registry + + +def test_simple_setting_registration(reg): + assert reg.get_registered_settings() == [] + reg.register( + 'AWX_SOME_SETTING_ENABLED', + field_class=fields.BooleanField, + category=_('System'), + category_slug='system', + ) + assert reg.get_registered_settings() == ['AWX_SOME_SETTING_ENABLED'] + + +def test_simple_setting_unregistration(reg): + reg.register( + 'AWX_SOME_SETTING_ENABLED', + field_class=fields.BooleanField, + category=_('System'), + category_slug='system', + ) + assert reg.get_registered_settings() == ['AWX_SOME_SETTING_ENABLED'] + + reg.unregister('AWX_SOME_SETTING_ENABLED') + assert reg.get_registered_settings() == [] + + +def test_duplicate_setting_registration(reg): + "ensure that settings cannot be registered twice." + with pytest.raises(ImproperlyConfigured): + for i in range(2): + reg.register( + 'AWX_SOME_SETTING_ENABLED', + field_class=fields.BooleanField, + category=_('System'), + category_slug='system', + ) + + +def test_field_class_required_for_registration(reg): + "settings must specify a field class to register" + with pytest.raises(ImproperlyConfigured): + reg.register('AWX_SOME_SETTING_ENABLED') + + +def test_get_registered_settings_by_slug(reg): + reg.register( + 'AWX_SOME_SETTING_ENABLED', + field_class=fields.BooleanField, + category=_('System'), + category_slug='system', + ) + assert reg.get_registered_settings(category_slug='system') == [ + 'AWX_SOME_SETTING_ENABLED' + ] + assert reg.get_registered_settings(category_slug='other') == [] + + +def test_get_registered_read_only_settings(reg): + reg.register( + 'AWX_SOME_SETTING_ENABLED', + field_class=fields.BooleanField, + category=_('System'), + category_slug='system' + ) + reg.register( + 'AWX_SOME_READ_ONLY', + field_class=fields.BooleanField, + category=_('System'), + category_slug='system', + read_only=True + ) + assert reg.get_registered_settings(read_only=True) ==[ + 'AWX_SOME_READ_ONLY' + ] + assert reg.get_registered_settings(read_only=False) == [ + 'AWX_SOME_SETTING_ENABLED' + ] + assert reg.get_registered_settings() == [ + 'AWX_SOME_SETTING_ENABLED', + 'AWX_SOME_READ_ONLY' + ] + + +def test_get_registered_settings_with_required_features(reg): + reg.register( + 'AWX_SOME_SETTING_ENABLED', + field_class=fields.BooleanField, + category=_('System'), + category_slug='system', + feature_required='superpowers', + ) + assert reg.get_registered_settings(features_enabled=[]) == [] + assert reg.get_registered_settings(features_enabled=['superpowers']) == [ + 'AWX_SOME_SETTING_ENABLED' + ] + + +def test_get_dependent_settings(reg): + reg.register( + 'AWX_SOME_SETTING_ENABLED', + field_class=fields.BooleanField, + category=_('System'), + category_slug='system' + ) + reg.register( + 'AWX_SOME_DEPENDENT_SETTING', + field_class=fields.BooleanField, + category=_('System'), + category_slug='system', + depends_on=['AWX_SOME_SETTING_ENABLED'] + ) + assert reg.get_dependent_settings('AWX_SOME_SETTING_ENABLED') == set([ + 'AWX_SOME_DEPENDENT_SETTING' + ]) + + +def test_get_registered_categories(reg): + reg.register( + 'AWX_SOME_SETTING_ENABLED', + field_class=fields.BooleanField, + category=_('System'), + category_slug='system' + ) + reg.register( + 'AWX_SOME_OTHER_SETTING_ENABLED', + field_class=fields.BooleanField, + category=_('OtherSystem'), + category_slug='other-system' + ) + assert reg.get_registered_categories() == { + 'all': _('All'), + 'changed': _('Changed'), + 'system': _('System'), + 'other-system': _('OtherSystem'), + } + + +def test_get_registered_categories_with_required_features(reg): + reg.register( + 'AWX_SOME_SETTING_ENABLED', + field_class=fields.BooleanField, + category=_('System'), + category_slug='system', + feature_required='superpowers' + ) + reg.register( + 'AWX_SOME_OTHER_SETTING_ENABLED', + field_class=fields.BooleanField, + category=_('OtherSystem'), + category_slug='other-system', + feature_required='sortapowers' + ) + assert reg.get_registered_categories(features_enabled=[]) == { + 'all': _('All'), + 'changed': _('Changed'), + } + assert reg.get_registered_categories(features_enabled=['superpowers']) == { + 'all': _('All'), + 'changed': _('Changed'), + 'system': _('System'), + } + assert reg.get_registered_categories(features_enabled=['sortapowers']) == { + 'all': _('All'), + 'changed': _('Changed'), + 'other-system': _('OtherSystem'), + } + assert reg.get_registered_categories( + features_enabled=['superpowers', 'sortapowers'] + ) == { + 'all': _('All'), + 'changed': _('Changed'), + 'system': _('System'), + 'other-system': _('OtherSystem'), + } + + +def test_is_setting_encrypted(reg): + reg.register( + 'AWX_SOME_SETTING_ENABLED', + field_class=fields.CharField, + category=_('System'), + category_slug='system' + ) + reg.register( + 'AWX_SOME_ENCRYPTED_SETTING', + field_class=fields.CharField, + category=_('System'), + category_slug='system', + encrypted=True + ) + assert reg.is_setting_encrypted('AWX_SOME_SETTING_ENABLED') is False + assert reg.is_setting_encrypted('AWX_SOME_ENCRYPTED_SETTING') is True + + +def test_simple_field(reg): + reg.register( + 'AWX_SOME_SETTING', + field_class=fields.CharField, + category=_('System'), + category_slug='system', + placeholder='Example Value', + feature_required='superpowers' + ) + + field = reg.get_setting_field('AWX_SOME_SETTING') + assert isinstance(field, fields.CharField) + assert field.category == _('System') + assert field.category_slug == 'system' + assert field.default is empty + assert field.placeholder == 'Example Value' + assert field.feature_required == 'superpowers' + + +def test_field_with_custom_attribute(reg): + reg.register( + 'AWX_SOME_SETTING_ENABLED', + field_class=fields.BooleanField, + category_slug='system', + ) + + field = reg.get_setting_field('AWX_SOME_SETTING_ENABLED', + category_slug='other-system') + assert field.category_slug == 'other-system' + + +def test_field_with_custom_mixin(reg): + class GreatMixin(object): + + def is_great(self): + return True + + reg.register( + 'AWX_SOME_SETTING_ENABLED', + field_class=fields.BooleanField, + category_slug='system', + ) + + field = reg.get_setting_field('AWX_SOME_SETTING_ENABLED', + mixin_class=GreatMixin) + assert isinstance(field, fields.BooleanField) + assert isinstance(field, GreatMixin) + assert field.is_great() is True + + +@pytest.mark.defaults(AWX_SOME_SETTING='DEFAULT') +def test_default_value_from_settings(reg): + reg.register( + 'AWX_SOME_SETTING', + field_class=fields.CharField, + category=_('System'), + category_slug='system', + ) + + field = reg.get_setting_field('AWX_SOME_SETTING') + assert field.default == 'DEFAULT' + + +@pytest.mark.defaults(AWX_SOME_SETTING='DEFAULT') +def test_default_value_from_settings_with_custom_representation(reg): + class LowercaseCharField(fields.CharField): + + def to_representation(self, value): + return value.lower() + + reg.register( + 'AWX_SOME_SETTING', + field_class=LowercaseCharField, + category=_('System'), + category_slug='system', + ) + + field = reg.get_setting_field('AWX_SOME_SETTING') + assert field.default == 'default' diff --git a/awx/conf/tests/unit/test_settings.py b/awx/conf/tests/unit/test_settings.py new file mode 100644 index 0000000000..feb42df803 --- /dev/null +++ b/awx/conf/tests/unit/test_settings.py @@ -0,0 +1,256 @@ +# Copyright (c) 2017 Ansible, Inc. +# All Rights Reserved. + +from contextlib import contextmanager +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 _ +from rest_framework import fields +import pytest + +from awx.conf import models +from awx.conf.settings import SettingsWrapper, SETTING_CACHE_NOTSET +from awx.conf.registry import SettingsRegistry + + +@contextmanager +def apply_patches(_patches): + [p.start() for p in _patches] + yield + [p.stop() for p in _patches] + + +@pytest.fixture() +def settings(request): + cache = LocMemCache(str(uuid4()), {}) # make a new random cache each time + settings = LazySettings() + registry = SettingsRegistry(settings) + + # @pytest.mark.readonly can be used to mark specific setting values as + # "read-only". 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/.py) + readonly_marker = request.node.get_marker('readonly') + defaults = readonly_marker.kwargs if readonly_marker else {} + defaults['DEFAULTS_SNAPSHOT'] = {} + settings.configure(**defaults) + settings._wrapped = SettingsWrapper(settings._wrapped, + cache, + registry) + return settings + + +@pytest.mark.readonly(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 + + +@pytest.mark.readonly(AWX_SOME_SETTING='DEFAULT') +def test_read_only_setting(settings): + settings.registry.register( + 'AWX_SOME_SETTING', + field_class=fields.CharField, + category=_('System'), + category_slug='system' + ) + 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.readonly(AWX_SOME_SETTING='DEFAULT') +def test_read_only_setting_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.readonly(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.readonly(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' + + +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' + ) + + settings_to_cache = [ + mocker.Mock(**{'order_by.return_value': []}), + mocker.Mock(**{'order_by.return_value.first.return_value': None}) + ] + with mocker.patch('awx.conf.models.Setting.objects.filter', side_effect=settings_to_cache): + 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' + ) + + settings_to_cache = [ + mocker.Mock(**{'order_by.return_value': [ + mocker.Mock(key='AWX_SOME_SETTING', value='FROM_DB') + ]}), + ] + with mocker.patch('awx.conf.models.Setting.objects.filter', side_effect=settings_to_cache): + assert settings.AWX_SOME_SETTING == 'FROM_DB' + assert settings.cache.get('AWX_SOME_SETTING') == 'FROM_DB' + + +@pytest.mark.readonly(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.readonly(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' diff --git a/tox.ini b/tox.ini index 14726b2bd0..54effa178f 100644 --- a/tox.ini +++ b/tox.ini @@ -48,7 +48,7 @@ commands = python setup.py develop # coverage run --help # coverage run -p --source awx/main/tests -m pytest {posargs} - py.test awx/main/tests {posargs:-k 'not old'} + py.test awx/main/tests awx/conf/tests {posargs:-k 'not old'} [testenv:ui] deps =