2013-08-23 11:07:43 +00:00
|
|
|
from django.test import TransactionTestCase
|
2013-07-25 12:52:35 +00:00
|
|
|
from django.db import connection
|
|
|
|
|
|
|
|
|
2013-08-23 11:07:43 +00:00
|
|
|
class MigrationTestBase(TransactionTestCase):
|
2013-07-25 12:52:35 +00:00
|
|
|
"""
|
|
|
|
Contains an extended set of asserts for testing migrations and schema operations.
|
|
|
|
"""
|
|
|
|
|
2013-08-23 11:07:43 +00:00
|
|
|
available_apps = ["migrations"]
|
|
|
|
|
2014-01-09 15:05:15 +00:00
|
|
|
def get_table_description(self, table):
|
|
|
|
with connection.cursor() as cursor:
|
|
|
|
return connection.introspection.get_table_description(cursor, table)
|
|
|
|
|
2013-07-25 12:52:35 +00:00
|
|
|
def assertTableExists(self, table):
|
2014-01-09 15:05:15 +00:00
|
|
|
with connection.cursor() as cursor:
|
|
|
|
self.assertIn(table, connection.introspection.get_table_list(cursor))
|
2013-07-25 12:52:35 +00:00
|
|
|
|
|
|
|
def assertTableNotExists(self, table):
|
2014-01-09 15:05:15 +00:00
|
|
|
with connection.cursor() as cursor:
|
|
|
|
self.assertNotIn(table, connection.introspection.get_table_list(cursor))
|
2013-07-25 12:52:35 +00:00
|
|
|
|
|
|
|
def assertColumnExists(self, table, column):
|
2014-01-09 15:05:15 +00:00
|
|
|
self.assertIn(column, [c.name for c in self.get_table_description(table)])
|
2013-07-25 12:52:35 +00:00
|
|
|
|
|
|
|
def assertColumnNotExists(self, table, column):
|
2014-01-09 15:05:15 +00:00
|
|
|
self.assertNotIn(column, [c.name for c in self.get_table_description(table)])
|
2013-07-25 12:52:35 +00:00
|
|
|
|
|
|
|
def assertColumnNull(self, table, column):
|
2014-01-09 15:05:15 +00:00
|
|
|
self.assertEqual([c.null_ok for c in self.get_table_description(table) if c.name == column][0], True)
|
2013-07-25 12:52:35 +00:00
|
|
|
|
|
|
|
def assertColumnNotNull(self, table, column):
|
2014-01-09 15:05:15 +00:00
|
|
|
self.assertEqual([c.null_ok for c in self.get_table_description(table) if c.name == column][0], False)
|
2013-07-25 12:52:35 +00:00
|
|
|
|
|
|
|
def assertIndexExists(self, table, columns, value=True):
|
2014-01-09 15:05:15 +00:00
|
|
|
with connection.cursor() as cursor:
|
|
|
|
self.assertEqual(
|
|
|
|
value,
|
|
|
|
any(
|
|
|
|
c["index"]
|
|
|
|
for c in connection.introspection.get_constraints(cursor, table).values()
|
|
|
|
if c['columns'] == list(columns)
|
|
|
|
),
|
|
|
|
)
|
2013-07-25 12:52:35 +00:00
|
|
|
|
|
|
|
def assertIndexNotExists(self, table, columns):
|
|
|
|
return self.assertIndexExists(table, columns, False)
|
2014-06-08 01:17:31 +00:00
|
|
|
|
|
|
|
def assertFKExists(self, table, columns, to, value=True):
|
|
|
|
with connection.cursor() as cursor:
|
|
|
|
self.assertEqual(
|
|
|
|
value,
|
|
|
|
any(
|
|
|
|
c["foreign_key"] == to
|
|
|
|
for c in connection.introspection.get_constraints(cursor, table).values()
|
|
|
|
if c['columns'] == list(columns)
|
|
|
|
),
|
|
|
|
)
|
|
|
|
|
|
|
|
def assertFKNotExists(self, table, columns, to, value=True):
|
|
|
|
return self.assertFKExists(table, columns, to, False)
|