<commit_before># -*- coding: utf-8 -*-
# Unit and doctests for specific database backends.
from __future__ import unicode_literals

import copy
import datetime
from decimal import Decimal
import re
import threading
import unittest

from django.conf import settings
from django.core.management.color import no_style
from django.db import (connection, connections, DEFAULT_DB_ALIAS,
    DatabaseError, IntegrityError, transaction)
from django.db.backends.signals import connection_created
from django.db.backends.sqlite3.base import DatabaseOperations
from django.db.backends.postgresql_psycopg2 import version as pg_version
from django.db.backends.utils import format_number, CursorWrapper
from django.db.models import Sum, Avg, Variance, StdDev
from django.db.models.fields import (AutoField, DateField, DateTimeField,
    DecimalField, IntegerField, TimeField)
from django.db.utils import ConnectionHandler
from django.test import (TestCase, TransactionTestCase, override_settings,
    skipUnlessDBFeature, skipIfDBFeature)
from django.test.utils import str_prefix
from django.utils import six
from django.utils.six.moves import xrange

from . import models


class DummyBackendTest(TestCase):
    def test_no_databases(self):
        """
        Test that empty DATABASES setting default to the dummy backend.
        """
        DATABASES = {}
        conns = ConnectionHandler(DATABASES)
        self.assertEqual(conns[DEFAULT_DB_ALIAS].settings_dict['ENGINE'],
            'django.db.backends.dummy')


class OracleChecks(unittest.TestCase):

    @unittest.skipUnless(connection.vendor == 'oracle',
                         "No need to check Oracle quote_name semantics")
    def test_quote_name(self):
        # Check that '%' chars are escaped for query execution.
        name = '"SOME%NAME"'
        quoted_name = connection.ops.quote_name(name)
        self.assertEqual(quoted_name % (), name)

    @unittest.skipUnless(connection.vendor == 'oracle',
                         "No need to check Oracle cursor semantics")
    def test_dbms_session(self):
        # If the backend is Oracle, test that we can call a standard
        # stored procedure through our cursor wrapper.
        from django.db.backends.oracle.base import convert_unicode

        cursor = connection.cursor()
        cursor.callproc(convert_unicode('DBMS_SESSION.SET_IDENTIFIER'),
                        [convert_unicode('_django_testing!')])

    @unittest.skipUnless(connection.vendor == 'oracle',
                         "No need to check Oracle cursor semantics")
    def test_cursor_var(self):
        # If the backend is Oracle, test that we can pass cursor variables
        # as query parameters.
        from django.db.backends.oracle.base import Database

        cursor = connection.cursor()
        var = cursor.var(Database.STRING)
        cursor.execute("BEGIN %s := 'X'; END; ", [var])
        self.assertEqual(var.getvalue(), 'X')

    @unittest.skipUnless(connection.vendor == 'oracle',
                         "No need to check Oracle cursor semantics")
    def test_long_string(self):
        # If the backend is Oracle, test that we can save a text longer
        # than 4000 chars and read it properly
        c = connection.cursor()
        c.execute('CREATE TABLE ltext ("TEXT" NCLOB)')
        long_str = ''.join(six.text_type(x) for x in xrange(4000))
        c.execute('INSERT INTO ltext VALUES (%s)', [long_str])
        c.execute('SELECT text FROM ltext')
        row = c.fetchone()
        self.assertEqual(long_str, row[0].read())
        c.execute('DROP TABLE ltext')

    @unittest.skipUnless(connection.vendor == 'oracle',
                         "No need to check Oracle connection semantics")
    def test_client_encoding(self):
        # If the backend is Oracle, test that the client encoding is set
        # correctly.  This was broken under Cygwin prior to r14781.
        connection.cursor()  # Ensure the connection is initialized.
        self.assertEqual(connection.connection.encoding, "UTF-8")
        self.assertEqual(connection.connection.nencoding, "UTF-8")

    @unittest.skipUnless(connection.vendor == 'oracle',
                         "No need to check Oracle connection semantics")
    def test_order_of_nls_parameters(self):
        # an 'almost right' datetime should work with configured
        # NLS parameters as per #18465.
        c = connection.cursor()
        query = "select 1 from dual where '1936-12-29 00:00' < sysdate"
        # Test that the query succeeds without errors - pre #18465 this
        # wasn't the case.
        c.execute(query)
        self.assertEqual(c.fetchone()[0], 1)


class SQLiteTests(TestCase):
    longMessage = True

    @unittest.skipUnless(connection.vendor == 'sqlite',
                        "Test valid only for SQLite")
    def test_autoincrement(self):
        """
        Check that auto_increment fields are created with the AUTOINCREMENT
        keyword in order to be monotonically increasing. Refs #10164.
        """
        statements = connection.creation.sql_create_model(models.Square,
            style=no_style())
        match = re.search('"id" ([^,]+),', statements[0][0])
        self.assertIsNotNone(match)
        self.assertEqual('integer NOT NULL PRIMARY KEY AUTOINCREMENT',
            match.group(1), "Wrong SQL used to create an auto-increment "
            "column on SQLite")


class MySQLTests(TestCase):
    @unittest.skipUnless(connection.vendor == 'mysql',
                        "Test valid only for MySQL")
    def test_autoincrement(self):
        """
        Check that auto_increment fields are reset correctly by sql_flush().
        Before MySQL version 5.0.13 TRUNCATE did not do auto_increment reset.
        Refs #16961.
        """
        statements = connection.ops.sql_flush(no_style(),
                                              tables=['test'],
                                              sequences=[{
                                                  'table': 'test',
                                                  'col': 'somecol',
                                              }])
        found_reset = False
        for sql in statements:
            found_reset = found_reset or 'ALTER TABLE' in sql
        if connection.mysql_version < (5, 0, 13):
            self.assertTrue(found_reset)
        else:
            self.assertFalse(found_reset)


class DateQuotingTest(TestCase):

    def test_django_date_trunc(self):
        """
        Test the custom ``django_date_trunc method``, in particular against
        fields which clash with strings passed to it (e.g. 'year') - see
        #12818__.

        __: http://code.djangoproject.com/ticket/12818

        """
        updated = datetime.datetime(2010, 2, 20)
        models.SchoolClass.objects.create(year=2009, last_updated=updated)
        years = models.SchoolClass.objects.dates('last_updated', 'year')
        self.assertEqual(list(years), [datetime.date(2010, 1, 1)])

    def test_django_date_extract(self):
        """
        Test the custom ``django_date_extract method``, in particular against fields
        which clash with strings passed to it (e.g. 'day') - see #12818__.

        __: http://code.djangoproject.com/ticket/12818

        """
        updated = datetime.datetime(2010, 2, 20)
        models.SchoolClass.objects.create(year=2009, last_updated=updated)
        classes = models.SchoolClass.objects.filter(last_updated__day=20)
        self.assertEqual(len(classes), 1)


@override_settings(DEBUG=True)
class LastExecutedQueryTest(TestCase):

    def test_last_executed_query(self):
        """
        last_executed_query should not raise an exception even if no previous
        query has been run.
        """
        cursor = connection.cursor()
        try:
            connection.ops.last_executed_query(cursor, '', ())
        except Exception:
            self.fail("'last_executed_query' should not raise an exception.")

    def test_debug_sql(self):
        list(models.Reporter.objects.filter(first_name="test"))
        sql = connection.queries[-1]['sql'].lower()
        self.assertIn("select", sql)
        self.assertIn(models.Reporter._meta.db_table, sql)

    def test_query_encoding(self):
        """
        Test that last_executed_query() returns an Unicode string
        """
        persons = models.Reporter.objects.filter(raw_data=b'\x00\x46  \xFE').extra(select={'föö': 1})
        sql, params = persons.query.sql_with_params()
        cursor = persons.query.get_compiler('default').execute_sql(None)
        last_sql = cursor.db.ops.last_executed_query(cursor, sql, params)
        self.assertIsInstance(last_sql, six.text_type)

    @unittest.skipUnless(connection.vendor == 'sqlite',
                         "This test is specific to SQLite.")
    def test_no_interpolation_on_sqlite(self):
        # Regression for #17158
        # This shouldn't raise an exception
        query = "SELECT strftime('%Y', 'now');"
        connection.cursor().execute(query)
        self.assertEqual(connection.queries[-1]['sql'],
            str_prefix("QUERY = %(_)s\"SELECT strftime('%%Y', 'now');\" - PARAMS = ()"))


class ParameterHandlingTest(TestCase):
    def test_bad_parameter_count(self):
        "An executemany call with too many/not enough parameters will raise an exception (Refs #12612)"
        cursor = connection.cursor()
        query = ('INSERT INTO %s (%s, %s) VALUES (%%s, %%s)' % (
            connection.introspection.table_name_converter('backends_square'),
            connection.ops.quote_name('root'),
            connection.ops.quote_name('square')
        ))
        self.assertRaises(Exception, cursor.executemany, query, [(1, 2, 3)])
        self.assertRaises(Exception, cursor.executemany, query, [(1,)])


# Unfortunately, the following tests would be a good test to run on all
# backends, but it breaks MySQL hard. Until #13711 is fixed, it can't be run
# everywhere (although it would be an effective test of #13711).
class LongNameTest(TestCase):
    """Long primary keys and model names can result in a sequence name
    that exceeds the database limits, which will result in truncation
    on certain databases (e.g., Postgres). The backend needs to use
    the correct sequence name in last_insert_id and other places, so
    check it is. Refs #8901.
    """

    @skipUnlessDBFeature('supports_long_model_names')
    def test_sequence_name_length_limits_create(self):
        """Test creation of model with long name and long pk name doesn't error. Ref #8901"""
        models.VeryLongModelNameZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ.objects.create()

    @skipUnlessDBFeature('supports_long_model_names')
    def test_sequence_name_length_limits_m2m(self):
        """Test an m2m save of a model with a long name and a long m2m field name doesn't error as on Django >=1.2 this now uses object saves. Ref #8901"""
        obj = models.VeryLongModelNameZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ.objects.create()
        rel_obj = models.Person.objects.create(first_name='Django', last_name='Reinhardt')
        obj.m2m_also_quite_long_zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz.add(rel_obj)

    @skipUnlessDBFeature('supports_long_model_names')
    def test_sequence_name_length_limits_flush(self):
        """Test that sequence resetting as part of a flush with model with long name and long pk name doesn't error. Ref #8901"""
        # A full flush is expensive to the full test, so we dig into the
        # internals to generate the likely offending SQL and run it manually

        # Some convenience aliases
        VLM = models.VeryLongModelNameZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ
        VLM_m2m = VLM.m2m_also_quite_long_zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz.through
        tables = [
            VLM._meta.db_table,
            VLM_m2m._meta.db_table,
        ]
        sequences = [
            {
                'column': VLM._meta.pk.column,
                'table': VLM._meta.db_table
            },
        ]
        cursor = connection.cursor()
        for statement in connection.ops.sql_flush(no_style(), tables, sequences):
            cursor.execute(statement)


class SequenceResetTest(TestCase):
    def test_generic_relation(self):
        "Sequence names are correct when resetting generic relations (Ref #13941)"
        # Create an object with a manually specified PK
        models.Post.objects.create(id=10, name='1st post', text='hello world')

        # Reset the sequences for the database
        cursor = connection.cursor()
        commands = connections[DEFAULT_DB_ALIAS].ops.sequence_reset_sql(no_style(), [models.Post])
        for sql in commands:
            cursor.execute(sql)

        # If we create a new object now, it should have a PK greater
        # than the PK we specified manually.
        obj = models.Post.objects.create(name='New post', text='goodbye world')
        self.assertTrue(obj.pk > 10)


class PostgresVersionTest(TestCase):
    def assert_parses(self, version_string, version):
        self.assertEqual(pg_version._parse_version(version_string), version)

    def test_parsing(self):
        """Test PostgreSQL version parsing from `SELECT version()` output"""
        self.assert_parses("PostgreSQL 8.3 beta4", 80300)
        self.assert_parses("PostgreSQL 8.3", 80300)
        self.assert_parses("EnterpriseDB 8.3", 80300)
        self.assert_parses("PostgreSQL 8.3.6", 80306)
        self.assert_parses("PostgreSQL 8.4beta1", 80400)
        self.assert_parses("PostgreSQL 8.3.1 on i386-apple-darwin9.2.2, compiled by GCC i686-apple-darwin9-gcc-4.0.1 (GCC) 4.0.1 (Apple Inc. build 5478)", 80301)

    def test_version_detection(self):
        """Test PostgreSQL version detection"""

        # Helper mocks
        class CursorMock(object):
            "Very simple mock of DB-API cursor"
            def execute(self, arg):
                pass

            def fetchone(self):
                return ["PostgreSQL 8.3"]

        class OlderConnectionMock(object):
            "Mock of psycopg2 (< 2.0.12) connection"
            def cursor(self):
                return CursorMock()

        # psycopg2 < 2.0.12 code path
        conn = OlderConnectionMock()
        self.assertEqual(pg_version.get_version(conn), 80300)


class PostgresNewConnectionTests(TestCase):

    @unittest.skipUnless(
        connection.vendor == 'postgresql',
        "This test applies only to PostgreSQL")
    def test_connect_and_rollback(self):
        """
        PostgreSQL shouldn't roll back SET TIME ZONE, even if the first
        transaction is rolled back (#17062).
        """
        databases = copy.deepcopy(settings.DATABASES)
        new_connections = ConnectionHandler(databases)
        new_connection = new_connections[DEFAULT_DB_ALIAS]
        try:
            # Ensure the database default time zone is different than
            # the time zone in new_connection.settings_dict. We can
            # get the default time zone by reset & show.
            cursor = new_connection.cursor()
            cursor.execute("RESET TIMEZONE")
            cursor.execute("SHOW TIMEZONE")
            db_default_tz = cursor.fetchone()[0]
            new_tz = 'Europe/Paris' if db_default_tz == 'UTC' else 'UTC'
            new_connection.close()

            # Fetch a new connection with the new_tz as default
            # time zone, run a query and rollback.
            new_connection.settings_dict['TIME_ZONE'] = new_tz
            new_connection.enter_transaction_management()
            cursor = new_connection.cursor()
            new_connection.rollback()

            # Now let's see if the rollback rolled back the SET TIME ZONE.
            cursor.execute("SHOW TIMEZONE")
            tz = cursor.fetchone()[0]
            self.assertEqual(new_tz, tz)
        finally:
            new_connection.close()

    @unittest.skipUnless(
        connection.vendor == 'postgresql',
        "This test applies only to PostgreSQL")
    def test_connect_non_autocommit(self):
        """
        The connection wrapper shouldn't believe that autocommit is enabled
        after setting the time zone when AUTOCOMMIT is False (#21452).
        """
        databases = copy.deepcopy(settings.DATABASES)
        new_connections = ConnectionHandler(databases)
        new_connection = new_connections[DEFAULT_DB_ALIAS]
        try:
            new_connection.settings_dict['AUTOCOMMIT'] = False
            self.assertFalse(new_connection.get_autocommit())
        finally:
            new_connection.close()


# This test needs to run outside of a transaction, otherwise closing the
# connection would implicitly rollback and cause problems during teardown.
class ConnectionCreatedSignalTest(TransactionTestCase):

    available_apps = []<commit_msg>Restored a test broken by overzealous PEP8 cleanup<commit_after># -*- coding: utf-8 -*-
# Unit and doctests for specific database backends.
from __future__ import unicode_literals

import copy
import datetime
from decimal import Decimal
import re
import threading
import unittest

from django.conf import settings
from django.core.management.color import no_style
from django.db import (connection, connections, DEFAULT_DB_ALIAS,
    DatabaseError, IntegrityError, transaction)
from django.db.backends.signals import connection_created
from django.db.backends.sqlite3.base import DatabaseOperations
from django.db.backends.postgresql_psycopg2 import version as pg_version
from django.db.backends.utils import format_number, CursorWrapper
from django.db.models import Sum, Avg, Variance, StdDev
from django.db.models.fields import (AutoField, DateField, DateTimeField,
    DecimalField, IntegerField, TimeField)
from django.db.utils import ConnectionHandler
from django.test import (TestCase, TransactionTestCase, override_settings,
    skipUnlessDBFeature, skipIfDBFeature)
from django.test.utils import str_prefix
from django.utils import six
from django.utils.six.moves import xrange

from . import models


class DummyBackendTest(TestCase):
    def test_no_databases(self):
        """
        Test that empty DATABASES setting default to the dummy backend.
        """
        DATABASES = {}
        conns = ConnectionHandler(DATABASES)
        self.assertEqual(conns[DEFAULT_DB_ALIAS].settings_dict['ENGINE'],
            'django.db.backends.dummy')


class OracleChecks(unittest.TestCase):

    @unittest.skipUnless(connection.vendor == 'oracle',
                         "No need to check Oracle quote_name semantics")
    def test_quote_name(self):
        # Check that '%' chars are escaped for query execution.
        name = '"SOME%NAME"'
        quoted_name = connection.ops.quote_name(name)
        self.assertEqual(quoted_name % (), name)

    @unittest.skipUnless(connection.vendor == 'oracle',
                         "No need to check Oracle cursor semantics")
    def test_dbms_session(self):
        # If the backend is Oracle, test that we can call a standard
        # stored procedure through our cursor wrapper.
        from django.db.backends.oracle.base import convert_unicode

        cursor = connection.cursor()
        cursor.callproc(convert_unicode('DBMS_SESSION.SET_IDENTIFIER'),
                        [convert_unicode('_django_testing!')])

    @unittest.skipUnless(connection.vendor == 'oracle',
                         "No need to check Oracle cursor semantics")
    def test_cursor_var(self):
        # If the backend is Oracle, test that we can pass cursor variables
        # as query parameters.
        from django.db.backends.oracle.base import Database

        cursor = connection.cursor()
        var = cursor.var(Database.STRING)
        cursor.execute("BEGIN %s := 'X'; END; ", [var])
        self.assertEqual(var.getvalue(), 'X')

    @unittest.skipUnless(connection.vendor == 'oracle',
                         "No need to check Oracle cursor semantics")
    def test_long_string(self):
        # If the backend is Oracle, test that we can save a text longer
        # than 4000 chars and read it properly
        c = connection.cursor()
        c.execute('CREATE TABLE ltext ("TEXT" NCLOB)')
        long_str = ''.join(six.text_type(x) for x in xrange(4000))
        c.execute('INSERT INTO ltext VALUES (%s)', [long_str])
        c.execute('SELECT text FROM ltext')
        row = c.fetchone()
        self.assertEqual(long_str, row[0].read())
        c.execute('DROP TABLE ltext')

    @unittest.skipUnless(connection.vendor == 'oracle',
                         "No need to check Oracle connection semantics")
    def test_client_encoding(self):
        # If the backend is Oracle, test that the client encoding is set
        # correctly.  This was broken under Cygwin prior to r14781.
        connection.cursor()  # Ensure the connection is initialized.
        self.assertEqual(connection.connection.encoding, "UTF-8")
        self.assertEqual(connection.connection.nencoding, "UTF-8")

    @unittest.skipUnless(connection.vendor == 'oracle',
                         "No need to check Oracle connection semantics")
    def test_order_of_nls_parameters(self):
        # an 'almost right' datetime should work with configured
        # NLS parameters as per #18465.
        c = connection.cursor()
        query = "select 1 from dual where '1936-12-29 00:00' < sysdate"
        # Test that the query succeeds without errors - pre #18465 this
        # wasn't the case.
        c.execute(query)
        self.assertEqual(c.fetchone()[0], 1)


class SQLiteTests(TestCase):
    longMessage = True

    @unittest.skipUnless(connection.vendor == 'sqlite',
                        "Test valid only for SQLite")
    def test_autoincrement(self):
        """
        Check that auto_increment fields are created with the AUTOINCREMENT
        keyword in order to be monotonically increasing. Refs #10164.
        """
        statements = connection.creation.sql_create_model(models.Square,
            style=no_style())
        match = re.search('"id" ([^,]+),', statements[0][0])
        self.assertIsNotNone(match)
        self.assertEqual('integer NOT NULL PRIMARY KEY AUTOINCREMENT',
            match.group(1), "Wrong SQL used to create an auto-increment "
            "column on SQLite")


class MySQLTests(TestCase):
    @unittest.skipUnless(connection.vendor == 'mysql',
                        "Test valid only for MySQL")
    def test_autoincrement(self):
        """
        Check that auto_increment fields are reset correctly by sql_flush().
        Before MySQL version 5.0.13 TRUNCATE did not do auto_increment reset.
        Refs #16961.
        """
        statements = connection.ops.sql_flush(no_style(),
                                              tables=['test'],
                                              sequences=[{
                                                  'table': 'test',
                                                  'col': 'somecol',
                                              }])
        found_reset = False
        for sql in statements:
            found_reset = found_reset or 'ALTER TABLE' in sql
        if connection.mysql_version < (5, 0, 13):
            self.assertTrue(found_reset)
        else:
            self.assertFalse(found_reset)


class DateQuotingTest(TestCase):

    def test_django_date_trunc(self):
        """
        Test the custom ``django_date_trunc method``, in particular against
        fields which clash with strings passed to it (e.g. 'year') - see
        #12818__.

        __: http://code.djangoproject.com/ticket/12818

        """
        updated = datetime.datetime(2010, 2, 20)
        models.SchoolClass.objects.create(year=2009, last_updated=updated)
        years = models.SchoolClass.objects.dates('last_updated', 'year')
        self.assertEqual(list(years), [datetime.date(2010, 1, 1)])

    def test_django_date_extract(self):
        """
        Test the custom ``django_date_extract method``, in particular against fields
        which clash with strings passed to it (e.g. 'day') - see #12818__.

        __: http://code.djangoproject.com/ticket/12818

        """
        updated = datetime.datetime(2010, 2, 20)
        models.SchoolClass.objects.create(year=2009, last_updated=updated)
        classes = models.SchoolClass.objects.filter(last_updated__day=20)
        self.assertEqual(len(classes), 1)


@override_settings(DEBUG=True)
class LastExecutedQueryTest(TestCase):

    def test_last_executed_query(self):
        """
        last_executed_query should not raise an exception even if no previous
        query has been run.
        """
        cursor = connection.cursor()
        try:
            connection.ops.last_executed_query(cursor, '', ())
        except Exception:
            self.fail("'last_executed_query' should not raise an exception.")

    def test_debug_sql(self):
        list(models.Reporter.objects.filter(first_name="test"))
        sql = connection.queries[-1]['sql'].lower()
        self.assertIn("select", sql)
        self.assertIn(models.Reporter._meta.db_table, sql)

    def test_query_encoding(self):
        """
        Test that last_executed_query() returns an Unicode string
        """
        persons = models.Reporter.objects.filter(raw_data=b'\x00\x46  \xFE').extra(select={'föö': 1})
        sql, params = persons.query.sql_with_params()
        cursor = persons.query.get_compiler('default').execute_sql(None)
        last_sql = cursor.db.ops.last_executed_query(cursor, sql, params)
        self.assertIsInstance(last_sql, six.text_type)

    @unittest.skipUnless(connection.vendor == 'sqlite',
                         "This test is specific to SQLite.")
    def test_no_interpolation_on_sqlite(self):
        # Regression for #17158
        # This shouldn't raise an exception
        query = "SELECT strftime('%Y', 'now');"
        connection.cursor().execute(query)
        self.assertEqual(connection.queries[-1]['sql'],
            str_prefix("QUERY = %(_)s\"SELECT strftime('%%Y', 'now');\" - PARAMS = ()"))


class ParameterHandlingTest(TestCase):
    def test_bad_parameter_count(self):
        "An executemany call with too many/not enough parameters will raise an exception (Refs #12612)"
        cursor = connection.cursor()
        query = ('INSERT INTO %s (%s, %s) VALUES (%%s, %%s)' % (
            connection.introspection.table_name_converter('backends_square'),
            connection.ops.quote_name('root'),
            connection.ops.quote_name('square')
        ))
        self.assertRaises(Exception, cursor.executemany, query, [(1, 2, 3)])
        self.assertRaises(Exception, cursor.executemany, query, [(1,)])


# Unfortunately, the following tests would be a good test to run on all
# backends, but it breaks MySQL hard. Until #13711 is fixed, it can't be run
# everywhere (although it would be an effective test of #13711).
class LongNameTest(TestCase):
    """Long primary keys and model names can result in a sequence name
    that exceeds the database limits, which will result in truncation
    on certain databases (e.g., Postgres). The backend needs to use
    the correct sequence name in last_insert_id and other places, so
    check it is. Refs #8901.
    """

    @skipUnlessDBFeature('supports_long_model_names')
    def test_sequence_name_length_limits_create(self):
        """Test creation of model with long name and long pk name doesn't error. Ref #8901"""
        models.VeryLongModelNameZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ.objects.create()

    @skipUnlessDBFeature('supports_long_model_names')
    def test_sequence_name_length_limits_m2m(self):
        """Test an m2m save of a model with a long name and a long m2m field name doesn't error as on Django >=1.2 this now uses object saves. Ref #8901"""
        obj = models.VeryLongModelNameZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ.objects.create()
        rel_obj = models.Person.objects.create(first_name='Django', last_name='Reinhardt')
        obj.m2m_also_quite_long_zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz.add(rel_obj)

    @skipUnlessDBFeature('supports_long_model_names')
    def test_sequence_name_length_limits_flush(self):
        """Test that sequence resetting as part of a flush with model with long name and long pk name doesn't error. Ref #8901"""
        # A full flush is expensive to the full test, so we dig into the
        # internals to generate the likely offending SQL and run it manually

        # Some convenience aliases
        VLM = models.VeryLongModelNameZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ
        VLM_m2m = VLM.m2m_also_quite_long_zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz.through
        tables = [
            VLM._meta.db_table,
            VLM_m2m._meta.db_table,
        ]
        sequences = [
            {
                'column': VLM._meta.pk.column,
                'table': VLM._meta.db_table
            },
        ]
        cursor = connection.cursor()
        for statement in connection.ops.sql_flush(no_style(), tables, sequences):
            cursor.execute(statement)


class SequenceResetTest(TestCase):
    def test_generic_relation(self):
        "Sequence names are correct when resetting generic relations (Ref #13941)"
        # Create an object with a manually specified PK
        models.Post.objects.create(id=10, name='1st post', text='hello world')

        # Reset the sequences for the database
        cursor = connection.cursor()
        commands = connections[DEFAULT_DB_ALIAS].ops.sequence_reset_sql(no_style(), [models.Post])
        for sql in commands:
            cursor.execute(sql)

        # If we create a new object now, it should have a PK greater
        # than the PK we specified manually.
        obj = models.Post.objects.create(name='New post', text='goodbye world')
        self.assertTrue(obj.pk > 10)


class PostgresVersionTest(TestCase):
    def assert_parses(self, version_string, version):
        self.assertEqual(pg_version._parse_version(version_string), version)

    def test_parsing(self):
        """Test PostgreSQL version parsing from `SELECT version()` output"""
        self.assert_parses("PostgreSQL 8.3 beta4", 80300)
        self.assert_parses("PostgreSQL 8.3", 80300)
        self.assert_parses("EnterpriseDB 8.3", 80300)
        self.assert_parses("PostgreSQL 8.3.6", 80306)
        self.assert_parses("PostgreSQL 8.4beta1", 80400)
        self.assert_parses("PostgreSQL 8.3.1 on i386-apple-darwin9.2.2, compiled by GCC i686-apple-darwin9-gcc-4.0.1 (GCC) 4.0.1 (Apple Inc. build 5478)", 80301)

    def test_version_detection(self):
        """Test PostgreSQL version detection"""

        # Helper mocks
        class CursorMock(object):
            "Very simple mock of DB-API cursor"
            def execute(self, arg):
                pass

            def fetchone(self):
                return ["PostgreSQL 8.3"]

        class OlderConnectionMock(object):
            "Mock of psycopg2 (< 2.0.12) connection"
            def cursor(self):
                return CursorMock()

        # psycopg2 < 2.0.12 code path
        conn = OlderConnectionMock()
        self.assertEqual(pg_version.get_version(conn), 80300)


class PostgresNewConnectionTests(TestCase):

    @unittest.skipUnless(
        connection.vendor == 'postgresql',
        "This test applies only to PostgreSQL")
    def test_connect_and_rollback(self):
        """
        PostgreSQL shouldn't roll back SET TIME ZONE, even if the first
        transaction is rolled back (#17062).
        """
        databases = copy.deepcopy(settings.DATABASES)
        new_connections = ConnectionHandler(databases)
        new_connection = new_connections[DEFAULT_DB_ALIAS]
        try:
            # Ensure the database default time zone is different than
            # the time zone in new_connection.settings_dict. We can
            # get the default time zone by reset & show.
            cursor = new_connection.cursor()
            cursor.execute("RESET TIMEZONE")
            cursor.execute("SHOW TIMEZONE")
            db_default_tz = cursor.fetchone()[0]
            new_tz = 'Europe/Paris' if db_default_tz == 'UTC' else 'UTC'
            new_connection.close()

            # Fetch a new connection with the new_tz as default
            # time zone, run a query and rollback.
            new_connection.settings_dict['TIME_ZONE'] = new_tz
            new_connection.enter_transaction_management()
            cursor = new_connection.cursor()
            new_connection.rollback()

            # Now let's see if the rollback rolled back the SET TIME ZONE.
            cursor.execute("SHOW TIMEZONE")
            tz = cursor.fetchone()[0]
            self.assertEqual(new_tz, tz)
        finally:
            new_connection.close()

    @unittest.skipUnless(
        connection.vendor == 'postgresql',
        "This test applies only to PostgreSQL")
    def test_connect_non_autocommit(self):
        """
        The connection wrapper shouldn't believe that autocommit is enabled
        after setting the time zone when AUTOCOMMIT is False (#21452).
        """
        databases = copy.deepcopy(settings.DATABASES)
        new_connections = ConnectionHandler(databases)
        new_connection = new_connections[DEFAULT_DB_ALIAS]
        try:
            new_connection.settings_dict['AUTOCOMMIT'] = False
            # Open a database connection.
            new_connection.cursor()
            self.assertFalse(new_connection.get_autocommit())
        finally:
            new_connection.close()


# This test needs to run outside of a transaction, otherwise closing the
# connection would implicitly rollback and cause problems during teardown.
class ConnectionCreatedSignalTest(TransactionTestCase):

    available_apps = []
{'fraction_non_alphanumeric': 0.0934200798462221, 'fraction_numerical': 0.016412834540884224, 'mean_word_length': 2.6557837837837837, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 10, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6166008', 'n_tokens_mistral': 9903, 'n_tokens_neox': 9256, 'n_words': 2554}