Skip to content

Commit

Permalink
More schema test fixing
Browse files Browse the repository at this point in the history
  • Loading branch information
andrewgodwin committed Sep 17, 2012
1 parent 9313dea commit d0b3536
Show file tree
Hide file tree
Showing 3 changed files with 63 additions and 52 deletions.
4 changes: 3 additions & 1 deletion django/core/management/commands/syncdb.py
Expand Up @@ -71,9 +71,11 @@ def handle_noargs(self, **options):
def model_installed(model):
opts = model._meta
converter = connection.introspection.table_name_converter
return not ((converter(opts.db_table) in tables) or
# Note that if a model is unmanaged we short-circuit and never try to install it
return opts.managed and not ((converter(opts.db_table) in tables) or
(opts.auto_created and converter(opts.auto_created._meta.db_table) in tables))


manifest = SortedDict(
(app_name, list(filter(model_installed, model_list)))
for app_name, model_list in all_models
Expand Down
3 changes: 2 additions & 1 deletion django/db/models/loading.py
Expand Up @@ -252,13 +252,14 @@ def save_state(self):
return {
"app_store": SortedDict(self.app_store.items()),
"app_labels": dict(self.app_labels.items()),
"app_models": SortedDict(self.app_models.items()),
"app_models": SortedDict((k, SortedDict(v.items())) for k, v in self.app_models.items()),
"app_errors": dict(self.app_errors.items()),
}

def restore_state(self, state):
"""
Restores the AppCache to a previous state from save_state.
Note that the state is used by reference, not copied in.
"""
self.app_store = state['app_store']
self.app_labels = state['app_labels']
Expand Down
108 changes: 58 additions & 50 deletions tests/modeltests/schema/tests.py
Expand Up @@ -31,20 +31,18 @@ def setUp(self):
# The unmanaged models need to be removed after the test in order to
# prevent bad interactions with the flush operation in other tests.
self.cache_state = cache.save_state()
cache.load_app("modeltests.schema")
for model in self.models:
model._meta.managed = True

def tearDown(self):
# Rollback anything that may have happened
connection.rollback()
# Delete any tables made for our models
self.delete_tables()
# Rollback anything that may have happened
connection.rollback()
connection.leave_transaction_management()
# Unhook our models
for model in self.models:
model._meta.managed = False
if "schema" in self.cache_state['app_labels']:
del self.cache_state['app_labels']['schema']
cache.restore_state(self.cache_state)

def delete_tables(self):
Expand Down Expand Up @@ -280,31 +278,36 @@ def test_m2m(self):
# Create an M2M field
new_field = ManyToManyField("schema.Tag", related_name="authors")
new_field.contribute_to_class(AuthorWithM2M, "tags")
# Ensure there's no m2m table there
self.assertRaises(DatabaseError, self.column_classes, new_field.rel.through)
connection.rollback()
# Add the field
editor = connection.schema_editor()
editor.start()
editor.create_field(
Author,
new_field,
)
editor.commit()
# Ensure there is now an m2m table there
columns = self.column_classes(new_field.rel.through)
self.assertEqual(columns['tag_id'][0], "IntegerField")
# Remove the M2M table again
editor = connection.schema_editor()
editor.start()
editor.delete_field(
Author,
new_field,
)
editor.commit()
# Ensure there's no m2m table there
self.assertRaises(DatabaseError, self.column_classes, new_field.rel.through)
connection.rollback()
try:
# Ensure there's no m2m table there
self.assertRaises(DatabaseError, self.column_classes, new_field.rel.through)
connection.rollback()
# Add the field
editor = connection.schema_editor()
editor.start()
editor.create_field(
Author,
new_field,
)
editor.commit()
# Ensure there is now an m2m table there
columns = self.column_classes(new_field.rel.through)
self.assertEqual(columns['tag_id'][0], "IntegerField")
# Remove the M2M table again
editor = connection.schema_editor()
editor.start()
editor.delete_field(
Author,
new_field,
)
editor.commit()
# Ensure there's no m2m table there
self.assertRaises(DatabaseError, self.column_classes, new_field.rel.through)
connection.rollback()
finally:
# Cleanup model states
AuthorWithM2M._meta.local_many_to_many.remove(new_field)
del AuthorWithM2M._meta._m2m_cache

def test_m2m_repoint(self):
"""
Expand All @@ -330,26 +333,31 @@ def test_m2m_repoint(self):
# Repoint the M2M
new_field = ManyToManyField(UniqueTest)
new_field.contribute_to_class(BookWithM2M, "uniques")
editor = connection.schema_editor()
editor.start()
editor.alter_field(
Author,
BookWithM2M._meta.get_field_by_name("tags")[0],
new_field,
)
editor.commit()
# Ensure old M2M is gone
self.assertRaises(DatabaseError, self.column_classes, BookWithM2M._meta.get_field_by_name("tags")[0].rel.through)
connection.rollback()
# Ensure the new M2M exists and points to UniqueTest
constraints = connection.introspection.get_constraints(connection.cursor(), new_field.rel.through._meta.db_table)
if connection.features.supports_foreign_keys:
for name, details in constraints.items():
if details['columns'] == set(["uniquetest_id"]) and details['foreign_key']:
self.assertEqual(details['foreign_key'], ('schema_uniquetest', 'id'))
break
else:
self.fail("No FK constraint for tag_id found")
try:
editor = connection.schema_editor()
editor.start()
editor.alter_field(
Author,
BookWithM2M._meta.get_field_by_name("tags")[0],
new_field,
)
editor.commit()
# Ensure old M2M is gone
self.assertRaises(DatabaseError, self.column_classes, BookWithM2M._meta.get_field_by_name("tags")[0].rel.through)
connection.rollback()
# Ensure the new M2M exists and points to UniqueTest
constraints = connection.introspection.get_constraints(connection.cursor(), new_field.rel.through._meta.db_table)
if connection.features.supports_foreign_keys:
for name, details in constraints.items():
if details['columns'] == set(["uniquetest_id"]) and details['foreign_key']:
self.assertEqual(details['foreign_key'], ('schema_uniquetest', 'id'))
break
else:
self.fail("No FK constraint for tag_id found")
finally:
# Cleanup model states
BookWithM2M._meta.local_many_to_many.remove(new_field)
del BookWithM2M._meta._m2m_cache

@skipUnless(connection.features.supports_check_constraints, "No check constraints")
def test_check_constraints(self):
Expand Down

0 comments on commit d0b3536

Please sign in to comment.