|
|
@@ -1,1060 +0,0 @@
|
|
|
-import datetime
|
|
|
-from warnings import filterwarnings
|
|
|
-
|
|
|
-from south.db import db, generic
|
|
|
-from django.db import connection, models, IntegrityError as DjangoIntegrityError
|
|
|
-
|
|
|
-from south.tests import unittest, skipIf, skipUnless
|
|
|
-from south.utils.py3 import text_type, with_metaclass
|
|
|
-
|
|
|
-# Create a list of error classes from the various database libraries
|
|
|
-errors = []
|
|
|
-try:
|
|
|
- from psycopg2 import ProgrammingError
|
|
|
- errors.append(ProgrammingError)
|
|
|
-except ImportError:
|
|
|
- pass
|
|
|
-errors = tuple(errors)
|
|
|
-
|
|
|
-# On SQL Server, the backend's IntegrityError is not (a subclass of) Django's.
|
|
|
-try:
|
|
|
- from sql_server.pyodbc.base import IntegrityError as SQLServerIntegrityError
|
|
|
- IntegrityError = (DjangoIntegrityError, SQLServerIntegrityError)
|
|
|
-except ImportError:
|
|
|
- IntegrityError = DjangoIntegrityError
|
|
|
-
|
|
|
-try:
|
|
|
- from south.db import mysql
|
|
|
-except ImportError:
|
|
|
- mysql = None
|
|
|
-
|
|
|
-
|
|
|
-class TestOperations(unittest.TestCase):
|
|
|
-
|
|
|
- """
|
|
|
- Tests if the various DB abstraction calls work.
|
|
|
- Can only test a limited amount due to DB differences.
|
|
|
- """
|
|
|
-
|
|
|
- def setUp(self):
|
|
|
- db.debug = False
|
|
|
- try:
|
|
|
- import MySQLdb
|
|
|
- except ImportError:
|
|
|
- pass
|
|
|
- else:
|
|
|
- filterwarnings('ignore', category=MySQLdb.Warning)
|
|
|
- db.clear_deferred_sql()
|
|
|
- db.start_transaction()
|
|
|
-
|
|
|
- def tearDown(self):
|
|
|
- db.rollback_transaction()
|
|
|
-
|
|
|
- def test_create(self):
|
|
|
- """
|
|
|
- Test creation of tables.
|
|
|
- """
|
|
|
- cursor = connection.cursor()
|
|
|
- # It needs to take at least 2 args
|
|
|
- self.assertRaises(TypeError, db.create_table)
|
|
|
- self.assertRaises(TypeError, db.create_table, "test1")
|
|
|
- # Empty tables (i.e. no columns) are not fine, so make at least 1
|
|
|
- db.create_table("test1", [('email_confirmed', models.BooleanField(default=False))])
|
|
|
- # And should exist
|
|
|
- cursor.execute("SELECT * FROM test1")
|
|
|
- # Make sure we can't do the same query on an empty table
|
|
|
- try:
|
|
|
- cursor.execute("SELECT * FROM nottheretest1")
|
|
|
- except:
|
|
|
- pass
|
|
|
- else:
|
|
|
- self.fail("Non-existent table could be selected!")
|
|
|
-
|
|
|
- @skipUnless(db.raises_default_errors, 'This database does not raise errors on missing defaults.')
|
|
|
- def test_create_default(self):
|
|
|
- """
|
|
|
- Test creation of tables, make sure defaults are not left in the database
|
|
|
- """
|
|
|
- db.create_table("test_create_default", [('a', models.IntegerField()),
|
|
|
- ('b', models.IntegerField(default=17))])
|
|
|
- cursor = connection.cursor()
|
|
|
- self.assertRaises(IntegrityError, cursor.execute, "INSERT INTO test_create_default(a) VALUES (17)")
|
|
|
-
|
|
|
- def test_delete(self):
|
|
|
- """
|
|
|
- Test deletion of tables.
|
|
|
- """
|
|
|
- cursor = connection.cursor()
|
|
|
- db.create_table("test_deltable", [('email_confirmed', models.BooleanField(default=False))])
|
|
|
- db.delete_table("test_deltable")
|
|
|
- # Make sure it went
|
|
|
- try:
|
|
|
- cursor.execute("SELECT * FROM test_deltable")
|
|
|
- except:
|
|
|
- pass
|
|
|
- else:
|
|
|
- self.fail("Just-deleted table could be selected!")
|
|
|
-
|
|
|
- def test_nonexistent_delete(self):
|
|
|
- """
|
|
|
- Test deletion of nonexistent tables.
|
|
|
- """
|
|
|
- try:
|
|
|
- db.delete_table("test_nonexistdeltable")
|
|
|
- except:
|
|
|
- pass
|
|
|
- else:
|
|
|
- self.fail("Non-existent table could be deleted!")
|
|
|
-
|
|
|
- def test_foreign_keys(self):
|
|
|
- """
|
|
|
- Tests foreign key creation, especially uppercase (see #61)
|
|
|
- """
|
|
|
- Test = db.mock_model(model_name='Test', db_table='test5a',
|
|
|
- db_tablespace='', pk_field_name='ID',
|
|
|
- pk_field_type=models.AutoField, pk_field_args=[])
|
|
|
- db.create_table("test5a", [('ID', models.AutoField(verbose_name='ID', primary_key=True, auto_created=True))])
|
|
|
- db.create_table("test5b", [
|
|
|
- ('id', models.AutoField(verbose_name='ID', primary_key=True, auto_created=True)),
|
|
|
- ('UNIQUE', models.ForeignKey(Test)),
|
|
|
- ])
|
|
|
- db.execute_deferred_sql()
|
|
|
-
|
|
|
- @skipUnless(db.supports_foreign_keys, 'Foreign keys can only be deleted on '
|
|
|
- 'engines that support them.')
|
|
|
- def test_recursive_foreign_key_delete(self):
|
|
|
- """
|
|
|
- Test that recursive foreign keys are deleted correctly (see #1065)
|
|
|
- """
|
|
|
- Test = db.mock_model(model_name='Test', db_table='test_rec_fk_del',
|
|
|
- db_tablespace='', pk_field_name='id',
|
|
|
- pk_field_type=models.AutoField, pk_field_args=[])
|
|
|
- db.create_table('test_rec_fk_del', [
|
|
|
- ('id', models.AutoField(primary_key=True, auto_created=True)),
|
|
|
- ('fk', models.ForeignKey(Test)),
|
|
|
- ])
|
|
|
- db.execute_deferred_sql()
|
|
|
- db.delete_foreign_key('test_rec_fk_del', 'fk_id')
|
|
|
-
|
|
|
- def test_rename(self):
|
|
|
- """
|
|
|
- Test column renaming
|
|
|
- """
|
|
|
- cursor = connection.cursor()
|
|
|
- db.create_table("test_rn", [('spam', models.BooleanField(default=False))])
|
|
|
- # Make sure we can select the column
|
|
|
- cursor.execute("SELECT spam FROM test_rn")
|
|
|
- # Rename it
|
|
|
- db.rename_column("test_rn", "spam", "eggs")
|
|
|
- cursor.execute("SELECT eggs FROM test_rn")
|
|
|
- db.commit_transaction()
|
|
|
- db.start_transaction()
|
|
|
- try:
|
|
|
- cursor.execute("SELECT spam FROM test_rn")
|
|
|
- except:
|
|
|
- pass
|
|
|
- else:
|
|
|
- self.fail("Just-renamed column could be selected!")
|
|
|
- db.rollback_transaction()
|
|
|
- db.delete_table("test_rn")
|
|
|
- db.start_transaction()
|
|
|
-
|
|
|
- def test_dry_rename(self):
|
|
|
- """
|
|
|
- Test column renaming while --dry-run is turned on (should do nothing)
|
|
|
- See ticket #65
|
|
|
- """
|
|
|
- cursor = connection.cursor()
|
|
|
- db.create_table("test_drn", [('spam', models.BooleanField(default=False))])
|
|
|
- # Make sure we can select the column
|
|
|
- cursor.execute("SELECT spam FROM test_drn")
|
|
|
- # Rename it
|
|
|
- db.dry_run = True
|
|
|
- db.rename_column("test_drn", "spam", "eggs")
|
|
|
- db.dry_run = False
|
|
|
- cursor.execute("SELECT spam FROM test_drn")
|
|
|
- db.commit_transaction()
|
|
|
- db.start_transaction()
|
|
|
- try:
|
|
|
- cursor.execute("SELECT eggs FROM test_drn")
|
|
|
- except:
|
|
|
- pass
|
|
|
- else:
|
|
|
- self.fail("Dry-renamed new column could be selected!")
|
|
|
- db.rollback_transaction()
|
|
|
- db.delete_table("test_drn")
|
|
|
- db.start_transaction()
|
|
|
-
|
|
|
- def test_table_rename(self):
|
|
|
- """
|
|
|
- Test column renaming
|
|
|
- """
|
|
|
- cursor = connection.cursor()
|
|
|
- db.create_table("testtr", [('spam', models.BooleanField(default=False))])
|
|
|
- # Make sure we can select the column
|
|
|
- cursor.execute("SELECT spam FROM testtr")
|
|
|
- # Rename it
|
|
|
- db.rename_table("testtr", "testtr2")
|
|
|
- cursor.execute("SELECT spam FROM testtr2")
|
|
|
- db.commit_transaction()
|
|
|
- db.start_transaction()
|
|
|
- try:
|
|
|
- cursor.execute("SELECT spam FROM testtr")
|
|
|
- except:
|
|
|
- pass
|
|
|
- else:
|
|
|
- self.fail("Just-renamed column could be selected!")
|
|
|
- db.rollback_transaction()
|
|
|
- db.delete_table("testtr2")
|
|
|
- db.start_transaction()
|
|
|
-
|
|
|
- def test_percents_in_defaults(self):
|
|
|
- """
|
|
|
- Test that % in a default gets escaped to %%.
|
|
|
- """
|
|
|
- try:
|
|
|
- db.create_table("testpind", [('cf', models.CharField(max_length=255, default="It should be 2%!"))])
|
|
|
- except IndexError:
|
|
|
- self.fail("% was not properly escaped in column SQL.")
|
|
|
- db.delete_table("testpind")
|
|
|
-
|
|
|
- def test_index(self):
|
|
|
- """
|
|
|
- Test the index operations
|
|
|
- """
|
|
|
- db.create_table("test3", [
|
|
|
- ('SELECT', models.BooleanField(default=False)),
|
|
|
- ('eggs', models.IntegerField(unique=True)),
|
|
|
- ])
|
|
|
- db.execute_deferred_sql()
|
|
|
- # Add an index on that column
|
|
|
- db.create_index("test3", ["SELECT"])
|
|
|
- # Add another index on two columns
|
|
|
- db.create_index("test3", ["SELECT", "eggs"])
|
|
|
- # Delete them both
|
|
|
- db.delete_index("test3", ["SELECT"])
|
|
|
- db.delete_index("test3", ["SELECT", "eggs"])
|
|
|
- # Delete the unique index/constraint
|
|
|
- if db.backend_name != "sqlite3":
|
|
|
- db.delete_unique("test3", ["eggs"])
|
|
|
- db.delete_table("test3")
|
|
|
-
|
|
|
- def test_primary_key(self):
|
|
|
- """
|
|
|
- Test the primary key operations
|
|
|
- """
|
|
|
-
|
|
|
- db.create_table("test_pk", [
|
|
|
- ('id', models.IntegerField(primary_key=True)),
|
|
|
- ('new_pkey', models.IntegerField()),
|
|
|
- ('eggs', models.IntegerField(unique=True)),
|
|
|
- ])
|
|
|
- db.execute_deferred_sql()
|
|
|
- # Remove the default primary key, and make eggs it
|
|
|
- db.delete_primary_key("test_pk")
|
|
|
- db.create_primary_key("test_pk", "new_pkey")
|
|
|
- # Try inserting a now-valid row pair
|
|
|
- db.execute("INSERT INTO test_pk (id, new_pkey, eggs) VALUES (1, 2, 3)")
|
|
|
- db.execute("INSERT INTO test_pk (id, new_pkey, eggs) VALUES (1, 3, 4)")
|
|
|
- db.delete_table("test_pk")
|
|
|
-
|
|
|
- def test_primary_key_implicit(self):
|
|
|
- """
|
|
|
- Tests that changing primary key implicitly fails.
|
|
|
- """
|
|
|
- db.create_table("test_pki", [
|
|
|
- ('id', models.IntegerField(primary_key=True)),
|
|
|
- ('new_pkey', models.IntegerField()),
|
|
|
- ('eggs', models.IntegerField(unique=True)),
|
|
|
- ])
|
|
|
- db.execute_deferred_sql()
|
|
|
- # Fiddle with alter_column to attempt to make it remove the primary key
|
|
|
- db.alter_column("test_pki", "id", models.IntegerField())
|
|
|
- db.alter_column("test_pki", "new_pkey", models.IntegerField(primary_key=True))
|
|
|
- # Try inserting a should-be-valid row pair
|
|
|
- db.execute("INSERT INTO test_pki (id, new_pkey, eggs) VALUES (1, 2, 3)")
|
|
|
- db.execute("INSERT INTO test_pki (id, new_pkey, eggs) VALUES (2, 2, 4)")
|
|
|
- db.delete_table("test_pki")
|
|
|
-
|
|
|
- def test_add_columns(self):
|
|
|
- """
|
|
|
- Test adding columns
|
|
|
- """
|
|
|
- db.create_table("test_addc", [
|
|
|
- ('spam', models.BooleanField(default=False)),
|
|
|
- ('eggs', models.IntegerField()),
|
|
|
- ])
|
|
|
- # Add a column
|
|
|
- db.add_column("test_addc", "add1", models.IntegerField(default=3))
|
|
|
- User = db.mock_model(model_name='User', db_table='auth_user', db_tablespace='', pk_field_name='id', pk_field_type=models.AutoField, pk_field_args=[], pk_field_kwargs={})
|
|
|
- # insert some data so we can test the default value of the added fkey
|
|
|
- db.execute("INSERT INTO test_addc (spam, eggs, add1) VALUES (%s, 1, 2)", [False])
|
|
|
- db.add_column("test_addc", "user", models.ForeignKey(User, null=True))
|
|
|
- db.execute_deferred_sql()
|
|
|
- # try selecting from the user_id column to make sure it was actually created
|
|
|
- val = db.execute("SELECT user_id FROM test_addc")[0][0]
|
|
|
- self.assertEquals(val, None)
|
|
|
- db.delete_column("test_addc", "add1")
|
|
|
- # make sure adding an indexed field works
|
|
|
- db.add_column("test_addc", "add2", models.CharField(max_length=15, db_index=True, default='pi'))
|
|
|
- db.execute_deferred_sql()
|
|
|
- db.delete_table("test_addc")
|
|
|
-
|
|
|
- def test_delete_columns(self):
|
|
|
- """
|
|
|
- Test deleting columns
|
|
|
- """
|
|
|
- db.create_table("test_delc", [
|
|
|
- ('spam', models.BooleanField(default=False)),
|
|
|
- ('eggs', models.IntegerField(db_index=True, unique=True)),
|
|
|
- ])
|
|
|
- db.delete_column("test_delc", "eggs")
|
|
|
-
|
|
|
- def test_add_nullbool_column(self):
|
|
|
- """
|
|
|
- Test adding NullBoolean columns
|
|
|
- """
|
|
|
- db.create_table("test_addnbc", [
|
|
|
- ('spam', models.BooleanField(default=False)),
|
|
|
- ('eggs', models.IntegerField()),
|
|
|
- ])
|
|
|
- # Add a column
|
|
|
- db.add_column("test_addnbc", "add1", models.NullBooleanField())
|
|
|
- # Add a column with a default
|
|
|
- db.add_column("test_addnbc", "add2", models.NullBooleanField(default=True))
|
|
|
- # insert some data so we can test the default values of the added column
|
|
|
- db.execute("INSERT INTO test_addnbc (spam, eggs) VALUES (%s, 1)", [False])
|
|
|
- # try selecting from the new columns to make sure they were properly created
|
|
|
- false, null1, null2 = db.execute("SELECT spam,add1,add2 FROM test_addnbc")[0][0:3]
|
|
|
- self.assertIsNone(null1, "Null boolean field with no value inserted returns non-null")
|
|
|
- self.assertIsNone(null2, "Null boolean field (added with default) with no value inserted returns non-null")
|
|
|
- self.assertEquals(false, False)
|
|
|
- db.delete_table("test_addnbc")
|
|
|
-
|
|
|
- def test_alter_columns(self):
|
|
|
- """
|
|
|
- Test altering columns
|
|
|
- """
|
|
|
- db.create_table("test_alterc", [
|
|
|
- ('spam', models.BooleanField(default=False)),
|
|
|
- ('eggs', models.IntegerField()),
|
|
|
- ])
|
|
|
- db.execute_deferred_sql()
|
|
|
- # Change eggs to be a FloatField
|
|
|
- db.alter_column("test_alterc", "eggs", models.FloatField())
|
|
|
- db.execute_deferred_sql()
|
|
|
- db.delete_table("test_alterc")
|
|
|
- db.execute_deferred_sql()
|
|
|
-
|
|
|
- def test_alter_char_default(self):
|
|
|
- """
|
|
|
- Test altering column defaults with char fields
|
|
|
- """
|
|
|
- db.create_table("test_altercd", [
|
|
|
- ('spam', models.CharField(max_length=30)),
|
|
|
- ('eggs', models.IntegerField()),
|
|
|
- ])
|
|
|
- # Change spam default
|
|
|
- db.alter_column("test_altercd", "spam", models.CharField(max_length=30, default="loof", null=True))
|
|
|
- # Assert the default is not in the database
|
|
|
- db.execute("INSERT INTO test_altercd (eggs) values (12)")
|
|
|
- null = db.execute("SELECT spam FROM test_altercd")[0][0]
|
|
|
- self.assertFalse(null, "Default for char field was installed into database")
|
|
|
-
|
|
|
- # Change again to a column with default and not null
|
|
|
- db.alter_column("test_altercd", "spam", models.CharField(max_length=30, default="loof", null=False))
|
|
|
- # Assert the default is not in the database
|
|
|
- if 'oracle' in db.backend_name:
|
|
|
- # Oracle special treatment -- nulls are always allowed in char columns, so
|
|
|
- # inserting doesn't raise an integrity error; so we check again as above
|
|
|
- db.execute("DELETE FROM test_altercd")
|
|
|
- db.execute("INSERT INTO test_altercd (eggs) values (12)")
|
|
|
- null = db.execute("SELECT spam FROM test_altercd")[0][0]
|
|
|
- self.assertFalse(null, "Default for char field was installed into database")
|
|
|
- else:
|
|
|
- # For other backends, insert should now just fail
|
|
|
- self.assertRaises(IntegrityError,
|
|
|
- db.execute, "INSERT INTO test_altercd (eggs) values (12)")
|
|
|
-
|
|
|
- @skipIf('oracle' in db.backend_name, "Oracle does not differentiate empty trings from null")
|
|
|
- def test_default_empty_string(self):
|
|
|
- """
|
|
|
- Test altering column defaults with char fields
|
|
|
- """
|
|
|
- db.create_table("test_cd_empty", [
|
|
|
- ('spam', models.CharField(max_length=30, default='')),
|
|
|
- ('eggs', models.CharField(max_length=30)),
|
|
|
- ])
|
|
|
- # Create a record
|
|
|
- db.execute("INSERT INTO test_cd_empty (spam, eggs) values ('1','2')")
|
|
|
- # Add a column
|
|
|
- db.add_column("test_cd_empty", "ham", models.CharField(max_length=30, default=''))
|
|
|
-
|
|
|
- empty = db.execute("SELECT ham FROM test_cd_empty")[0][0]
|
|
|
- self.assertEquals(empty, "", "Empty Default for char field isn't empty string")
|
|
|
-
|
|
|
- @skipUnless('oracle' in db.backend_name, "Oracle does not differentiate empty trings from null")
|
|
|
- def test_oracle_strings_null(self):
|
|
|
- """
|
|
|
- Test that under Oracle, CherFields are created as null even when specified not-null,
|
|
|
- because otherwise they would not be able to hold empty strings (which Oracle equates
|
|
|
- with nulls).
|
|
|
- Verify fix of #1269.
|
|
|
- """
|
|
|
- db.create_table("test_ora_char_nulls", [
|
|
|
- ('spam', models.CharField(max_length=30, null=True)),
|
|
|
- ('eggs', models.CharField(max_length=30)),
|
|
|
- ])
|
|
|
- db.add_column("test_ora_char_nulls", "ham", models.CharField(max_length=30))
|
|
|
- db.alter_column("test_ora_char_nulls", "spam", models.CharField(max_length=30, null=False))
|
|
|
- # So, by the look of it, we should now have three not-null columns
|
|
|
- db.execute("INSERT INTO test_ora_char_nulls VALUES (NULL, NULL, NULL)")
|
|
|
-
|
|
|
-
|
|
|
- def test_mysql_defaults(self):
|
|
|
- """
|
|
|
- Test MySQL default handling for BLOB and TEXT.
|
|
|
- """
|
|
|
- db.create_table("test_altermyd", [
|
|
|
- ('spam', models.BooleanField(default=False)),
|
|
|
- ('eggs', models.TextField()),
|
|
|
- ])
|
|
|
- # Change eggs to be a FloatField
|
|
|
- db.alter_column("test_altermyd", "eggs", models.TextField(null=True))
|
|
|
- db.delete_table("test_altermyd")
|
|
|
-
|
|
|
- def test_alter_column_postgres_multiword(self):
|
|
|
- """
|
|
|
- Tests altering columns with multiple words in Postgres types (issue #125)
|
|
|
- e.g. 'datetime with time zone', look at django/db/backends/postgresql/creation.py
|
|
|
- """
|
|
|
- db.create_table("test_multiword", [
|
|
|
- ('col_datetime', models.DateTimeField(null=True)),
|
|
|
- ('col_integer', models.PositiveIntegerField(null=True)),
|
|
|
- ('col_smallint', models.PositiveSmallIntegerField(null=True)),
|
|
|
- ('col_float', models.FloatField(null=True)),
|
|
|
- ])
|
|
|
-
|
|
|
- # test if 'double precision' is preserved
|
|
|
- db.alter_column('test_multiword', 'col_float', models.FloatField('float', null=True))
|
|
|
-
|
|
|
- # test if 'CHECK ("%(column)s" >= 0)' is stripped
|
|
|
- db.alter_column('test_multiword', 'col_integer', models.PositiveIntegerField(null=True))
|
|
|
- db.alter_column('test_multiword', 'col_smallint', models.PositiveSmallIntegerField(null=True))
|
|
|
-
|
|
|
- # test if 'with timezone' is preserved
|
|
|
- if db.backend_name == "postgres":
|
|
|
- db.execute("INSERT INTO test_multiword (col_datetime) VALUES ('2009-04-24 14:20:55+02')")
|
|
|
- db.alter_column('test_multiword', 'col_datetime', models.DateTimeField(auto_now=True))
|
|
|
- assert db.execute("SELECT col_datetime = '2009-04-24 14:20:55+02' FROM test_multiword")[0][0]
|
|
|
-
|
|
|
- db.delete_table("test_multiword")
|
|
|
-
|
|
|
- @skipUnless(db.has_check_constraints, 'Only applies to databases that '
|
|
|
- 'support CHECK constraints.')
|
|
|
- def test_alter_constraints(self):
|
|
|
- """
|
|
|
- Tests that going from a PostiveIntegerField to an IntegerField drops
|
|
|
- the constraint on the database.
|
|
|
- """
|
|
|
- # Make the test table
|
|
|
- db.create_table("test_alterc", [
|
|
|
- ('num', models.PositiveIntegerField()),
|
|
|
- ])
|
|
|
- db.execute_deferred_sql()
|
|
|
- # Add in some test values
|
|
|
- db.execute("INSERT INTO test_alterc (num) VALUES (1)")
|
|
|
- db.execute("INSERT INTO test_alterc (num) VALUES (2)")
|
|
|
- # Ensure that adding a negative number is bad
|
|
|
- db.commit_transaction()
|
|
|
- db.start_transaction()
|
|
|
- try:
|
|
|
- db.execute("INSERT INTO test_alterc (num) VALUES (-3)")
|
|
|
- except:
|
|
|
- db.rollback_transaction()
|
|
|
- else:
|
|
|
- self.fail("Could insert a negative integer into a PositiveIntegerField.")
|
|
|
- # Alter it to a normal IntegerField
|
|
|
- db.alter_column("test_alterc", "num", models.IntegerField())
|
|
|
- db.execute_deferred_sql()
|
|
|
- # It should now work
|
|
|
- db.execute("INSERT INTO test_alterc (num) VALUES (-3)")
|
|
|
- db.delete_table("test_alterc")
|
|
|
- # We need to match up for tearDown
|
|
|
- db.start_transaction()
|
|
|
-
|
|
|
- @skipIf(db.backend_name == "sqlite3", "SQLite backend doesn't support this "
|
|
|
- "yet.")
|
|
|
- def test_unique(self):
|
|
|
- """
|
|
|
- Tests creating/deleting unique constraints.
|
|
|
- """
|
|
|
- db.create_table("test_unique2", [
|
|
|
- ('id', models.AutoField(primary_key=True)),
|
|
|
- ])
|
|
|
- db.create_table("test_unique", [
|
|
|
- ('spam', models.BooleanField(default=False)),
|
|
|
- ('eggs', models.IntegerField()),
|
|
|
- ('ham', models.ForeignKey(db.mock_model('Unique2', 'test_unique2'))),
|
|
|
- ])
|
|
|
- db.execute_deferred_sql()
|
|
|
- # Add a constraint
|
|
|
- db.create_unique("test_unique", ["spam"])
|
|
|
- db.execute_deferred_sql()
|
|
|
- # Shouldn't do anything during dry-run
|
|
|
- db.dry_run = True
|
|
|
- db.delete_unique("test_unique", ["spam"])
|
|
|
- db.dry_run = False
|
|
|
- db.delete_unique("test_unique", ["spam"])
|
|
|
- db.create_unique("test_unique", ["spam"])
|
|
|
- # Special preparations for Sql Server
|
|
|
- if db.backend_name == "pyodbc":
|
|
|
- db.execute("SET IDENTITY_INSERT test_unique2 ON;")
|
|
|
- db.execute("INSERT INTO test_unique2 (id) VALUES (1)")
|
|
|
- db.execute("INSERT INTO test_unique2 (id) VALUES (2)")
|
|
|
- db.commit_transaction()
|
|
|
- db.start_transaction()
|
|
|
-
|
|
|
-
|
|
|
- # Test it works
|
|
|
- TRUE = (True,)
|
|
|
- FALSE = (False,)
|
|
|
- db.execute("INSERT INTO test_unique (spam, eggs, ham_id) VALUES (%s, 0, 1)", TRUE)
|
|
|
- db.execute("INSERT INTO test_unique (spam, eggs, ham_id) VALUES (%s, 1, 2)", FALSE)
|
|
|
- try:
|
|
|
- db.execute("INSERT INTO test_unique (spam, eggs, ham_id) VALUES (%s, 2, 1)", FALSE)
|
|
|
- except:
|
|
|
- db.rollback_transaction()
|
|
|
- else:
|
|
|
- self.fail("Could insert non-unique item.")
|
|
|
-
|
|
|
- # Drop that, add one only on eggs
|
|
|
- db.delete_unique("test_unique", ["spam"])
|
|
|
- db.execute("DELETE FROM test_unique")
|
|
|
- db.create_unique("test_unique", ["eggs"])
|
|
|
- db.start_transaction()
|
|
|
-
|
|
|
- # Test similarly
|
|
|
- db.execute("INSERT INTO test_unique (spam, eggs, ham_id) VALUES (%s, 0, 1)", TRUE)
|
|
|
- db.execute("INSERT INTO test_unique (spam, eggs, ham_id) VALUES (%s, 1, 2)", FALSE)
|
|
|
- try:
|
|
|
- db.execute("INSERT INTO test_unique (spam, eggs, ham_id) VALUES (%s, 1, 1)", TRUE)
|
|
|
- except:
|
|
|
- db.rollback_transaction()
|
|
|
- else:
|
|
|
- self.fail("Could insert non-unique item.")
|
|
|
-
|
|
|
- # Drop those, test combined constraints
|
|
|
- db.delete_unique("test_unique", ["eggs"])
|
|
|
- db.execute("DELETE FROM test_unique")
|
|
|
- db.create_unique("test_unique", ["spam", "eggs", "ham_id"])
|
|
|
- db.start_transaction()
|
|
|
- # Test similarly
|
|
|
- db.execute("INSERT INTO test_unique (spam, eggs, ham_id) VALUES (%s, 0, 1)", TRUE)
|
|
|
- db.execute("INSERT INTO test_unique (spam, eggs, ham_id) VALUES (%s, 1, 1)", FALSE)
|
|
|
- try:
|
|
|
- db.execute("INSERT INTO test_unique (spam, eggs, ham_id) VALUES (%s, 0, 1)", TRUE)
|
|
|
- except:
|
|
|
- db.rollback_transaction()
|
|
|
- else:
|
|
|
- self.fail("Could insert non-unique pair.")
|
|
|
- db.delete_unique("test_unique", ["spam", "eggs", "ham_id"])
|
|
|
- db.start_transaction()
|
|
|
-
|
|
|
- def test_alter_unique(self):
|
|
|
- """
|
|
|
- Tests that unique constraints are not affected when
|
|
|
- altering columns (that's handled by create_/delete_unique)
|
|
|
- """
|
|
|
- db.create_table("test_alter_unique", [
|
|
|
- ('spam', models.IntegerField()),
|
|
|
- ('eggs', models.IntegerField(unique=True)),
|
|
|
- ])
|
|
|
- db.execute_deferred_sql()
|
|
|
-
|
|
|
- # Make sure the unique constraint is created
|
|
|
- db.execute('INSERT INTO test_alter_unique (spam, eggs) VALUES (0, 42)')
|
|
|
- db.commit_transaction()
|
|
|
- db.start_transaction()
|
|
|
- try:
|
|
|
- db.execute("INSERT INTO test_alter_unique (spam, eggs) VALUES (1, 42)")
|
|
|
- except:
|
|
|
- pass
|
|
|
- else:
|
|
|
- self.fail("Could insert the same integer twice into a unique field.")
|
|
|
- db.rollback_transaction()
|
|
|
-
|
|
|
- # Alter without unique=True (should not affect anything)
|
|
|
- db.alter_column("test_alter_unique", "eggs", models.IntegerField())
|
|
|
-
|
|
|
- # Insertion should still fail
|
|
|
- db.start_transaction()
|
|
|
- try:
|
|
|
- db.execute("INSERT INTO test_alter_unique (spam, eggs) VALUES (1, 42)")
|
|
|
- except:
|
|
|
- pass
|
|
|
- else:
|
|
|
- self.fail("Could insert the same integer twice into a unique field after alter_column with unique=False.")
|
|
|
- db.rollback_transaction()
|
|
|
-
|
|
|
- # Delete the unique index/constraint
|
|
|
- if db.backend_name != "sqlite3":
|
|
|
- db.delete_unique("test_alter_unique", ["eggs"])
|
|
|
- db.delete_table("test_alter_unique")
|
|
|
- db.start_transaction()
|
|
|
-
|
|
|
- # Test multi-field constraint
|
|
|
- db.create_table("test_alter_unique2", [
|
|
|
- ('spam', models.IntegerField()),
|
|
|
- ('eggs', models.IntegerField()),
|
|
|
- ])
|
|
|
- db.create_unique('test_alter_unique2', ('spam', 'eggs'))
|
|
|
- db.execute_deferred_sql()
|
|
|
- db.execute('INSERT INTO test_alter_unique2 (spam, eggs) VALUES (0, 42)')
|
|
|
- db.commit_transaction()
|
|
|
- # Verify that constraint works
|
|
|
- db.start_transaction()
|
|
|
- try:
|
|
|
- db.execute("INSERT INTO test_alter_unique2 (spam, eggs) VALUES (1, 42)")
|
|
|
- except:
|
|
|
- self.fail("Looks like multi-field unique constraint applied to only one field.")
|
|
|
- db.rollback_transaction()
|
|
|
- db.start_transaction()
|
|
|
- try:
|
|
|
- db.execute("INSERT INTO test_alter_unique2 (spam, eggs) VALUES (0, 43)")
|
|
|
- except:
|
|
|
- self.fail("Looks like multi-field unique constraint applied to only one field.")
|
|
|
- db.rollback_transaction()
|
|
|
- db.start_transaction()
|
|
|
- try:
|
|
|
- db.execute("INSERT INTO test_alter_unique2 (spam, eggs) VALUES (0, 42)")
|
|
|
- except:
|
|
|
- pass
|
|
|
- else:
|
|
|
- self.fail("Could insert the same pair twice into unique-together fields.")
|
|
|
- db.rollback_transaction()
|
|
|
- # Altering one column should not drop or modify multi-column constraint
|
|
|
- db.alter_column("test_alter_unique2", "eggs", models.PositiveIntegerField())
|
|
|
- db.start_transaction()
|
|
|
- try:
|
|
|
- db.execute("INSERT INTO test_alter_unique2 (spam, eggs) VALUES (1, 42)")
|
|
|
- except:
|
|
|
- self.fail("Altering one column broken multi-column unique constraint.")
|
|
|
- db.rollback_transaction()
|
|
|
- db.start_transaction()
|
|
|
- try:
|
|
|
- db.execute("INSERT INTO test_alter_unique2 (spam, eggs) VALUES (0, 43)")
|
|
|
- except:
|
|
|
- self.fail("Altering one column broken multi-column unique constraint.")
|
|
|
- db.rollback_transaction()
|
|
|
- db.start_transaction()
|
|
|
- try:
|
|
|
- db.execute("INSERT INTO test_alter_unique2 (spam, eggs) VALUES (0, 42)")
|
|
|
- except:
|
|
|
- pass
|
|
|
- else:
|
|
|
- self.fail("Could insert the same pair twice into unique-together fields after alter_column with unique=False.")
|
|
|
- db.rollback_transaction()
|
|
|
- db.delete_table("test_alter_unique2")
|
|
|
- db.start_transaction()
|
|
|
-
|
|
|
- def test_capitalised_constraints(self):
|
|
|
- """
|
|
|
- Under PostgreSQL at least, capitalised constraints must be quoted.
|
|
|
- """
|
|
|
- db.create_table("test_capconst", [
|
|
|
- ('SOMECOL', models.PositiveIntegerField(primary_key=True)),
|
|
|
- ])
|
|
|
- # Alter it so it's not got the check constraint
|
|
|
- db.alter_column("test_capconst", "SOMECOL", models.IntegerField())
|
|
|
-
|
|
|
- def test_text_default(self):
|
|
|
- """
|
|
|
- MySQL cannot have blank defaults on TEXT columns.
|
|
|
- """
|
|
|
- db.create_table("test_textdef", [
|
|
|
- ('textcol', models.TextField(blank=True)),
|
|
|
- ])
|
|
|
-
|
|
|
- def test_text_to_char(self):
|
|
|
- """
|
|
|
- On Oracle, you can't simply ALTER TABLE MODIFY a textfield to a charfield
|
|
|
- """
|
|
|
- value = "kawabanga"
|
|
|
- db.create_table("test_text_to_char", [
|
|
|
- ('textcol', models.TextField()),
|
|
|
- ])
|
|
|
- db.execute_deferred_sql()
|
|
|
- db.execute("INSERT INTO test_text_to_char VALUES (%s)", [value])
|
|
|
- db.alter_column("test_text_to_char", "textcol", models.CharField(max_length=100))
|
|
|
- db.execute_deferred_sql()
|
|
|
- after = db.execute("select * from test_text_to_char")[0][0]
|
|
|
- self.assertEqual(value, after, "Change from text to char altered value [ %r != %r ]" % (value, after))
|
|
|
-
|
|
|
- def test_char_to_text(self):
|
|
|
- """
|
|
|
- On Oracle, you can't simply ALTER TABLE MODIFY a charfield to a textfield either
|
|
|
- """
|
|
|
- value = "agnabawak"
|
|
|
- db.create_table("test_char_to_text", [
|
|
|
- ('textcol', models.CharField(max_length=100)),
|
|
|
- ])
|
|
|
- db.execute_deferred_sql()
|
|
|
- db.execute("INSERT INTO test_char_to_text VALUES (%s)", [value])
|
|
|
- db.alter_column("test_char_to_text", "textcol", models.TextField())
|
|
|
- db.execute_deferred_sql()
|
|
|
- after = db.execute("select * from test_char_to_text")[0][0]
|
|
|
- after = text_type(after) # Oracle text fields return a sort of lazy string -- force evaluation
|
|
|
- self.assertEqual(value, after, "Change from char to text altered value [ %r != %r ]" % (value, after))
|
|
|
-
|
|
|
- @skipUnless(db.raises_default_errors, 'This database does not raise errors on missing defaults.')
|
|
|
- def test_datetime_default(self):
|
|
|
- """
|
|
|
- Test that defaults are correctly not created for datetime columns
|
|
|
- """
|
|
|
- end_of_world = datetime.datetime(2012, 12, 21, 0, 0, 1)
|
|
|
-
|
|
|
- try:
|
|
|
- from django.utils import timezone
|
|
|
- except ImportError:
|
|
|
- pass
|
|
|
- else:
|
|
|
- from django.conf import settings
|
|
|
- if getattr(settings, 'USE_TZ', False):
|
|
|
- end_of_world = end_of_world.replace(tzinfo=timezone.utc)
|
|
|
-
|
|
|
- db.create_table("test_datetime_def", [
|
|
|
- ('col0', models.IntegerField(null=True)),
|
|
|
- ('col1', models.DateTimeField(default=end_of_world)),
|
|
|
- ('col2', models.DateTimeField(null=True)),
|
|
|
- ])
|
|
|
- db.execute_deferred_sql()
|
|
|
- # insert a row
|
|
|
- db.execute("INSERT INTO test_datetime_def (col0, col1, col2) values (null,%s,null)", [end_of_world])
|
|
|
- db.alter_column("test_datetime_def", "col2", models.DateTimeField(default=end_of_world))
|
|
|
- db.add_column("test_datetime_def", "col3", models.DateTimeField(default=end_of_world))
|
|
|
- db.execute_deferred_sql()
|
|
|
- db.commit_transaction()
|
|
|
- # In the single existing row, we now expect col1=col2=col3=end_of_world...
|
|
|
- db.start_transaction()
|
|
|
- ends = db.execute("select col1,col2,col3 from test_datetime_def")[0]
|
|
|
- self.failUnlessEqual(len(ends), 3)
|
|
|
- for e in ends:
|
|
|
- self.failUnlessEqual(e, end_of_world)
|
|
|
- db.commit_transaction()
|
|
|
- # ...but there should not be a default in the database for col1 or col3
|
|
|
- for cols in ["col1,col2", "col2,col3"]:
|
|
|
- db.start_transaction()
|
|
|
- statement = "insert into test_datetime_def (col0,%s) values (null,%%s,%%s)" % cols
|
|
|
- self.assertRaises(
|
|
|
- IntegrityError,
|
|
|
- db.execute, statement, [end_of_world, end_of_world]
|
|
|
- )
|
|
|
- db.rollback_transaction()
|
|
|
-
|
|
|
- db.start_transaction() # To preserve the sanity and semantics of this test class
|
|
|
-
|
|
|
- def test_add_unique_fk(self):
|
|
|
- """
|
|
|
- Test adding a ForeignKey with unique=True or a OneToOneField
|
|
|
- """
|
|
|
- db.create_table("test_add_unique_fk", [
|
|
|
- ('spam', models.BooleanField(default=False))
|
|
|
- ])
|
|
|
-
|
|
|
- db.add_column("test_add_unique_fk", "mock1", models.ForeignKey(db.mock_model('User', 'auth_user'), null=True, unique=True))
|
|
|
- db.add_column("test_add_unique_fk", "mock2", models.OneToOneField(db.mock_model('User', 'auth_user'), null=True))
|
|
|
- db.execute_deferred_sql()
|
|
|
-
|
|
|
- db.delete_table("test_add_unique_fk")
|
|
|
-
|
|
|
- @skipUnless(db.has_check_constraints, 'Only applies to databases that '
|
|
|
- 'support CHECK constraints.')
|
|
|
- def test_column_constraint(self):
|
|
|
- """
|
|
|
- Tests that the value constraint of PositiveIntegerField is enforced on
|
|
|
- the database level.
|
|
|
- """
|
|
|
- db.create_table("test_column_constraint", [
|
|
|
- ('spam', models.PositiveIntegerField()),
|
|
|
- ])
|
|
|
- db.execute_deferred_sql()
|
|
|
-
|
|
|
- # Make sure we can't insert negative values
|
|
|
- db.commit_transaction()
|
|
|
- db.start_transaction()
|
|
|
- try:
|
|
|
- db.execute("INSERT INTO test_column_constraint VALUES (-42)")
|
|
|
- except:
|
|
|
- pass
|
|
|
- else:
|
|
|
- self.fail("Could insert a negative value into a PositiveIntegerField.")
|
|
|
- db.rollback_transaction()
|
|
|
-
|
|
|
- # remove constraint
|
|
|
- db.alter_column("test_column_constraint", "spam", models.IntegerField())
|
|
|
- db.execute_deferred_sql()
|
|
|
- # make sure the insertion works now
|
|
|
- db.execute('INSERT INTO test_column_constraint VALUES (-42)')
|
|
|
- db.execute('DELETE FROM test_column_constraint')
|
|
|
-
|
|
|
- # add it back again
|
|
|
- db.alter_column("test_column_constraint", "spam", models.PositiveIntegerField())
|
|
|
- db.execute_deferred_sql()
|
|
|
- # it should fail again
|
|
|
- db.start_transaction()
|
|
|
- try:
|
|
|
- db.execute("INSERT INTO test_column_constraint VALUES (-42)")
|
|
|
- except:
|
|
|
- pass
|
|
|
- else:
|
|
|
- self.fail("Could insert a negative value after changing an IntegerField to a PositiveIntegerField.")
|
|
|
- db.rollback_transaction()
|
|
|
-
|
|
|
- db.delete_table("test_column_constraint")
|
|
|
- db.start_transaction()
|
|
|
-
|
|
|
- def test_sql_defaults(self):
|
|
|
- """
|
|
|
- Test that sql default value is correct for non-string field types.
|
|
|
- Datetimes are handled in test_datetime_default.
|
|
|
- """
|
|
|
-
|
|
|
- class CustomField(with_metaclass(models.SubfieldBase, models.CharField)):
|
|
|
- description = 'CustomField'
|
|
|
- def get_default(self):
|
|
|
- if self.has_default():
|
|
|
- if callable(self.default):
|
|
|
- return self.default()
|
|
|
- return self.default
|
|
|
- return super(CustomField, self).get_default()
|
|
|
- def get_prep_value(self, value):
|
|
|
- if not value:
|
|
|
- return value
|
|
|
- return ','.join(map(str, value))
|
|
|
- def to_python(self, value):
|
|
|
- if not value or isinstance(value, list):
|
|
|
- return value
|
|
|
- return list(map(int, value.split(',')))
|
|
|
-
|
|
|
- false_value = db.has_booleans and 'False' or '0'
|
|
|
- defaults = (
|
|
|
- (models.CharField(default='sukasuka'), 'DEFAULT \'sukasuka'),
|
|
|
- (models.BooleanField(default=False), 'DEFAULT %s' % false_value),
|
|
|
- (models.IntegerField(default=42), 'DEFAULT 42'),
|
|
|
- (CustomField(default=[2012, 2018, 2021, 2036]), 'DEFAULT \'2012,2018,2021,2036')
|
|
|
- )
|
|
|
- for field, sql_test_str in defaults:
|
|
|
- sql = db.column_sql('fish', 'YAAAAAAZ', field)
|
|
|
- if sql_test_str not in sql:
|
|
|
- self.fail("default sql value was not properly generated for field %r.\nSql was %s" % (field, sql))
|
|
|
-
|
|
|
- def test_make_added_foreign_key_not_null(self):
|
|
|
- # Table for FK to target
|
|
|
- User = db.mock_model(model_name='User', db_table='auth_user', db_tablespace='', pk_field_name='id', pk_field_type=models.AutoField, pk_field_args=[], pk_field_kwargs={})
|
|
|
- # Table with no foreign key
|
|
|
- db.create_table("test_fk", [
|
|
|
- ('eggs', models.IntegerField()),
|
|
|
- ])
|
|
|
- db.execute_deferred_sql()
|
|
|
-
|
|
|
- # Add foreign key
|
|
|
- db.add_column("test_fk", 'foreik', models.ForeignKey(User, null=True))
|
|
|
- db.execute_deferred_sql()
|
|
|
-
|
|
|
- # Make the FK not null
|
|
|
- db.alter_column("test_fk", "foreik_id", models.ForeignKey(User))
|
|
|
- db.execute_deferred_sql()
|
|
|
-
|
|
|
- def test_make_foreign_key_null(self):
|
|
|
- # Table for FK to target
|
|
|
- User = db.mock_model(model_name='User', db_table='auth_user', db_tablespace='', pk_field_name='id', pk_field_type=models.AutoField, pk_field_args=[], pk_field_kwargs={})
|
|
|
- # Table with no foreign key
|
|
|
- db.create_table("test_make_fk_null", [
|
|
|
- ('eggs', models.IntegerField()),
|
|
|
- ('foreik', models.ForeignKey(User))
|
|
|
- ])
|
|
|
- db.execute_deferred_sql()
|
|
|
-
|
|
|
- # Make the FK null
|
|
|
- db.alter_column("test_make_fk_null", "foreik_id", models.ForeignKey(User, null=True))
|
|
|
- db.execute_deferred_sql()
|
|
|
-
|
|
|
- def test_change_foreign_key_target(self):
|
|
|
- # Tables for FK to target
|
|
|
- User = db.mock_model(model_name='User', db_table='auth_user', db_tablespace='', pk_field_name='id', pk_field_type=models.AutoField, pk_field_args=[], pk_field_kwargs={})
|
|
|
- db.create_table("test_fk_changed_target", [
|
|
|
- ('eggs', models.IntegerField(primary_key=True)),
|
|
|
- ])
|
|
|
- Egg = db.mock_model(model_name='Egg', db_table='test_fk_changed_target', db_tablespace='', pk_field_name='eggs', pk_field_type=models.AutoField, pk_field_args=[], pk_field_kwargs={})
|
|
|
- # Table with a foreign key to the wrong table
|
|
|
- db.create_table("test_fk_changing", [
|
|
|
- ('egg', models.ForeignKey(User, null=True)),
|
|
|
- ])
|
|
|
- db.execute_deferred_sql()
|
|
|
-
|
|
|
- # Change foreign key pointing
|
|
|
- db.alter_column("test_fk_changing", "egg_id", models.ForeignKey(Egg, null=True))
|
|
|
- db.execute_deferred_sql()
|
|
|
-
|
|
|
- # Test that it is pointing at the right table now
|
|
|
- try:
|
|
|
- non_user_id = db.execute("SELECT MAX(id) FROM auth_user")[0][0] + 1
|
|
|
- except (TypeError, IndexError):
|
|
|
- # Got a "None" or no records, treat as 0
|
|
|
- non_user_id = 17
|
|
|
- db.execute("INSERT INTO test_fk_changed_target (eggs) VALUES (%s)", [non_user_id])
|
|
|
- db.execute("INSERT INTO test_fk_changing (egg_id) VALUES (%s)", [non_user_id])
|
|
|
- db.commit_transaction()
|
|
|
- db.start_transaction() # The test framework expects tests to end in transaction
|
|
|
-
|
|
|
- def test_alter_double_indexed_column(self):
|
|
|
- # Table for FK to target
|
|
|
- User = db.mock_model(model_name='User', db_table='auth_user', db_tablespace='', pk_field_name='id', pk_field_type=models.AutoField, pk_field_args=[], pk_field_kwargs={})
|
|
|
- # Table with no foreign key
|
|
|
- db.create_table("test_2indexed", [
|
|
|
- ('eggs', models.IntegerField()),
|
|
|
- ('foreik', models.ForeignKey(User))
|
|
|
- ])
|
|
|
- db.create_unique("test_2indexed", ["eggs", "foreik_id"])
|
|
|
- db.execute_deferred_sql()
|
|
|
-
|
|
|
- # Make the FK null
|
|
|
- db.alter_column("test_2indexed", "foreik_id", models.ForeignKey(User, null=True))
|
|
|
- db.execute_deferred_sql()
|
|
|
-
|
|
|
-class TestCacheGeneric(unittest.TestCase):
|
|
|
- base_ops_cls = generic.DatabaseOperations
|
|
|
- def setUp(self):
|
|
|
- class CacheOps(self.base_ops_cls):
|
|
|
- def __init__(self):
|
|
|
- self._constraint_cache = {}
|
|
|
- self.cache_filled = 0
|
|
|
- self.settings = {'NAME': 'db'}
|
|
|
-
|
|
|
- def _fill_constraint_cache(self, db, table):
|
|
|
- self.cache_filled += 1
|
|
|
- self._constraint_cache.setdefault(db, {})
|
|
|
- self._constraint_cache[db].setdefault(table, {})
|
|
|
-
|
|
|
- @generic.invalidate_table_constraints
|
|
|
- def clear_con(self, table):
|
|
|
- pass
|
|
|
-
|
|
|
- @generic.copy_column_constraints
|
|
|
- def cp_column(self, table, column_old, column_new):
|
|
|
- pass
|
|
|
-
|
|
|
- @generic.delete_column_constraints
|
|
|
- def rm_column(self, table, column):
|
|
|
- pass
|
|
|
-
|
|
|
- @generic.copy_column_constraints
|
|
|
- @generic.delete_column_constraints
|
|
|
- def mv_column(self, table, column_old, column_new):
|
|
|
- pass
|
|
|
-
|
|
|
- def _get_setting(self, attr):
|
|
|
- return self.settings[attr]
|
|
|
- self.CacheOps = CacheOps
|
|
|
-
|
|
|
- def test_cache(self):
|
|
|
- ops = self.CacheOps()
|
|
|
- self.assertEqual(0, ops.cache_filled)
|
|
|
- self.assertFalse(ops.lookup_constraint('db', 'table'))
|
|
|
- self.assertEqual(1, ops.cache_filled)
|
|
|
- self.assertFalse(ops.lookup_constraint('db', 'table'))
|
|
|
- self.assertEqual(1, ops.cache_filled)
|
|
|
- ops.clear_con('table')
|
|
|
- self.assertEqual(1, ops.cache_filled)
|
|
|
- self.assertFalse(ops.lookup_constraint('db', 'table'))
|
|
|
- self.assertEqual(2, ops.cache_filled)
|
|
|
- self.assertFalse(ops.lookup_constraint('db', 'table', 'column'))
|
|
|
- self.assertEqual(2, ops.cache_filled)
|
|
|
-
|
|
|
- cache = ops._constraint_cache
|
|
|
- cache['db']['table']['column'] = 'constraint'
|
|
|
- self.assertEqual('constraint', ops.lookup_constraint('db', 'table', 'column'))
|
|
|
- self.assertEqual([('column', 'constraint')], ops.lookup_constraint('db', 'table'))
|
|
|
- self.assertEqual(2, ops.cache_filled)
|
|
|
-
|
|
|
- # invalidate_table_constraints
|
|
|
- ops.clear_con('new_table')
|
|
|
- self.assertEqual('constraint', ops.lookup_constraint('db', 'table', 'column'))
|
|
|
- self.assertEqual(2, ops.cache_filled)
|
|
|
-
|
|
|
- self.assertFalse(ops.lookup_constraint('db', 'new_table'))
|
|
|
- self.assertEqual(3, ops.cache_filled)
|
|
|
-
|
|
|
- # delete_column_constraints
|
|
|
- cache['db']['table']['column'] = 'constraint'
|
|
|
- self.assertEqual('constraint', ops.lookup_constraint('db', 'table', 'column'))
|
|
|
- ops.rm_column('table', 'column')
|
|
|
- self.assertEqual([], ops.lookup_constraint('db', 'table', 'column'))
|
|
|
- self.assertEqual([], ops.lookup_constraint('db', 'table', 'noexist_column'))
|
|
|
-
|
|
|
- # copy_column_constraints
|
|
|
- cache['db']['table']['column'] = 'constraint'
|
|
|
- self.assertEqual('constraint', ops.lookup_constraint('db', 'table', 'column'))
|
|
|
- ops.cp_column('table', 'column', 'column_new')
|
|
|
- self.assertEqual('constraint', ops.lookup_constraint('db', 'table', 'column_new'))
|
|
|
- self.assertEqual('constraint', ops.lookup_constraint('db', 'table', 'column'))
|
|
|
-
|
|
|
- # copy + delete
|
|
|
- cache['db']['table']['column'] = 'constraint'
|
|
|
- self.assertEqual('constraint', ops.lookup_constraint('db', 'table', 'column'))
|
|
|
- ops.mv_column('table', 'column', 'column_new')
|
|
|
- self.assertEqual('constraint', ops.lookup_constraint('db', 'table', 'column_new'))
|
|
|
- self.assertEqual([], ops.lookup_constraint('db', 'table', 'column'))
|
|
|
-
|
|
|
- def test_valid(self):
|
|
|
- ops = self.CacheOps()
|
|
|
- # none of these should vivify a table into a valid state
|
|
|
- self.assertFalse(ops._is_valid_cache('db', 'table'))
|
|
|
- self.assertFalse(ops._is_valid_cache('db', 'table'))
|
|
|
- ops.clear_con('table')
|
|
|
- self.assertFalse(ops._is_valid_cache('db', 'table'))
|
|
|
- ops.rm_column('table', 'column')
|
|
|
- self.assertFalse(ops._is_valid_cache('db', 'table'))
|
|
|
-
|
|
|
- # these should change the cache state
|
|
|
- ops.lookup_constraint('db', 'table')
|
|
|
- self.assertTrue(ops._is_valid_cache('db', 'table'))
|
|
|
- ops.lookup_constraint('db', 'table', 'column')
|
|
|
- self.assertTrue(ops._is_valid_cache('db', 'table'))
|
|
|
- ops.clear_con('table')
|
|
|
- self.assertFalse(ops._is_valid_cache('db', 'table'))
|
|
|
-
|
|
|
- def test_valid_implementation(self):
|
|
|
- # generic fills the cache on a per-table basis
|
|
|
- ops = self.CacheOps()
|
|
|
- self.assertFalse(ops._is_valid_cache('db', 'table'))
|
|
|
- self.assertFalse(ops._is_valid_cache('db', 'other_table'))
|
|
|
- ops.lookup_constraint('db', 'table')
|
|
|
- self.assertTrue(ops._is_valid_cache('db', 'table'))
|
|
|
- self.assertFalse(ops._is_valid_cache('db', 'other_table'))
|
|
|
- ops.lookup_constraint('db', 'other_table')
|
|
|
- self.assertTrue(ops._is_valid_cache('db', 'table'))
|
|
|
- self.assertTrue(ops._is_valid_cache('db', 'other_table'))
|
|
|
- ops.clear_con('table')
|
|
|
- self.assertFalse(ops._is_valid_cache('db', 'table'))
|
|
|
- self.assertTrue(ops._is_valid_cache('db', 'other_table'))
|
|
|
-
|
|
|
-if mysql:
|
|
|
- class TestCacheMysql(TestCacheGeneric):
|
|
|
- base_ops_cls = mysql.DatabaseOperations
|
|
|
-
|
|
|
- def test_valid_implementation(self):
|
|
|
- # mysql fills the cache on a per-db basis
|
|
|
- ops = self.CacheOps()
|
|
|
- self.assertFalse(ops._is_valid_cache('db', 'table'))
|
|
|
- self.assertFalse(ops._is_valid_cache('db', 'other_table'))
|
|
|
- ops.lookup_constraint('db', 'table')
|
|
|
- self.assertTrue(ops._is_valid_cache('db', 'table'))
|
|
|
- self.assertTrue(ops._is_valid_cache('db', 'other_table'))
|
|
|
- ops.lookup_constraint('db', 'other_table')
|
|
|
- self.assertTrue(ops._is_valid_cache('db', 'table'))
|
|
|
- self.assertTrue(ops._is_valid_cache('db', 'other_table'))
|
|
|
- ops.clear_con('table')
|
|
|
- self.assertFalse(ops._is_valid_cache('db', 'table'))
|
|
|
- self.assertTrue(ops._is_valid_cache('db', 'other_table'))
|