Python django.test.utils 模块，CaptureQueriesContext() 实例源码

def test_caching_enabled(admin_client, router, destination):
    # Only sqlite3 logs a begin query within transaction
    atomic_queries = 1 if connection.vendor == 'sqlite' else 0

    with override_settings(ROUTING_CACHE=True):
        with CaptureQueriesContext(connection=connection) as c:
            response = admin_client.get(router.source, follow=True)
            assert response.status_code == 200
            assert_string_equal(response.content, 'destination')
            first = len(c)
            assert first - atomic_queries == 5
            response = admin_client.get(router.source, follow=True)
            assert response.status_code == 200
            assert_string_equal(response.content, 'destination')
            # Should only query for user and session because of condition
            assert len(c) - first - atomic_queries == 2

        router.delete()

        with CaptureQueriesContext(connection=connection) as c:
            response = admin_client.get(router.source, follow=True)
            assert response.status_code == 200
            assert_string_equal(response.content, 'home')
            # Only the router query
            assert len(c) == 1

def test_validate_bearer_token_should_not_reach_db_when_cached(
        self,
        access_token,
        validator,
        http_request,
        scopes
    ):
        db_result = self._warm_up_cache(
            validator,
            access_token.token,
            scopes,
            http_request
        )

        with CaptureQueriesContext(connection) as context:
            cached_result = validator.validate_bearer_token(
                access_token.token,
                scopes,
                http_request
            )

        assert len(context.captured_queries) == 0
        assert db_result == cached_result

def assertSQLQueries(self, model=None, check_other_fields=True):
        """
        Asserts that the SQL from the queries don't mention
        the readonly field. SELECTS are authorized, though.
        model allow you to specify the model for which readonly
        fields will be checked
        """
        model = model or Car
        readonly_fields = model.ReadonlyMeta.readonly
        with CaptureQueriesContext(connection=connection) as capture:
            yield capture

        unchecked_queries = frozenset("SELECT SAVEPOINT RELEASE".split())

        for query in self._filter_queries(capture, exclude=unchecked_queries):

            for field in readonly_fields:
                self.assertNotIn(field, query['sql'])

            if check_other_fields:
                fields = {field.name
                          for field in model._meta.fields
                          if not field.primary_key}
                read_write_fields = fields - frozenset(readonly_fields)
                for field in read_write_fields:
                    self.assertIn(field, query['sql'])

def test_lbheartbeat_makes_no_db_queries(dockerflow_middleware, rf):
    queries = CaptureQueriesContext(connection)
    request = rf.get('/__lbheartbeat__')
    with queries:
        response = dockerflow_middleware.process_request(request)
        assert response.status_code == 200
    assert len(queries) == 0

def test_validate_bearer_token_should_get_cache_expiration_from_token(
        self,
        access_token,
        validator,
        scopes,
        http_request
    ):
        expires = timezone.now() - timedelta(seconds=5)
        access_token.expires = expires
        access_token.save()

        self._warm_up_cache(
            validator,
            access_token.token,
            scopes,
            http_request
        )

        with CaptureQueriesContext(connection) as context:
            validator.validate_bearer_token(
                access_token.token,
                scopes,
                http_request
            )

        assert len(context.captured_queries) == 1

def test_query_count(client):
    policy = mommy.make(Policy, omb_policy_id='M-O-A-R')
    root = random_doc(20, save=True, policy=policy, text='placeholder')

    # select 3 nodes to have external links
    for node in random.sample(list(root.walk()), 3):
        node.externallinks.create(start=0, end=1, href='http://example.com/')

    # select 3 nodes to have inline requirements
    for node in random.sample(list(root.walk()), 3):
        node.inlinerequirements.create(
            start=1, end=2, requirement=mommy.make(Requirement))

    # select 3 nodes to add footnote citations
    citing_nodes = random.sample(list(root.walk()), 3)
    footnotes = [citing.add_child('footnote') for citing in citing_nodes]
    root.nested_set_renumber(bulk_create=False)
    for node in root.walk():
        node.model.save()
    for citing, footnote in zip(citing_nodes, footnotes):
        citing.footnotecitations.create(
            start=2, end=3, footnote_node=footnote.model)
    # pytest will alter the connection, so we only want to load it within this
    # test
    from django.db import connection
    with CaptureQueriesContext(connection) as capture:
        client.get("/M-O-A-R")
        # Query 01: Lookup the policy
        # 02: Lookup the root docnode
        # 03: fetch footnote citations _and_ referenced node for the root
        # 04: fetch external links for the root
        # 05: fetch inline requirements _and_ referenced req for root
        # 06: fetch cite elements for the root
        # 06: fetch nodes for table of contents
        # 08: fetch child nodes
        # 09: fetch footnote citations _and_ referenced node for child nodes
        # 10: fetch external links for child nodes
        # 11: fetch inline requirements _and_ referenced req for child nodes
        # 12: fetch cite elements for child nodes
        assert len(capture) == 12