# Path: django_performance_testing/queries.py
# class QueryCollector(BaseCollector):
# 
#     type_name = 'queries'
# 
#     def __enter__(self):
#         self.queries = []
#         self.nr_of_queries_when_entering = len(connection.queries)
#         self.orig_force_debug_cursor = connection.force_debug_cursor
#         connection.force_debug_cursor = True
#         before_clearing_queries_log.connect(
#             self.queries_about_to_be_reset_handler)
#         return self
# 
#     def queries_about_to_be_reset_handler(self,
#                                           signal, sender, queries, **kwargs):
#         self.store_queries()
#         self.nr_of_queries_when_entering = 0
# 
#     def before_exit(self):
#         before_clearing_queries_log.disconnect(
#             self.queries_about_to_be_reset_handler)
#         connection.force_debug_cursor = self.orig_force_debug_cursor
#         self.store_queries()
# 
#     def get_results_to_send(self):
#         by_type = dict(read=[], write=[], other=[])
#         for result in self.queries:
#             tp = classify_query(result['sql'])
#             by_type[tp].append(result)
#         by_type['total'] = self.queries
#         return list(
#             QueryCountResult(name=tp, queries=q)
#             for (tp, q) in six.iteritems(by_type))
# 
#     @classmethod
#     def get_sample_results(cls):
#         read = """QUERY - 'SELECT "auth_user"."id" FROM "auth_group"'""" \
#             """- PARAMS = ()"""
#         write = """QUERY - 'UPDATE "auth_group" SET "name" = %s'""" \
#             """- PARAMS = ('bar',)"""
#         other = """QUERY - 'BEGIN TRANSACTION' - PARAMS = ()"""
# 
#         def to_query(sql):
#             return {'sql': sql, 'time': '%.3f' % random.random()}
# 
#         def to_single_result(*sqls):
#             qc = cls()
#             qc.queries = [to_query(sql) for sql in sqls]
#             return qc.get_results_to_send()
# 
#         return [
#             to_single_result(*sqls)
#             for sqls in [
#                 [read], [read, write], [read, read], [write, write],
#                 [other, other], [read, write, other]
#             ]
#         ]
# 
#     def store_queries(self):
#         self.queries += connection.queries[self.nr_of_queries_when_entering:]
# 
# Path: django_performance_testing/signals.py



import pytest
from django.contrib.auth.models import Group
from django.core import signals
from django.db import connection, reset_queries
from django.db.transaction import atomic
from django_performance_testing.queries import QueryCollector
from django_performance_testing.signals import results_collected
from testapp.test_helpers import capture_result_collected


@pytest.mark.parametrize(
    'code,total_lo_limit,write_lo_limit,read_lo_limit',
    [
        (lambda: Group.objects.create(name='foo'), 1, 1, 0),
        (lambda: list(Group.objects.all()), 1, 0, 1),
        (lambda: Group.objects.update(name='bar'), 1, 1, 0),
        (lambda: Group.objects.all().delete(), 1, 1, 0),
    ], ids=['insert', 'select', 'update', 'delete'])
def test_captures_and_classifies_each_query_type(
        db, code, total_lo_limit, write_lo_limit, read_lo_limit):

    # 'coz of new, 'smart' delete need an item
    Group.objects.create(name='random')
    with capture_result_collected() as captured:

# Path: django_performance_testing/queries.py
# class QueryCollector(BaseCollector):
# 
#     type_name = 'queries'
# 
#     def __enter__(self):
#         self.queries = []
#         self.nr_of_queries_when_entering = len(connection.queries)
#         self.orig_force_debug_cursor = connection.force_debug_cursor
#         connection.force_debug_cursor = True
#         before_clearing_queries_log.connect(
#             self.queries_about_to_be_reset_handler)
#         return self
# 
#     def queries_about_to_be_reset_handler(self,
#                                           signal, sender, queries, **kwargs):
#         self.store_queries()
#         self.nr_of_queries_when_entering = 0
# 
#     def before_exit(self):
#         before_clearing_queries_log.disconnect(
#             self.queries_about_to_be_reset_handler)
#         connection.force_debug_cursor = self.orig_force_debug_cursor
#         self.store_queries()
# 
#     def get_results_to_send(self):
#         by_type = dict(read=[], write=[], other=[])
#         for result in self.queries:
#             tp = classify_query(result['sql'])
#             by_type[tp].append(result)
#         by_type['total'] = self.queries
#         return list(
#             QueryCountResult(name=tp, queries=q)
#             for (tp, q) in six.iteritems(by_type))
# 
#     @classmethod
#     def get_sample_results(cls):
#         read = """QUERY - 'SELECT "auth_user"."id" FROM "auth_group"'""" \
#             """- PARAMS = ()"""
#         write = """QUERY - 'UPDATE "auth_group" SET "name" = %s'""" \
#             """- PARAMS = ('bar',)"""
#         other = """QUERY - 'BEGIN TRANSACTION' - PARAMS = ()"""
# 
#         def to_query(sql):
#             return {'sql': sql, 'time': '%.3f' % random.random()}
# 
#         def to_single_result(*sqls):
#             qc = cls()
#             qc.queries = [to_query(sql) for sql in sqls]
#             return qc.get_results_to_send()
# 
#         return [
#             to_single_result(*sqls)
#             for sqls in [
#                 [read], [read, write], [read, read], [write, write],
#                 [other, other], [read, write, other]
#             ]
#         ]
# 
#     def store_queries(self):
#         self.queries += connection.queries[self.nr_of_queries_when_entering:]
# 
# Path: django_performance_testing/signals.py



# Path: tests/testapp/tests/test_query_collector.py
import pytest
from django.contrib.auth.models import Group
from django.core import signals
from django.db import connection, reset_queries
from django.db.transaction import atomic
from django_performance_testing.queries import QueryCollector
from django_performance_testing.signals import results_collected
from testapp.test_helpers import capture_result_collected



@pytest.mark.parametrize(
    'code,total_lo_limit,write_lo_limit,read_lo_limit',
    [
        (lambda: Group.objects.create(name='foo'), 1, 1, 0),
        (lambda: list(Group.objects.all()), 1, 0, 1),
        (lambda: Group.objects.update(name='bar'), 1, 1, 0),
        (lambda: Group.objects.all().delete(), 1, 1, 0),
    ], ids=['insert', 'select', 'update', 'delete'])
def test_captures_and_classifies_each_query_type(
        db, code, total_lo_limit, write_lo_limit, read_lo_limit):

    # 'coz of new, 'smart' delete need an item
    Group.objects.create(name='random')
    with capture_result_collected() as captured:

        with QueryCollector():