"""不依赖外部数据库的 ORM 约束和 migration 入口检查。""" from __future__ import annotations import ast from pathlib import Path import pytest from alembic.config import Config from alembic.script import ScriptDirectory from sqlalchemy import create_engine from sqlalchemy.orm import Session from backend.database import AgentCallbackRequest, AgentCallbackRequestRepository, DatabaseError, GLOBAL_LANE_RESOURCE_KEY, ScopeContext, TaskLaneFairness, TaskLaneResourceSlot from backend import database from backend import metadata from backend.persistence import models def test_single_forward_migration_head(): """仓库只暴露一个前向 revision head，回滚由 migration 明确拒绝。""" script = ScriptDirectory.from_config(Config("alembic.ini")) assert script.get_heads() == ["0028_image_library_sort"] assert (Path("alembic/versions/0001_postgres_scoped.py")).is_file() def test_content_identity_migration_preflights_history_and_installs_strict_unique_constraint(): """内容身份迁移必须拒绝历史脏记录并安装 scope 内唯一约束。""" migration = Path("alembic/versions/0022_separate_image_display_names_and_content_identity.py").read_text(encoding="utf-8") assert "image_identity_preflight_failed" in migration assert "GROUP BY scope_id, sha256, extension" in migration assert "uq_memes_scope_content" in migration assert "NOT VALID" not in migration assert "ALTER COLUMN display_name SET NOT NULL" in migration assert "ALTER COLUMN metadata_schema_version SET DEFAULT 2" not in migration assert "ALTER COLUMN metadata_schema_version SET DEFAULT 1" in migration assert metadata.SCHEMA_VERSION == 1 assert models.Meme.metadata_schema_version.default.arg == 1 assert models.Meme.metadata_schema_version.server_default.arg.text == "1" assert "regexp_replace(storage_key" not in migration def test_image_processing_fixed_plan_migration_is_forward_only_and_fail_closed_on_active_history(): """图片处理计划迁移必须回填历史、建立目标索引并阻止未归类活动任务。""" migration = Path("alembic/versions/0023_image_processing_fixed_plans.py").read_text(encoding="utf-8") assert 'down_revision = "0022_separate_image_display_names_and_content_identity"' in migration assert "processing_mode" in migration assert "target_meme_id" in migration and "target_image_sha256" in migration assert "image_processing_history_unresolved" in migration assert "ix_tasks_image_target_active" in migration assert "raise RuntimeError" in migration def test_image_processing_retry_hardening_checks_active_task_target_identity(): """0024 必须接在固定计划迁移后，并拒绝 SHA 已变化的活动历史 Task。""" migration = Path("alembic/versions/0024_image_processing_retry_hardening.py").read_text(encoding="utf-8") assert 'down_revision = "0023_image_processing_fixed_plans"' in migration assert "LEFT JOIN memes AS meme" in migration assert "lower(meme.sha256) <> lower(task.target_image_sha256)" in migration assert "image_processing_history_unresolved" in migration assert "raise RuntimeError" in migration def test_image_processing_migration_is_chained_and_rebuilds_legacy_checks(): """0012 必须从 0011 升级，并显式重建旧三阶段约束。""" migration = Path("alembic/versions/0012_image_processing_options_auto_rename.py").read_text(encoding="utf-8") assert 'down_revision = "0011_harden_operation_grant_association"' in migration assert "ADD COLUMN IF NOT EXISTS auto_name BOOLEAN NOT NULL DEFAULT FALSE" in migration assert "DROP CONSTRAINT IF EXISTS ck_task_image_stage" in migration assert "DROP CONSTRAINT IF EXISTS ck_task_image_stage_type" in migration assert "DROP CONSTRAINT IF EXISTS ck_image_processing_stage_name" in migration assert "DROP CONSTRAINT IF EXISTS ck_image_processing_stage_status" in migration assert "image_auto_rename" in migration assert "'skipped','warning'" in migration def test_startup_compatibility_ddl_closes_nullable_auto_name(): """启动兼容路径必须把旧可空列收束为安全的非空默认值。""" database = Path("backend/persistence/engine.py").read_text(encoding="utf-8") assert "UPDATE image_processing_jobs SET auto_name = FALSE WHERE auto_name IS NULL" in database assert "ALTER TABLE image_processing_jobs ALTER COLUMN auto_name SET NOT NULL" in database def test_callback_binding_migration_is_forward_only_and_fail_closed_on_history_conflicts(): """callback 迁移先检查缺失/重复事实，再安装逻辑唯一索引。""" migration = Path("alembic/versions/0015_bind_agent_callback_request_ids.py").read_text(encoding="utf-8") assert 'down_revision = "0014_scope_aware_opencode_workspace"' in migration assert "incomplete_count" in migration assert "duplicate_count" in migration assert "历史重复逻辑绑定" in migration assert "CREATE UNIQUE INDEX IF NOT EXISTS uq_agent_callback_requests_logical" in migration assert "raise RuntimeError" in migration def test_fair_scheduling_migration_is_chained_and_forward_only(): """公平状态迁移必须接在 callback head 后，并拒绝危险回滚。""" migration = Path("alembic/versions/0016_agent_fair_scheduling.py").read_text(encoding="utf-8") assert 'down_revision = "0015_bind_agent_callback_request_ids"' in migration assert "task_lane_fairness" in migration assert "last_dispatch_sequence" in migration assert "raise RuntimeError" in migration def test_resource_scheduling_migration_backfills_global_resource_and_adds_resource_slots(): """资源迁移必须回填旧任务、公平行并安装独立资源槽位表。""" migration = Path("alembic/versions/0019_task_lane_resource_scheduling.py").read_text(encoding="utf-8") assert 'down_revision = "0018_operation_grant_metering_units"' in migration assert "lane_resource_key" in migration assert "__global__" in migration assert "task_lane_resource_slots" in migration assert "PRIMARY KEY (lane, resource_key, slot_number)" in migration assert "PRIMARY KEY (lane, resource_key, scope_id)" in migration assert "raise RuntimeError" in migration def test_callback_model_keeps_request_id_and_logic_unique_facts(): """ORM callback 表同时保留旧 request ID 主键和新的复合逻辑唯一约束。""" table = AgentCallbackRequest.__table__ assert {column.name for column in table.primary_key.columns} == {"scope_id", "request_id"} assert any(constraint.name == "uq_agent_callback_requests_logical" for constraint in table.constraints) def test_fairness_model_is_lane_scope_keyed_and_has_dispatch_index(): """公平事实以 lane/资源/scope 为复合主键，并可按持久序号排序。""" table = TaskLaneFairness.__table__ assert {column.name for column in table.primary_key.columns} == {"lane", "resource_key", "scope_id"} assert {column.name for column in table.columns} >= {"resource_key", "last_dispatch_sequence", "created_at", "updated_at"} assert any(index.name == "ix_task_lane_fairness_dispatch" for index in table.indexes) def test_resource_slot_model_keeps_global_slot_separate_from_resource_slot(): """资源槽位使用独立表，旧全局槽位表仍保留原主键。""" table = TaskLaneResourceSlot.__table__ assert {column.name for column in table.primary_key.columns} == {"lane", "resource_key", "slot_number"} assert {column.name for column in table.columns} >= {"resource_key", "claim_generation", "lease_expires_at"} assert GLOBAL_LANE_RESOURCE_KEY == "__global__" def test_callback_repository_fails_closed_without_postgres_schema(): """callback repository 不回退到非 PostgreSQL 或 request-ID-only 事实层。""" engine = create_engine("sqlite:///:memory:") with Session(engine) as session: repository = AgentCallbackRequestRepository(session, ScopeContext("local")) with pytest.raises(DatabaseError, match="callback_binding_schema_unavailable"): repository.ensure_schema_ready() def test_database_facade_reexports_one_model_declaration_source(): """旧 database 导入路径必须指向持久化模型模块的同一组对象。""" model_names = ( "Base", "ScopeContext", "Scope", "InstallationState", "Meme", "MemeCollection", "MemeCollectionItem", "StorageOperation", "SearchGeneration", "SearchHead", "MemeEmbedding", "MemeVisualEmbedding", "Task", "ReverseImageUsageEvent", "AgentCallbackRequest", "OperationGrant", "ImageProcessingJob", "ImageProcessingStage", "ImageProcessingAttempt", "MemeTextEmbedding", "SearchMigrationState", "TaskBatch", "TaskBatchItem", "TaskLaneResourceSlot", "TaskLaneSlot", "TaskLaneFairness", ) for name in model_names: assert getattr(database, name) is getattr(models, name) assert getattr(models, name).__module__ == models.__name__ assert database.EMBEDDING_DIMENSIONS == models.EMBEDDING_DIMENSIONS assert database.VISUAL_EMBEDDING_DIMENSIONS == models.VISUAL_EMBEDDING_DIMENSIONS assert database.UTC is models.UTC assert database.utcnow is models.utcnow assert database.OPTIONAL_CONTROL_TABLES is models.OPTIONAL_CONTROL_TABLES assert database.Base.metadata is models.Base.metadata for name in ( "BigInteger", "Boolean", "CheckConstraint", "DateTime", "DeclarativeBase", "ForeignKey", "ForeignKeyConstraint", "Index", "Integer", "JSON", "JSONB", "Mapped", "String", "UniqueConstraint", "Uuid", "Vector", "mapped_column", "timezone", "unicodedata", ): assert getattr(database, name) is getattr(models, name) def test_model_module_does_not_reintroduce_database_or_runtime_boundaries(): """模型模块不能反向依赖 facade、Repository 或文件存储装配。""" source = Path(models.__file__).read_text(encoding="utf-8") tree = ast.parse(source) imported_modules = { alias.name for node in ast.walk(tree) if isinstance(node, ast.Import) for alias in node.names } imported_modules.update( node.module for node in ast.walk(tree) if isinstance(node, ast.ImportFrom) and node.module is not None ) assert "backend.database" not in imported_modules assert "StorageCoordinator" not in source assert "BlobStore" not in source assert "class Scope(" not in Path(database.__file__).read_text(encoding="utf-8")