Fixed #23275: Unmanaged models kept by autodetector, ignored by ops

django/db/migrations/autodetector.py

@@ -108,11 +108,15 @@ class MigrationAutodetector(object):
         self.new_apps = self.to_state.render()
         self.old_model_keys = []
         self.old_proxy_keys = []
+        self.old_unmanaged_keys = []
         self.new_model_keys = []
         self.new_proxy_keys = []
+        self.new_unmanaged_keys = []
         for al, mn in sorted(self.from_state.models.keys()):
             model = self.old_apps.get_model(al, mn)
-            if model._meta.managed and al not in self.from_state.real_apps:
+            if not model._meta.managed:
+                self.old_unmanaged_keys.append((al, mn))
+            elif al not in self.from_state.real_apps:
                 if model._meta.proxy:
                     self.old_proxy_keys.append((al, mn))
                 else:
@@ -120,7 +124,9 @@ class MigrationAutodetector(object):
 
         for al, mn in sorted(self.to_state.models.keys()):
             model = self.new_apps.get_model(al, mn)
-            if model._meta.managed and (
+            if not model._meta.managed:
+                self.new_unmanaged_keys.append((al, mn))
+            elif (
                 al not in self.from_state.real_apps or
                 (convert_apps and al in convert_apps)
             ):
@@ -136,6 +142,8 @@ class MigrationAutodetector(object):
         # through models in the old state so we can make dependencies
         # from the through model deletion to the field that uses it.
         self.kept_model_keys = set(self.old_model_keys).intersection(self.new_model_keys)
+        self.kept_proxy_keys = set(self.old_proxy_keys).intersection(self.new_proxy_keys)
+        self.kept_unmanaged_keys = set(self.old_unmanaged_keys).intersection(self.new_unmanaged_keys)
         self.through_users = {}
         self.old_field_keys = set()
         self.new_field_keys = set()
@@ -164,6 +172,8 @@ class MigrationAutodetector(object):
         self.generate_created_models()
         self.generate_deleted_proxies()
         self.generate_created_proxies()
+        self.generate_deleted_unmanaged()
+        self.generate_created_unmanaged()
         self.generate_altered_options()
 
         # Generate field operations
@@ -539,17 +549,23 @@ class MigrationAutodetector(object):
                     ]
                 )
 
-    def generate_created_proxies(self):
+    def generate_created_proxies(self, unmanaged=False):
         """
         Makes CreateModel statements for proxy models.
         We use the same statements as that way there's less code duplication,
         but of course for proxy models we can skip all that pointless field
         stuff and just chuck out an operation.
         """
-        added_proxies = set(self.new_proxy_keys) - set(self.old_proxy_keys)
+        if unmanaged:
-        for app_label, model_name in sorted(added_proxies):
+            added = set(self.new_unmanaged_keys) - set(self.old_unmanaged_keys)
+        else:
+            added = set(self.new_proxy_keys) - set(self.old_proxy_keys)
+        for app_label, model_name in sorted(added):
             model_state = self.to_state.models[app_label, model_name]
-            assert model_state.options.get("proxy", False)
+            if unmanaged:
+                assert not model_state.options.get("managed", True)
+            else:
+                assert model_state.options.get("proxy", False)
             # Depend on the deletion of any possible non-proxy version of us
             dependencies = [
                 (app_label, model_name, None, False),
@@ -572,6 +588,15 @@ class MigrationAutodetector(object):
                 dependencies=dependencies,
             )
 
+    def generate_created_unmanaged(self):
+        """
+        Similar to generate_created_proxies but for unmanaged
+        (they are similar to us in that we need to supply them, but they don't
+        affect the DB)
+        """
+        # Just re-use the same code in *_proxies
+        self.generate_created_proxies(unmanaged=True)
+
     def generate_deleted_models(self):
         """
         Find all deleted models and make creation operations for them,
@@ -668,14 +693,20 @@ class MigrationAutodetector(object):
                 dependencies=list(set(dependencies)),
             )
 
-    def generate_deleted_proxies(self):
+    def generate_deleted_proxies(self, unmanaged=False):
         """
         Makes DeleteModel statements for proxy models.
         """
-        deleted_proxies = set(self.old_proxy_keys) - set(self.new_proxy_keys)
+        if unmanaged:
-        for app_label, model_name in sorted(deleted_proxies):
+            deleted = set(self.old_unmanaged_keys) - set(self.new_unmanaged_keys)
+        else:
+            deleted = set(self.old_proxy_keys) - set(self.new_proxy_keys)
+        for app_label, model_name in sorted(deleted):
             model_state = self.from_state.models[app_label, model_name]
-            assert model_state.options.get("proxy", False)
+            if unmanaged:
+                assert not model_state.options.get("managed", True)
+            else:
+                assert model_state.options.get("proxy", False)
             self.add_operation(
                 app_label,
                 operations.DeleteModel(
@@ -683,6 +714,12 @@ class MigrationAutodetector(object):
                 ),
             )
 
+    def generate_deleted_unmanaged(self):
+        """
+        Makes DeleteModel statements for unmanaged models
+        """
+        self.generate_deleted_proxies(unmanaged=True)
+
     def generate_renamed_fields(self):
         """
         Works out renamed fields
@@ -852,7 +889,7 @@ class MigrationAutodetector(object):
         makes an operation to represent them in state changes (in case Python
         code in migrations needs them)
         """
-        models_to_check = self.kept_model_keys.union(set(self.new_proxy_keys).intersection(self.old_proxy_keys))
+        models_to_check = self.kept_model_keys.union(self.kept_proxy_keys).union(self.kept_unmanaged_keys)
         for app_label, model_name in sorted(models_to_check):
             old_model_name = self.renamed_models.get((app_label, model_name), model_name)
             old_model_state = self.from_state.models[app_label, old_model_name]

django/db/migrations/operations/base.py

@@ -101,12 +101,13 @@ class Operation(object):
     def allowed_to_migrate(self, connection_alias, model):
         """
         Returns if we're allowed to migrate the model. Checks the router,
-        if it's a proxy, and if it's swapped out.
+        if it's a proxy, if it's managed, and if it's swapped out.
         """
         return (
             router.allow_migrate(connection_alias, model) and
             not model._meta.proxy and
-            not model._meta.swapped
+            not model._meta.swapped and
+            model._meta.managed
         )
 
     def __repr__(self):

tests/migrations/test_autodetector.py

@@ -639,30 +639,28 @@ class AutodetectorTests(TestCase):
         self.assertOperationAttributes(changes, "testapp", 0, 0, name="AuthorProxy")
         self.assertOperationAttributes(changes, "testapp", 0, 1, name="AuthorProxy", options={})
 
-    def test_unmanaged_ignorance(self):
+    def test_unmanaged(self):
-        "Tests that the autodetector correctly ignores managed models"
+        "Tests that the autodetector correctly deals with managed models"
         # First, we test adding an unmanaged model
         before = self.make_project_state([self.author_empty])
         after = self.make_project_state([self.author_empty, self.author_unmanaged])
         autodetector = MigrationAutodetector(before, after)
         changes = autodetector._detect_changes()
         # Right number of migrations?
-        self.assertEqual(len(changes), 0)
+        self.assertNumberMigrations(changes, 'testapp', 1)
-
+        self.assertOperationTypes(changes, 'testapp', 0, ["CreateModel"])
+        self.assertOperationAttributes(changes, 'testapp', 0, 0, name="AuthorUnmanaged")
+        self.assertEqual(changes['testapp'][0].operations[0].options['managed'], False)
         # Now, we test turning an unmanaged model into a managed model
         before = self.make_project_state([self.author_empty, self.author_unmanaged])
         after = self.make_project_state([self.author_empty, self.author_unmanaged_managed])
         autodetector = MigrationAutodetector(before, after)
         changes = autodetector._detect_changes()
         # Right number of migrations?
-        self.assertEqual(len(changes['testapp']), 1)
+        self.assertNumberMigrations(changes, 'testapp', 1)
-        # Right number of actions?
+        self.assertOperationTypes(changes, 'testapp', 0, ["DeleteModel", "CreateModel"])
-        migration = changes['testapp'][0]
+        self.assertOperationAttributes(changes, 'testapp', 0, 0, name="AuthorUnmanaged")
-        self.assertEqual(len(migration.operations), 1)
+        self.assertOperationAttributes(changes, 'testapp', 0, 1, name="AuthorUnmanaged")
-        # Right action?
-        action = migration.operations[0]
-        self.assertEqual(action.__class__.__name__, "CreateModel")
-        self.assertEqual(action.name, "AuthorUnmanaged")
 
     @override_settings(AUTH_USER_MODEL="thirdapp.CustomUser")
     def test_swappable(self):

tests/migrations/test_operations.py

@@ -326,6 +326,35 @@ class OperationTests(OperationTestBase):
         self.assertTableNotExists("test_crprmo_proxypony")
         self.assertTableExists("test_crprmo_pony")
 
+    def test_create_unmanaged_model(self):
+        """
+        Tests that CreateModel ignores unmanaged models.
+        """
+        project_state = self.set_up_test_model("test_crummo")
+        # Test the state alteration
+        operation = migrations.CreateModel(
+            "UnmanagedPony",
+            [],
+            options={"proxy": True},
+            bases=("test_crummo.Pony", ),
+        )
+        self.assertEqual(operation.describe(), "Create proxy model UnmanagedPony")
+        new_state = project_state.clone()
+        operation.state_forwards("test_crummo", new_state)
+        self.assertIn(("test_crummo", "unmanagedpony"), new_state.models)
+        # Test the database alteration
+        self.assertTableNotExists("test_crummo_unmanagedpony")
+        self.assertTableExists("test_crummo_pony")
+        with connection.schema_editor() as editor:
+            operation.database_forwards("test_crummo", editor, project_state, new_state)
+        self.assertTableNotExists("test_crummo_unmanagedpony")
+        self.assertTableExists("test_crummo_pony")
+        # And test reversal
+        with connection.schema_editor() as editor:
+            operation.database_backwards("test_crummo", editor, new_state, project_state)
+        self.assertTableNotExists("test_crummo_unmanagedpony")
+        self.assertTableExists("test_crummo_pony")
+
     def test_delete_model(self):
         """
         Tests the DeleteModel operation.