MemeMeow/tests/test_database_models.py

"""不依赖外部数据库的 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")