[3.49] Fix ALLOWED_CONTENT_CHECKSUMS check failing on settings import

CHANGES/7380.bugfix

@@ -0,0 +1 @@
+Moved artifact checksum validation from module-level settings code to a Django system check.

pulpcore/app/checks.py

@@ -1,7 +1,16 @@
 from pathlib import Path
 
 from django.conf import settings
-from django.core.checks import Error as CheckError, Warning as CheckWarning, register
+from django.core.checks import (
+    Debug as CheckDebug,
+    Error as CheckError,
+    Warning as CheckWarning,
+    Tags,
+    register,
+)
+from django.db.models import Q
+
+from pulpcore import constants
 
 
 @register(deploy=True)
@@ -56,3 +65,55 @@ def storage_paths(app_configs, **kwargs):
             )
 
     return warnings
+
+
+@register(Tags.database)
+def check_artifact_checksums(app_configs, **kwargs):
+    from pulpcore.app.models import Artifact, RemoteArtifact
+
+    messages = []
+    allowed = set(settings.ALLOWED_CONTENT_CHECKSUMS)
+    forbidden = set(constants.ALL_KNOWN_CONTENT_CHECKSUMS).difference(allowed)
+
+    try:
+        for checksum in allowed:
+            if Artifact.objects.filter(**{checksum: None}).exists():
+                messages.append(
+                    CheckError(
+                        f"There have been identified artifacts missing checksum '{checksum}'. "
+                        "Run 'pulpcore-manager handle-artifact-checksums' first to populate "
+                        "missing artifact checksums.",
+                        id="pulpcore.E002",
+                    )
+                )
+        for checksum in forbidden:
+            if Artifact.objects.exclude(**{checksum: None}).exists():
+                messages.append(
+                    CheckWarning(
+                        f"There have been identified artifacts with forbidden checksum "
+                        f"'{checksum}'. Run 'pulpcore-manager handle-artifact-checksums' "
+                        "to unset forbidden checksums.",
+                        id="pulpcore.W004",
+                    )
+                )
+
+        has_any_checksum = ~Q(**{c: None for c in constants.ALL_KNOWN_CONTENT_CHECKSUMS})
+        missing_allowed = Q(**{c: None for c in allowed})
+        if RemoteArtifact.objects.filter(has_any_checksum & missing_allowed).exists():
+            messages.append(
+                CheckWarning(
+                    "Detected remote content without allowed checksums. "
+                    "Run 'pulpcore-manager handle-artifact-checksums --report' to "
+                    "view this content.",
+                    id="pulpcore.W005",
+                )
+            )
+    except Exception:
+        messages.append(
+            CheckDebug(
+                "Skipping artifact checksum checks (table may not exist yet).",
+                id="pulpcore.D001",
+            )
+        )
+
+    return messages

pulpcore/app/settings.py

@@ -19,7 +19,6 @@
 from django.conf import global_settings
 from django.core.files.storage import storages  # noqa: F401
 from django.core.exceptions import ImproperlyConfigured
-from django.db import connection
 
 from pulpcore import constants
 
@@ -476,69 +475,6 @@
     ALLOWED_CONTENT_CHECKSUMS
 )
 
-_SKIPPED_COMMANDS_FOR_CONTENT_CHECKS = [
-    "handle-artifact-checksums",
-    "migrate",
-    "collectstatic",
-    "openapi",
-]
-
-if not (len(sys.argv) >= 2 and sys.argv[1] in _SKIPPED_COMMANDS_FOR_CONTENT_CHECKS):
-    try:
-        with connection.cursor() as cursor:
-            for checksum in ALLOWED_CONTENT_CHECKSUMS:
-                # can't import Artifact here so use a direct db connection
-                cursor.execute(f"SELECT count(pulp_id) FROM core_artifact WHERE {checksum} IS NULL")
-                row = cursor.fetchone()
-                if row[0] > 0:
-                    raise ImproperlyConfigured(
-                        (
-                            "There have been identified artifacts missing checksum '{}'. "
-                            "Run 'pulpcore-manager handle-artifact-checksums' first to populate "
-                            "missing artifact checksums."
-                        ).format(checksum)
-                    )
-            for checksum in FORBIDDEN_CHECKSUMS:
-                # can't import Artifact here so use a direct db connection
-                cursor.execute(
-                    f"SELECT count(pulp_id) FROM core_artifact WHERE {checksum} IS NOT NULL"
-                )
-                row = cursor.fetchone()
-                if row[0] > 0:
-                    raise ImproperlyConfigured(
-                        (
-                            "There have been identified artifacts with forbidden checksum '{}'. "
-                            "Run 'pulpcore-manager handle-artifact-checksums' first to unset "
-                            "forbidden checksums."
-                        ).format(checksum)
-                    )
-
-            # warn if there are remote artifacts with checksums but no allowed checksums
-            cond = " AND ".join([f"{c} IS NULL" for c in constants.ALL_KNOWN_CONTENT_CHECKSUMS])
-            no_checksum_query = f"SELECT pulp_id FROM core_remoteartifact WHERE {cond}"
-            cond = " AND ".join([f"{c} IS NULL" for c in ALLOWED_CONTENT_CHECKSUMS])
-            cursor.execute(
-                f"SELECT count(pulp_id) FROM core_remoteartifact WHERE {cond} AND "
-                f"pulp_id NOT IN ({no_checksum_query})"
-            )
-            row = cursor.fetchone()
-            if row[0] > 0:
-                _logger.warning(
-                    (
-                        "Warning: detected remote content without allowed checksums. "
-                        "Run 'pulpcore-manager handle-artifact-checksums --report' to "
-                        "view this content."
-                    )
-                )
-
-    except ImproperlyConfigured as e:
-        raise e
-    except Exception:
-        # our check could fail if the table hasn't been created yet or we can't get a db connection
-        pass
-    finally:
-        connection.close()
-
 # Ensures the cached property storage.backends uses the right value after dynaconf init
 storages._backends = settings.STORAGES.copy()
 storages.backends