76 lines
3.2 KiB
Python
76 lines
3.2 KiB
Python
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, "0003_retrieval_runs")
|
|
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"))
|
|
== "0004_auth_sessions"
|
|
)
|
|
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)
|