add unit tests for awx.conf

This commit is contained in:
Ryan Petrello 2017-01-30 09:36:14 -05:00
parent 4fcc74012d
commit 836ca21b7e
7 changed files with 618 additions and 39 deletions

View File

@ -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)

View File

@ -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:

View File

@ -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:

View File

@ -0,0 +1,2 @@
# Copyright (c) 2017 Ansible, Inc.
# All Rights Reserved.

View File

@ -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'

View File

@ -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/<something>.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'

View File

@ -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 =