test(database): validate media identity constraints semantically (#6422)

tests/test_database_index_migration.py

@@ -71,7 +71,7 @@ ensure_sites_stub()
 
 from alembic.config import Config
 from alembic.script import ScriptDirectory
-from sqlalchemy import inspect, text
+from sqlalchemy import inspect, MetaData, Table, text
 from sqlalchemy.exc import IntegrityError
 
 from app.runtime.config import settings
@@ -110,8 +110,8 @@ with get_engine().connect() as connection:
             )
             for index in inspector.get_indexes(table_name)
         }}
-        constraints = {{
-            constraint["name"]: constraint.get("sqltext") or ""
+        constraint_names = {{
+            constraint["name"]
             for constraint in inspector.get_check_constraints(table_name)
         }}
         assert {{"media_source", "media_id"}}.issubset(columns), (
@@ -129,57 +129,71 @@ with get_engine().connect() as connection:
                 indexes,
             )
         constraint_name = f"ck_{{table_name}}_media_identity"
-        assert constraint_name in constraints, (
+        assert constraint_name in constraint_names, (
             table_name,
-            constraints,
-        )
-        normalized_sql = "".join(
-            constraints[constraint_name].lower().replace('"', '').split()
-        )
-        for text_cast in ("::text[]", "::text", "::charactervarying"):
-            normalized_sql = normalized_sql.replace(text_cast, "")
-        for fragment in (
-            "media_sourceisnull",
-            "media_idisnull",
-            "media_sourceisnotnull",
-            "media_idisnotnull",
-            "length(media_source)",
-            "media_sourcenotlike'%:%'",
-        ):
-            assert fragment in normalized_sql, (
-                table_name,
-                constraints[constraint_name],
-            )
-        assert any(
-            trim_form in normalized_sql
-            for trim_form in (
-                "trim(media_id)",
-                "trim(bothfrommedia_id)",
-            )
-        ), (table_name, constraints[constraint_name])
-        assert "<>''" in normalized_sql, (
-            table_name,
-            constraints[constraint_name],
-        )
-        assert "<>'0'" in normalized_sql, (
-            table_name,
-            constraints[constraint_name],
+            constraint_names,
         )
 
-    constraint_name = "ck_mediaserveritem_media_identity"
-    try:
-        with connection.begin_nested():
-            connection.execute(
-                text(
-                    "INSERT INTO mediaserveritem (media_source, media_id) "
-                    "VALUES (:media_source, :media_id)"
-                ),
-                {{"media_source": "invalid:source", "media_id": "1"}},
-            )
-    except IntegrityError as error:
-        assert constraint_name in str(error.orig), str(error.orig)
-    else:
-        raise AssertionError("格式非法的媒体身份未被具名检查约束拒绝")
+    required_values = {{
+        "subscribe": {{"name": "constraint-test", "state": "N"}},
+        "subscribehistory": {{"name": "constraint-test"}},
+        "downloadhistory": {{
+            "path": "/constraint-test",
+            "type": "电影",
+            "title": "constraint-test",
+        }},
+        "transferhistory": {{"src_storage": "local"}},
+        "downloadfailure": {{"fingerprint": "constraint-test"}},
+        "mediaserveritem": {{}},
+    }}
+    invalid_identities = (
+        (None, "1"),
+        ("acme.video", None),
+        ("", "1"),
+        (" acme.video", "1"),
+        ("acme.video ", "1"),
+        ("Acme.Video", "1"),
+        ("a" * 65, "1"),
+        ("invalid:source", "1"),
+        ("invalid source", "1"),
+        ("acme.video", ""),
+        ("acme.video", "  "),
+        ("acme.video", "0"),
+    )
+    for table_name in media_tables:
+        table = Table(table_name, MetaData(), autoload_with=connection)
+        constraint_name = f"ck_{{table_name}}_media_identity"
+        for media_source, media_id in (
+            (None, None),
+            ("acme.video", "custom-1"),
+        ):
+            values = {{
+                **required_values[table_name],
+                "media_source": media_source,
+                "media_id": media_id,
+            }}
+            savepoint = connection.begin_nested()
+            try:
+                connection.execute(table.insert(), values)
+            finally:
+                savepoint.rollback()
+
+        for media_source, media_id in invalid_identities:
+            values = {{
+                **required_values[table_name],
+                "media_source": media_source,
+                "media_id": media_id,
+            }}
+            try:
+                with connection.begin_nested():
+                    connection.execute(table.insert(), values)
+            except IntegrityError as error:
+                assert constraint_name in str(error.orig), str(error.orig)
+            else:
+                raise AssertionError(
+                    "格式非法的媒体身份未被具名检查约束拒绝: "
+                    f"{{table_name}}, {{media_source!r}}, {{media_id!r}}"
+                )
 """.format(
     media_tables=MEDIA_TABLES,
     legacy_identity_columns=LEGACY_IDENTITY_COLUMNS,