Added check for name clashes between static and dynamic tenants

django_pgschemas/checks.py

@@ -5,7 +5,7 @@
 from django.core.exceptions import ImproperlyConfigured
 from django.utils.module_loading import import_module
 
-from .utils import get_tenant_model, get_domain_model
+from .utils import get_tenant_model, get_domain_model, get_clone_reference
 
 
 def get_tenant_app():
@@ -108,3 +108,21 @@ def check_other_apps(app_configs, **kwargs):
                     )
                 )
     return errors
+
+
+@checks.register(checks.Tags.database)
+def check_schema_names(app_configs, **kwargs):
+    errors = []
+    static_names = set(settings.TENANTS.keys())
+    clone_reference = get_clone_reference()
+    if clone_reference:
+        static_names.add(clone_reference)
+    dynamic_names = set(get_tenant_model().objects.values_list("schema_name", flat=True))
+    intersection = static_names & dynamic_names
+    if intersection:
+        errors.append(
+            checks.Critical(
+                "Name clash found between static and dynamic tenants: %s" % intersection, id="pgschemas.W004",
+            )
+        )
+    return errors

django_pgschemas/management/commands/cloneschema.py

@@ -2,6 +2,7 @@
 
 from distutils.util import strtobool
 
+from django.core.checks import Tags, run_checks
 from django.core.management.base import BaseCommand, CommandError
 
 from ...utils import get_tenant_model, get_domain_model, clone_schema
@@ -10,6 +11,11 @@
 class Command(BaseCommand):
     help = "Clones a schema"
 
+    def _run_checks(self, **kwargs):
+        issues = run_checks(tags=[Tags.database])
+        issues.extend(super()._run_checks(**kwargs))
+        return issues
+
     def add_arguments(self, parser):
         super().add_arguments(parser)
         parser.add_argument("source", help="The name of the schema you want to clone")

django_pgschemas/management/commands/createrefschema.py

@@ -1,3 +1,4 @@
+from django.core.checks import Tags, run_checks
 from django.core.management.base import BaseCommand, CommandError
 
 from ...utils import get_clone_reference, create_schema, drop_schema
@@ -6,6 +7,11 @@
 class Command(BaseCommand):
     help = "Creates the reference schema for faster dynamic tenant creation"
 
+    def _run_checks(self, **kwargs):
+        issues = run_checks(tags=[Tags.database])
+        issues.extend(super()._run_checks(**kwargs))
+        return issues
+
     def add_arguments(self, parser):
         super().add_arguments(parser)
         parser.add_argument("--recreate", action="store_true", dest="recreate", help="Recreate reference schema.")

django_pgschemas/management/commands/migrateschema.py

@@ -1,4 +1,5 @@
 from django.core import management
+from django.core.checks import Tags, run_checks
 from django.core.management.base import BaseCommand
 from django.core.management.commands.migrate import Command as MigrateCommand
 
@@ -13,6 +14,11 @@ class NonInteractiveRunSchemaCommand(RunSchemaCommand):
 class MigrateSchemaCommand(WrappedSchemaOption, BaseCommand):
     allow_interactive = False
 
+    def _run_checks(self, **kwargs):
+        issues = run_checks(tags=[Tags.database])
+        issues.extend(super()._run_checks(**kwargs))
+        return issues
+
     def add_arguments(self, parser):
         super().add_arguments(parser)
         MigrateCommand.add_arguments(self, parser)

docs/troubleshooting.rst

@@ -45,4 +45,24 @@ all migrations of said app via ``manage.py migrate app zero --fake -s schema``
 and then run migrations again.
 
 In order to remove the tables from the source app, you will have to actually
-do a zero migrate before removing the app from the said schema apps.
+do a zero migrate before removing the app from the said schema apps.
+
+Name clash between static and dynamic schemas
+---------------------------------------------
+
+It is possible to define a static tenant whose name clashes with an existing
+dynamic tenant. This is especially true for the clone reference, which can be
+added as an afterthought, in order to speed up dynamic tenant creation. It is
+also possible to create a dynamic tenant with a name already present in the
+static tenant configuration.
+
+We do not provide an out-of-the-box validation mechanism for dynamic tenants
+upon creation, as attempt to prevent name clashes with static tenants.
+However, we do provide a system check that fails with a critical error message
+if a name clash is found. Since this check must query the database in order to
+fetch the schema name for all dynamic tenants, it is tagged as a database check,
+which makes it run only in database related operations and management commands.
+This means that the check will not be run via ``runserver``, but will be run in
+commands like ``migrate``, ``cloneschema`` and ``createrefschema``. If
+absolutely needed, you can silence this check through the code
+``pgschemas.W004``.

dpgs_sandbox/tests/test_checks.py

@@ -3,15 +3,17 @@
 from django.core import checks
 from django.test import TestCase, override_settings
 
-from django_pgschemas.checks import check_principal_apps, check_other_apps, get_user_app
+from django_pgschemas.utils import get_tenant_model
+from django_pgschemas.checks import check_principal_apps, check_other_apps, check_schema_names, get_user_app
 
 
+TenantModel = get_tenant_model()
 BASE_PUBLIC = {"TENANT_MODEL": "shared_public.Tenant", "DOMAIN_MODEL": "shared_public.DOMAIN"}
 
 
-class AppConfigTestCase(TestCase):
+class AppChecksTestCase(TestCase):
     """
-    Tests TENANTS settings is properly defined.
+    Tests multiple checks regarding applications in tenants.
     """
 
     def setUp(self):
@@ -89,3 +91,41 @@ def test_user_session_location(self):
                 )
             ]
             self.assertEqual(errors, expected_errors)
+
+
+class NameClashCheckTestCase(TestCase):
+    """
+    Tests checks regarding name clash between static and dynamic tenants.
+    """
+
+    def setUp(self):
+        self.app_config = apps.get_app_config("django_pgschemas")
+
+    def test_name_clash(self):
+        backup_create = TenantModel.auto_create_schema
+        TenantModel.auto_create_schema = False
+        # public
+        TenantModel.objects.create(schema_name="public")
+        errors = check_schema_names(self.app_config)
+        expected_errors = [
+            checks.Critical("Name clash found between static and dynamic tenants: {'public'}", id="pgschemas.W004"),
+        ]
+        self.assertEqual(errors, expected_errors)
+        TenantModel.objects.all().delete()
+        # www
+        TenantModel.objects.create(schema_name="www")
+        errors = check_schema_names(self.app_config)
+        expected_errors = [
+            checks.Critical("Name clash found between static and dynamic tenants: {'www'}", id="pgschemas.W004"),
+        ]
+        self.assertEqual(errors, expected_errors)
+        TenantModel.objects.all().delete()
+        # sample
+        TenantModel.objects.create(schema_name="sample")
+        errors = check_schema_names(self.app_config)
+        expected_errors = [
+            checks.Critical("Name clash found between static and dynamic tenants: {'sample'}", id="pgschemas.W004"),
+        ]
+        self.assertEqual(errors, expected_errors)
+        TenantModel.objects.all().delete()
+        TenantModel.auto_create_schema = backup_create