diff --git a/tests/test_database_index_migration.py b/tests/test_database_index_migration.py index 9c4161a8c..32dd9effa 100644 --- a/tests/test_database_index_migration.py +++ b/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,