import pytest
from alembic.autogenerate import compare_metadata
from alembic.migration import MigrationContext
from conftest import migrate
from fastapi.testclient import TestClient
from sqlalchemy import text

from app.db.session import Base
from app.main import create_app


def test_startup_requires_migration(database):
    with pytest.raises(RuntimeError, match="alembic upgrade head"):
        with TestClient(create_app(database)):
            pass


def test_startup_rejects_previous_schema(database):
    migrate(database, "0005_chunk_indexes")
    with pytest.raises(RuntimeError, match="alembic upgrade head"):
        with TestClient(create_app(database)):
            pass


def test_upgrade_preserves_original_rows_and_repeated_upgrade(database):
    migrate(database, "0001_baseline")
    with database.begin() as connection:
        connection.execute(
            text("""INSERT INTO knowledge_items
            (title, category, content, source_url, verification_status)
            VALUES ('示例旧数据', '未分类', '原始正文', 'https://example.com/', 'verified')""")
        )
        # 模拟 create_all 建表后没有 Alembic 版本记录的旧工程。
        connection.execute(text("DELETE FROM alembic_version"))
    migrate(database)
    migrate(database)
    with database.connect() as connection:
        row = connection.execute(text("SELECT * FROM knowledge_items")).mappings().one()
        assert row["title"] == "示例旧数据" and row["content"] == "原始正文"
        assert row["collected_at"] is None
        assert row["verification_status"] == "pending"
        assert connection.scalar(text("SELECT count(*) FROM audit_events")) == 2
        assert connection.scalar(text("SELECT version_num FROM alembic_version")) == "0006_vectors"
        assert compare_metadata(MigrationContext.configure(connection), Base.metadata) == []
    migrate(database, "0001_baseline", downgrade=True)
    with database.connect() as connection:
        assert connection.scalar(text("SELECT content FROM knowledge_items")) == "原始正文"
    migrate(database)


def test_auth_upgrade_and_downgrade_preserve_knowledge_and_retrieval(database):
    migrate(database, "0003_retrieval_runs")
    with database.begin() as c:
        c.execute(
            text("""INSERT INTO knowledge_items
        (title, category, content, verification_status, version, updated_at, is_deleted)
        VALUES ('占位', '其他', '保留的占位正文', 'pending', 1, now(), false)""")
        )
        c.execute(
            text("""INSERT INTO retrieval_runs
        (id, query, mode, corpus_fingerprint, response)
        VALUES ('00000000-0000-0000-0000-000000000001', '占位', 'keyword', 'placeholder', '{}')""")
        )
    migrate(database)
    with database.connect() as c:
        assert c.scalar(text("SELECT count(*) FROM users")) == 0
        assert c.scalar(text("SELECT count(*) FROM auth_sessions")) == 0
    migrate(database, "0003_retrieval_runs", downgrade=True)
    with database.connect() as c:
        assert c.scalar(text("SELECT content FROM knowledge_items")) == "保留的占位正文"
        assert c.scalar(text("SELECT count(*) FROM retrieval_runs")) == 1
    migrate(database)


def test_chunk_index_downgrade_preserves_users_sessions_knowledge_and_retrieval(
    client, migrated_database
):
    from test_indexing import build, publish
    from test_retrieval import add_item, search

    item = add_item(client)
    run = search(client)
    version = build(client)
    assert publish(client, version).status_code == 200
    migrate(migrated_database, "0004_auth_sessions", downgrade=True)
    with migrated_database.connect() as c:
        assert c.scalar(text("SELECT count(*) FROM users")) == 1
        assert c.scalar(text("SELECT count(*) FROM auth_sessions")) == 1
        assert (
            c.scalar(text("SELECT content FROM knowledge_items WHERE id=:id"), {"id": item["id"]})
            == item["content"]
        )
        assert c.scalar(text("SELECT id FROM retrieval_runs")) == run["run_id"]
    migrate(migrated_database)
    assert client.get("/api/auth/me").status_code == 200
    assert client.get("/api/indexes").json()["total"] == 0
    assert client.get("/api/indexes/active").json()["generation"] == 0


def test_vector_downgrade_retains_text_publications_and_prior_retrieval_runs(
    client, migrated_database
):
    from test_retrieval import add_item
    from test_vectors import ToyProvider, text_version, vector_build, vector_publish, vector_search

    from app.services.embedding_runtime import get_embedding_provider

    client.app.dependency_overrides[get_embedding_provider] = ToyProvider
    add_item(client)
    text_ = text_version(client)
    version = vector_build(client, text_)
    vector_publish(client, version)
    history = vector_search(client).json()
    migrate(migrated_database, "0005_chunk_indexes", downgrade=True)
    with migrated_database.connect() as c:
        assert c.scalar(text("SELECT active_version_id FROM index_state")) == text_["id"]
        assert c.scalar(text("SELECT count(*) FROM chunks")) == 1
        assert c.scalar(text("SELECT count(*) FROM users")) == 1
        assert c.scalar(text("SELECT count(*) FROM index_publications")) == 1
        assert c.scalar(text("SELECT id FROM retrieval_runs")) == history["run_id"]
    migrate(migrated_database)
    assert client.get("/api/vector-indexes/active").json()["generation"] == 0
    assert client.get(f"/api/retrieval/runs/{history['run_id']}").json() == history