<commit_before>import datetime
import os
import unittest
from unittest.mock import patch

import django
from django.contrib.auth.models import User
from django.db import connection
from django.db.models import Count
from django.db.utils import DatabaseError
from django.shortcuts import render
from django.test.utils import override_settings

import debug_toolbar.panels.sql.tracking as sql_tracking
from debug_toolbar import settings as dt_settings

from ..base import BaseTestCase

try:
    from psycopg2._json import Json as PostgresJson
except ImportError:
    PostgresJson = None

if connection.vendor == "postgresql":
    from ..models import PostgresJSON as PostgresJSONModel
else:
    PostgresJSONModel = None


class SQLPanelTestCase(BaseTestCase):
    panel_id = "SQLPanel"

    def test_disabled(self):
        config = {"DISABLE_PANELS": {"debug_toolbar.panels.sql.SQLPanel"}}
        self.assertTrue(self.panel.enabled)
        with self.settings(DEBUG_TOOLBAR_CONFIG=config):
            self.assertFalse(self.panel.enabled)

    def test_recording(self):
        self.assertEqual(len(self.panel._queries), 0)

        list(User.objects.all())

        # ensure query was logged
        self.assertEqual(len(self.panel._queries), 1)
        query = self.panel._queries[0]
        self.assertEqual(query[0], "default")
        self.assertTrue("sql" in query[1])
        self.assertTrue("duration" in query[1])
        self.assertTrue("stacktrace" in query[1])

        # ensure the stacktrace is populated
        self.assertTrue(len(query[1]["stacktrace"]) > 0)

    @unittest.skipUnless(
        connection.vendor == "postgresql", "Test valid only on PostgreSQL"
    )
    def test_recording_chunked_cursor(self):
        self.assertEqual(len(self.panel._queries), 0)

        list(User.objects.all().iterator())

        # ensure query was logged
        self.assertEqual(len(self.panel._queries), 1)

    @patch("debug_toolbar.panels.sql.tracking.state", wraps=sql_tracking.state)
    def test_cursor_wrapper_singleton(self, mock_state):
        list(User.objects.all())

        # ensure that cursor wrapping is applied only once
        self.assertEqual(mock_state.Wrapper.call_count, 1)

    @patch("debug_toolbar.panels.sql.tracking.state", wraps=sql_tracking.state)
    def test_chunked_cursor_wrapper_singleton(self, mock_state):
        list(User.objects.all().iterator())

        # ensure that cursor wrapping is applied only once
        self.assertEqual(mock_state.Wrapper.call_count, 1)

    def test_generate_server_timing(self):
        self.assertEqual(len(self.panel._queries), 0)

        list(User.objects.all())

        response = self.panel.process_request(self.request)
        self.panel.generate_stats(self.request, response)
        self.panel.generate_server_timing(self.request, response)

        # ensure query was logged
        self.assertEqual(len(self.panel._queries), 1)
        query = self.panel._queries[0]

        expected_data = {
            "sql_time": {"title": "SQL 1 queries", "value": query[1]["duration"]}
        }

        self.assertEqual(self.panel.get_server_timing_stats(), expected_data)

    def test_non_ascii_query(self):
        self.assertEqual(len(self.panel._queries), 0)

        # non-ASCII text query
        list(User.objects.extra(where=["username = 'apéro'"]))
        self.assertEqual(len(self.panel._queries), 1)

        # non-ASCII text parameters
        list(User.objects.filter(username="thé"))
        self.assertEqual(len(self.panel._queries), 2)

        # non-ASCII bytes parameters
        list(User.objects.filter(username="café".encode()))
        self.assertEqual(len(self.panel._queries), 3)

        response = self.panel.process_request(self.request)
        self.panel.generate_stats(self.request, response)

        # ensure the panel renders correctly
        self.assertIn("café", self.panel.content)

    def test_param_conversion(self):
        self.assertEqual(len(self.panel._queries), 0)

        list(
            User.objects.filter(first_name="Foo")
            .filter(is_staff=True)
            .filter(is_superuser=False)
        )
        list(
            User.objects.annotate(group_count=Count("groups__id"))
            .filter(group_count__lt=10)
            .filter(group_count__gt=1)
        )
        list(User.objects.filter(date_joined=datetime.datetime(2017, 12, 22, 16, 7, 1)))

        response = self.panel.process_request(self.request)
        self.panel.generate_stats(self.request, response)

        # ensure query was logged
        self.assertEqual(len(self.panel._queries), 3)

        if django.VERSION >= (3, 1):
            self.assertEqual(
                tuple([q[1]["params"] for q in self.panel._queries]),
                ('["Foo"]', "[10, 1]", '["2017-12-22 16:07:01"]'),
            )
        else:
            self.assertEqual(
                tuple([q[1]["params"] for q in self.panel._queries]),
                ('["Foo", true, false]', "[10, 1]", '["2017-12-22 16:07:01"]'),
            )<commit_msg>Fix test_param_conversion for Django 4.1 and mysql<commit_after>import datetime
import os
import unittest
from unittest.mock import patch

import django
from django.contrib.auth.models import User
from django.db import connection
from django.db.models import Count
from django.db.utils import DatabaseError
from django.shortcuts import render
from django.test.utils import override_settings

import debug_toolbar.panels.sql.tracking as sql_tracking
from debug_toolbar import settings as dt_settings

from ..base import BaseTestCase

try:
    from psycopg2._json import Json as PostgresJson
except ImportError:
    PostgresJson = None

if connection.vendor == "postgresql":
    from ..models import PostgresJSON as PostgresJSONModel
else:
    PostgresJSONModel = None


class SQLPanelTestCase(BaseTestCase):
    panel_id = "SQLPanel"

    def test_disabled(self):
        config = {"DISABLE_PANELS": {"debug_toolbar.panels.sql.SQLPanel"}}
        self.assertTrue(self.panel.enabled)
        with self.settings(DEBUG_TOOLBAR_CONFIG=config):
            self.assertFalse(self.panel.enabled)

    def test_recording(self):
        self.assertEqual(len(self.panel._queries), 0)

        list(User.objects.all())

        # ensure query was logged
        self.assertEqual(len(self.panel._queries), 1)
        query = self.panel._queries[0]
        self.assertEqual(query[0], "default")
        self.assertTrue("sql" in query[1])
        self.assertTrue("duration" in query[1])
        self.assertTrue("stacktrace" in query[1])

        # ensure the stacktrace is populated
        self.assertTrue(len(query[1]["stacktrace"]) > 0)

    @unittest.skipUnless(
        connection.vendor == "postgresql", "Test valid only on PostgreSQL"
    )
    def test_recording_chunked_cursor(self):
        self.assertEqual(len(self.panel._queries), 0)

        list(User.objects.all().iterator())

        # ensure query was logged
        self.assertEqual(len(self.panel._queries), 1)

    @patch("debug_toolbar.panels.sql.tracking.state", wraps=sql_tracking.state)
    def test_cursor_wrapper_singleton(self, mock_state):
        list(User.objects.all())

        # ensure that cursor wrapping is applied only once
        self.assertEqual(mock_state.Wrapper.call_count, 1)

    @patch("debug_toolbar.panels.sql.tracking.state", wraps=sql_tracking.state)
    def test_chunked_cursor_wrapper_singleton(self, mock_state):
        list(User.objects.all().iterator())

        # ensure that cursor wrapping is applied only once
        self.assertEqual(mock_state.Wrapper.call_count, 1)

    def test_generate_server_timing(self):
        self.assertEqual(len(self.panel._queries), 0)

        list(User.objects.all())

        response = self.panel.process_request(self.request)
        self.panel.generate_stats(self.request, response)
        self.panel.generate_server_timing(self.request, response)

        # ensure query was logged
        self.assertEqual(len(self.panel._queries), 1)
        query = self.panel._queries[0]

        expected_data = {
            "sql_time": {"title": "SQL 1 queries", "value": query[1]["duration"]}
        }

        self.assertEqual(self.panel.get_server_timing_stats(), expected_data)

    def test_non_ascii_query(self):
        self.assertEqual(len(self.panel._queries), 0)

        # non-ASCII text query
        list(User.objects.extra(where=["username = 'apéro'"]))
        self.assertEqual(len(self.panel._queries), 1)

        # non-ASCII text parameters
        list(User.objects.filter(username="thé"))
        self.assertEqual(len(self.panel._queries), 2)

        # non-ASCII bytes parameters
        list(User.objects.filter(username="café".encode()))
        self.assertEqual(len(self.panel._queries), 3)

        response = self.panel.process_request(self.request)
        self.panel.generate_stats(self.request, response)

        # ensure the panel renders correctly
        self.assertIn("café", self.panel.content)

    def test_param_conversion(self):
        self.assertEqual(len(self.panel._queries), 0)

        list(
            User.objects.filter(first_name="Foo")
            .filter(is_staff=True)
            .filter(is_superuser=False)
        )
        list(
            User.objects.annotate(group_count=Count("groups__id"))
            .filter(group_count__lt=10)
            .filter(group_count__gt=1)
        )
        list(User.objects.filter(date_joined=datetime.datetime(2017, 12, 22, 16, 7, 1)))

        response = self.panel.process_request(self.request)
        self.panel.generate_stats(self.request, response)

        # ensure query was logged
        self.assertEqual(len(self.panel._queries), 3)

        if (
            django.VERSION >= (3, 1)
            # Django 4.1 started passing true/false back for boolean
            # comparisons in MySQL.
            and not (django.VERSION >= (4, 1) and connection.vendor == "mysql")
        ):
            self.assertEqual(
                tuple([q[1]["params"] for q in self.panel._queries]),
                ('["Foo"]', "[10, 1]", '["2017-12-22 16:07:01"]'),
            )
        else:
            self.assertEqual(
                tuple([q[1]["params"] for q in self.panel._queries]),
                ('["Foo", true, false]', "[10, 1]", '["2017-12-22 16:07:01"]'),
            )
{'fraction_non_alphanumeric': 0.11792543944369326, 'fraction_numerical': 0.015646127100637435, 'mean_word_length': 2.8638059701492535, 'pattern_counts': {'":': 10, '<': 3, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4299722', 'n_tokens_mistral': 3205, 'n_tokens_neox': 2997, 'n_words': 601}