# 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'
# 
# Path: django_performance_testing/timing.py
# class TimeCollector(BaseCollector):
# 
#     type_name = 'time'
# 
#     def __enter__(self):
#         self.start = time()
#         return self
# 
#     def get_results_to_send(self):
#         return [NameValueResult(name='total', value=time() - self.start)]
# 
#     @classmethod
#     def get_sample_results(cls):
#         return [0.01, 1.00, 3.2]
# 
# class TimeLimit(BaseLimit):
#     collector_cls = TimeCollector
# 
#     quantifier = 'many'
#     items_name = 'elapsed seconds'


import os
import tempfile
import pytest
from django_performance_testing.queries import QueryCollector, QueryBatchLimit
from django_performance_testing.timing import TimeCollector, TimeLimit


@pytest.fixture(params=[QueryCollector, TimeCollector])
def collector_cls(request):
    return request.param



# 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'
# 
# Path: django_performance_testing/timing.py
# class TimeCollector(BaseCollector):
# 
#     type_name = 'time'
# 
#     def __enter__(self):
#         self.start = time()
#         return self
# 
#     def get_results_to_send(self):
#         return [NameValueResult(name='total', value=time() - self.start)]
# 
#     @classmethod
#     def get_sample_results(cls):
#         return [0.01, 1.00, 3.2]
# 
# class TimeLimit(BaseLimit):
#     collector_cls = TimeCollector
# 
#     quantifier = 'many'
#     items_name = 'elapsed seconds'


# Path: tests/testapp/tests/conftest.py
import os
import tempfile
import pytest
from django_performance_testing.queries import QueryCollector, QueryBatchLimit
from django_performance_testing.timing import TimeCollector, TimeLimit



@pytest.fixture(params=[QueryCollector, TimeCollector])
def collector_cls(request):
    return request.param



@pytest.fixture(params=[QueryBatchLimit, TimeLimit])