# 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:]
# 
# class QueryBatchLimit(BaseLimit):
#     collector_cls = QueryCollector
# 
#     quantifier = 'many'
#     items_name = 'queries'
# 
# class QueryCountResult(NameValueResult):
# 
#     def __init__(self, queries, name):
#         self.queries = queries
#         super(QueryCountResult, self).__init__(
#             name=name, value=self.number_of_queries)
# 
#     @property
#     def number_of_queries(self):
#         return len(self.queries)
# 
# Path: django_performance_testing/core.py
# class LimitViolationError(RuntimeError):
# 
#     def __init__(self, limit_obj, result, context, tb=None):
#         self.limit_obj = limit_obj
#         self.result = result
#         self.context = context
#         self.tb = tb
#         super(LimitViolationError, self).__init__(self.error_msg)
# 
#     @property
#     def error_msg(self):
#         base = self.base_error_msg
#         ctx = self.context_repr
#         tb = self.tb_msg
#         if ctx:
#             ctx = ' {}'.format(ctx)
#         if tb:
#             tb = '\n{}'.format(tb)
#         return ''.join([base, ctx, tb])
# 
#     @property
#     def tb_msg(self):
#         if self.tb:
#             return self.tb
#         return ''
# 
#     @property
#     def context_repr(self):
#         if self.context:
#             return pprint.pformat(self.context)
#         return ''
# 
#     @property
#     def base_error_msg(self):
#         return 'Too {} ({}) {} {}{} (limit: {})'.format(
#             self.quantifier,
#             self.actual,
#             self.name,
#             self.items_name,
#             self.collector_text,
#             self.limit
#         )
# 
#     @property
#     def quantifier(self):
#         return self.limit_obj.quantifier
# 
#     @property
#     def items_name(self):
#         return self.limit_obj.items_name
# 
#     @property
#     def name(self):
#         return self.result.name
# 
#     @property
#     def collector_text(self):
#         if self.limit_obj.is_anonymous():
#             return ''
#         return ' (for {})'.format(self.limit_obj.collector_id)
# 
#     @property
#     def actual(self):
#         return str(self.result)
# 
#     @property
#     def limit(self):
#         return self.limit_obj.limit_for(self.result)
# 
#     def clone_with_more_info(self, orig_tb):
#         return LimitViolationError(
#             limit_obj=self.limit_obj, result=self.result,
#             context=self.context, tb=orig_tb)


import pytest
from django.contrib.auth.models import Group
from django_performance_testing.queries import \
    QueryCollector, QueryBatchLimit, QueryCountResult
from django_performance_testing.core import LimitViolationError
from testapp.test_helpers import override_current_context


def test_it_has_the_correct_collector():
    assert QueryBatchLimit.collector_cls == QueryCollector


def test_it_has_the_correct_attributes_for_limitviolationerror():
    assert QueryBatchLimit.quantifier == 'many'
    assert QueryBatchLimit.items_name == 'queries'


def test_integration_test_with_db(db):

# 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:]
# 
# class QueryBatchLimit(BaseLimit):
#     collector_cls = QueryCollector
# 
#     quantifier = 'many'
#     items_name = 'queries'
# 
# class QueryCountResult(NameValueResult):
# 
#     def __init__(self, queries, name):
#         self.queries = queries
#         super(QueryCountResult, self).__init__(
#             name=name, value=self.number_of_queries)
# 
#     @property
#     def number_of_queries(self):
#         return len(self.queries)
# 
# Path: django_performance_testing/core.py
# class LimitViolationError(RuntimeError):
# 
#     def __init__(self, limit_obj, result, context, tb=None):
#         self.limit_obj = limit_obj
#         self.result = result
#         self.context = context
#         self.tb = tb
#         super(LimitViolationError, self).__init__(self.error_msg)
# 
#     @property
#     def error_msg(self):
#         base = self.base_error_msg
#         ctx = self.context_repr
#         tb = self.tb_msg
#         if ctx:
#             ctx = ' {}'.format(ctx)
#         if tb:
#             tb = '\n{}'.format(tb)
#         return ''.join([base, ctx, tb])
# 
#     @property
#     def tb_msg(self):
#         if self.tb:
#             return self.tb
#         return ''
# 
#     @property
#     def context_repr(self):
#         if self.context:
#             return pprint.pformat(self.context)
#         return ''
# 
#     @property
#     def base_error_msg(self):
#         return 'Too {} ({}) {} {}{} (limit: {})'.format(
#             self.quantifier,
#             self.actual,
#             self.name,
#             self.items_name,
#             self.collector_text,
#             self.limit
#         )
# 
#     @property
#     def quantifier(self):
#         return self.limit_obj.quantifier
# 
#     @property
#     def items_name(self):
#         return self.limit_obj.items_name
# 
#     @property
#     def name(self):
#         return self.result.name
# 
#     @property
#     def collector_text(self):
#         if self.limit_obj.is_anonymous():
#             return ''
#         return ' (for {})'.format(self.limit_obj.collector_id)
# 
#     @property
#     def actual(self):
#         return str(self.result)
# 
#     @property
#     def limit(self):
#         return self.limit_obj.limit_for(self.result)
# 
#     def clone_with_more_info(self, orig_tb):
#         return LimitViolationError(
#             limit_obj=self.limit_obj, result=self.result,
#             context=self.context, tb=orig_tb)


# Path: tests/testapp/tests/test_querycount_limits.py
import pytest
from django.contrib.auth.models import Group
from django_performance_testing.queries import \
    QueryCollector, QueryBatchLimit, QueryCountResult
from django_performance_testing.core import LimitViolationError
from testapp.test_helpers import override_current_context



def test_it_has_the_correct_collector():
    assert QueryBatchLimit.collector_cls == QueryCollector


def test_it_has_the_correct_attributes_for_limitviolationerror():
    assert QueryBatchLimit.quantifier == 'many'
    assert QueryBatchLimit.items_name == 'queries'


def test_integration_test_with_db(db):

    with pytest.raises(LimitViolationError) as excinfo: