python类override_settings()的实例源码

def test_exam_data_auditor_enabled(auditor, missing_settings, mocker):
    """Test that audit_file() respected the enabled flag"""
    mocker.patch.multiple(auditor, configure=DEFAULT, upload_encrypted_file=DEFAULT)
    # auditor is disabled
    with override_settings(**{
        'EXAMS_AUDIT_ENABLED': False,
    }):
        auditor.audit_file('file', 'filetype')
    assert auditor.configure.call_count == 0
    # auditor is enabled
    with override_settings(**{
        'EXAMS_AUDIT_ENABLED': True,
    }):
        auditor.audit_file('file', 'filetype')
    assert auditor.configure.call_count == 1

def test_reencrypt_transcript_credentials_invalid_keys(self, mock_logger):
        """
        Test transcript credentials would not be re-encrypted if an decryption key is not provided with which
        data was encypted before.
        """
        # Verify fernet keys.
        self.assertEqual(settings.FERNET_KEYS, OLD_FERNET_KEYS_LIST)

        # Verify we are able to access the record.
        self.verify_access_credentials()

        # Modify key set so that old key is not presnet in the key list. Note that now we are not providing
        # a decryption key for data to be decrypted.
        new_keys_set = ['new-fernet-key']

        with override_settings(FERNET_KEYS=new_keys_set):
            self.assertEqual(settings.FERNET_KEYS, new_keys_set)
            # Run re-encryption process.
            call_command('re_encrypt_transcript_credentials')

            # Verify logging.
            mock_logger.info.assert_called_with('[Transcript credentials re-encryption] Process started.')
            mock_logger.exception.assert_called_with(
                '[Transcript credentials re-encryption] No valid fernet key present to decrypt. Process halted.'
            )

            # Verify we are not able to access the record, we should get an error due to decryption key not present.
            with self.assertRaises(InvalidToken):
                self.verify_access_credentials()

def custom_override_settings(**settings_kwargs):
    """
    Override the settings as override_settings from django.
    This decorator also reloads the settings.py module so the settings are changed as expected.
    """
    def _my_decorator(func):
        @override_settings(**settings_kwargs)
        def _decorator(func2, *args, **kwargs):
            reload(settings)
            return func(func2, *args, **kwargs)
        return wraps(func)(_decorator)
    return _my_decorator

def custom_override_settings(**settings_kwargs):
    """
    Override the settings as override_settings from django.
    This decorator also reloads the settings.py module so the settings are changed as expected.
    """
    def _my_decorator(func):
        @override_settings(**settings_kwargs)
        def _decorator(func2, *args, **kwargs):
            reload(settings)
            return func(func2, *args, **kwargs)
        return wraps(func)(_decorator)
    return _my_decorator

def test_init(self, mockldap3):
        # initialize and then check expected behavior against
        # mock ldap3
        LDAPSearch()

        test_servers = []
        for test_server in self.ldap_servers:
            mockldap3.Server.assert_any_call(test_server,
                get_info=mockldap3.ALL, use_ssl=True)

        # initialized servers are collected into server pool
        servers = [mockldap3.Server.return_value
                   for test_server in self.ldap_servers]
        mockldap3.ServerPool.assert_called_with(servers,
            mockldap3.ROUND_ROBIN, active=True, exhaust=5)

        # server pool is used for connection
        mockldap3.Connection.assert_called_with(mockldap3.ServerPool.return_value,
            auto_bind=True)

        with override_settings(PUCAS_LDAP={
            'SERVERS': self.ldap_servers,
            'BIND_DN': self.dn,
            'BIND_PASSWORD': self.password,
         }):
            LDAPSearch()
            # server pool is used for connection, now with password
            mockldap3.Connection.assert_called_with(mockldap3.ServerPool.return_value,
                auto_bind=True, user=self.dn, password=self.password)

        with pytest.raises(LDAPException):
            mockldap3.Connection.side_effect = LDAPException
            LDAPSearch()

def test_django_user_main_attribute_lookup(self):
        backend = Saml2Backend()

        self.assertEqual(backend.get_django_user_main_attribute_lookup(), '')

        with override_settings(
                SAML_DJANGO_USER_MAIN_ATTRIBUTE_LOOKUP='__iexact'):
            self.assertEqual(
                backend.get_django_user_main_attribute_lookup(),
                '__iexact')

def test_search(self):
        with open('tests/fixtures/twitter.json', 'r') as feed_file:
            page1 = json.loads("".join(feed_file.readlines()))
        with open('tests/fixtures/twitter.2.json', 'r') as feed_file:
            page2 = json.loads("".join(feed_file.readlines()))

        responses.add(responses.GET,
                      re.compile('(?!.*max_id=\d*)https?://api.twitter.com.*'),
                      json=page1, status=200)

        responses.add(responses.GET,
                      re.compile('(?=.*max_id=\d*)https?://api.twitter.com.*'),
                      json=page2, status=200)

        q = "release"
        cache_key = "{}:q-{}".format(self.cache_key, q)

        self.assertIsNone(cache.get(cache_key))

        # Ensure we set the SEARCH_MAX_HISTORY big enough for both twitter
        # pages to be included
        now = datetime.datetime.now(tzutc())
        last_post_date = TwitterFeedItem.get_post_date(page2[-1])
        delta = (now - last_post_date) + datetime.timedelta(seconds=10)
        with override_settings(WAGTAIL_SOCIALFEED_SEARCH_MAX_HISTORY=delta):
            stream = self.stream.get_items(config=self.feedconfig,
                                           query_string=q)

        self.assertIsNotNone(cache.get(cache_key))
        self.assertEqual(len(stream), 2)
        for s in stream:
            self.assertIn('release', s.text)

def test_search(self):
        with open('tests/fixtures/facebook.json', 'r') as feed_file:
            page1 = json.loads("".join(feed_file.readlines()))
        with open('tests/fixtures/facebook.2.json', 'r') as feed_file:
            page2 = json.loads("".join(feed_file.readlines()))

        responses.add(
            responses.GET,
            re.compile('(?!.*paging_token)https?://graph.facebook.com.*'),
            json=page1, status=200)

        responses.add(
            responses.GET,
            re.compile('(?=.*paging_token)https?://graph.facebook.com.*'),
            json=page2, status=200)

        q = "tutorials"
        cache_key = "{}:q-{}".format(self.cache_key, q)

        self.assertIsNone(cache.get(cache_key))
        # Ensure we set the SEARCH_MAX_HISTORY big enough for both facebook
        # pages to be included
        now = datetime.datetime.now(tzutc())
        last_post_date = FacebookFeedItem.get_post_date(page2['data'][-1])
        delta = (now - last_post_date) + datetime.timedelta(seconds=10)
        with override_settings(WAGTAIL_SOCIALFEED_SEARCH_MAX_HISTORY=delta):
            stream = self.stream.get_items(config=self.feedconfig,
                                           query_string=q)

        self.assertIsNotNone(cache.get(cache_key))
        self.assertEqual(len(stream), 2)
        for s in stream:
            self.assertIn('tutorials', s.text)

def test_proxy_router_disabled(client, admin_client, router, destination):
    with override_settings(ENABLE_PROXY_ROUTING=False):
        router.action = 'proxy'
        router.save()

        response = client.get(router.source)
        assert response.status_code == 200
        assert_string_equal(response.content, 'home')

        response = admin_client.get(router.source)
        assert response.status_code == 200
        assert_string_equal(response.content, 'home')

def test_proxy_router_enabled(client, admin_client, router, destination):
    with override_settings(ENABLE_PROXY_ROUTING=True):
        router.action = 'proxy'
        router.save()

        response = client.get(router.source)
        assert response.status_code == 200
        assert_string_equal(response.content, 'home')

        response = admin_client.get(router.source)
        assert response.status_code == 200
        assert_string_equal(response.content, 'destination')

def test_routing_disabled(admin_client, router, destination):
    with override_settings(ROUTING_ENABLED=False):
        response = admin_client.get(router.source, follow=True)
        assert response.status_code == 200
        assert_string_equal(response.content, 'home')

def test_token(self):
        with override_settings(SLACK_TOKEN='foo'):
            response = self.send_command('/foo', token='baz')
            self.assertContains(
                response, 'Invalid or missing slack token.', status_code=400)

def site_root(request, live_server):
    # django live_server always sets DEBUG to False. Override that for test.
    settings_context = override_settings(DEBUG=True)
    settings_context.__enter__()

    def fin():
        settings_context.__exit__(None, None, None)
    request.addfinalizer(fin)
    return live_server.url

def test_catalog_admin_url_template(self, catalog_api_url, expected_url):
        """
        Validate that `get_catalog_admin_url_template` utility functions
        returns catalog admin page url template.

        Arguments:
            catalog_api_url (str): course catalog api url coming from DDT data decorator.
            expected_url (str): django admin catalog details page url coming from DDT data decorator.
        """
        with override_settings(COURSE_CATALOG_API_URL=catalog_api_url):
            url = utils.get_catalog_admin_url_template()
            assert url == expected_url

def test_catalog_admin_url(self, catalog_id, catalog_api_url, expected_url):
        """
        Validate that `get_catalog_admin_url` utility functions returns catalog admin page url.

        Arguments:
            catalog_id (int): catalog id coming from DDT data decorator.
            catalog_api_url (str): course catalog api url coming from DDT data decorator.
            expected_url (str): django admin catalog details page url coming from DDT data decorator.
        """
        with override_settings(COURSE_CATALOG_API_URL=catalog_api_url):
            url = utils.get_catalog_admin_url(catalog_id)
            assert url == expected_url

def override_middleware(fun):
    classes = [
        'django.contrib.sessions.middleware.SessionMiddleware',
        'mozilla_django_oidc.middleware.RefreshIDToken',
    ]
    if DJANGO_VERSION >= (1, 10):
        return override_settings(MIDDLEWARE=classes)(fun)
    return override_settings(MIDDLEWARE_CLASSES=classes)(fun)

def start_cache_isolation(cls):
        """
        Start cache isolation by overriding the settings.CACHES and
        flushing the cache.
        """
        cache_settings = None
        if cls.CACHES is not None and cls.ENABLED_CACHES is not None:
            raise Exception(
                "Use either CACHES or ENABLED_CACHES, but not both"
            )

        if cls.CACHES is not None:
            cache_settings = cls.CACHES
        elif cls.ENABLED_CACHES is not None:
            cache_settings = {
                'default': {
                    'BACKEND': 'django.core.cache.backends.dummy.DummyCache',
                }
            }

            cache_settings.update({
                cache_name: {
                    'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
                    'LOCATION': cache_name,
                    'KEY_FUNCTION': 'util.memcache.safe_key',
                } for cache_name in cls.ENABLED_CACHES
            })

        if cache_settings is None:
            return

        cls.__old_settings.append(copy.deepcopy(settings.CACHES))
        override = override_settings(CACHES=cache_settings)
        override.__enter__()
        cls.__settings_overrides.append(override)

        assert settings.CACHES == cache_settings

        # Start with empty caches
        cls.clear_caches()

def setup(self):
        # use a context manager to ensure these settings are
        # only used here
        with override_settings(
                WAGTAILTRANS_SYNC_TREE=True,
                WAGTAILTRANS_LANGUAGES_PER_SITE=True):
            register_site_languages()(SiteLanguages)
            self.site = sites.SiteFactory()
            SiteLanguages.for_site(self.site)
            self.default_language = Language.objects.get(code='en')
            self.site.sitelanguages.default_language = self.default_language