feat: implement backend knowledge management

backend/tests/test_migrations.py

@@ -0,0 +1,41 @@
+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_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")) == "0002_catalog"
+        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)