fix: correct migration test required_tables, clean alembic.ini, add drift detection

- Replace incorrect required_tables list with actual 48 table names extracted
  from migration a79329c23b20_initial_complete_schema.py
- Remove tables that don't exist: citations, subscription_plans, agent_executions,
  analytics_events, client_brands
- Add missing tables: agent_registry, api_keys, optimization_insights, etc.
- Clean hardcoded database password from alembic.ini (env.py uses settings.DATABASE_URL)
- Add test_alembic_check_detects_drift with CI-only skipif
- Add DB-unavailable skip conditions to subprocess-based alembic tests

backend/alembic.ini

@@ -86,7 +86,7 @@ path_separator = os
 # database URL.  This is consumed by the user-maintained env.py script only.
 # other means of configuring database URLs may be customized within the env.py
 # file.
-sqlalchemy.url = postgresql+asyncpg://postgres:geo_pg_dev_2026@127.0.0.1:5433/geo_platform
+sqlalchemy.url = sqlite:///dummy.db
 
 
 [post_write_hooks]

backend/tests/test_infrastructure/test_database_migration.py

@@ -6,6 +6,58 @@ from pathlib import Path
 # 项目根目录
 PROJECT_ROOT = "/Users/Chiguyong/Code/Fischer/geo/backend"
 
+# 从迁移文件 a79329c23b20_initial_complete_schema.py 提取的实际表名
+REQUIRED_TABLES = [
+    "agent_registry",
+    "api_keys",
+    "brands",
+    "optimization_insights",
+    "organizations",
+    "platform_rule_versions",
+    "platform_rules",
+    "publish_records",
+    "alert_settings",
+    "alerts",
+    "competitor_insights",
+    "competitors",
+    "content_metrics",
+    "detection_tasks",
+    "diagnosis_records",
+    "monitoring_records",
+    "schema_suggestions",
+    "suggestions",
+    "trend_insights",
+    "usage_records",
+    "users",
+    "agent_configs",
+    "brand_knowledge",
+    "content_baselines",
+    "geo_plans",
+    "knowledge_bases",
+    "knowledge_search_logs",
+    "lifecycle_projects",
+    "org_members",
+    "payment_orders",
+    "queries",
+    "subscriptions",
+    "agent_tasks",
+    "citation_records",
+    "contents",
+    "geo_plan_actions",
+    "keywords",
+    "knowledge_documents",
+    "project_stages",
+    "query_tasks",
+    "agent_task_logs",
+    "attribution_records",
+    "content_reviews",
+    "content_versions",
+    "distribution_schedules",
+    "knowledge_chunks",
+    "knowledge_entities",
+    "knowledge_relations",
+]
+
 
 class TestDatabaseMigration:
     """数据库迁移验证测试"""
@@ -18,6 +70,8 @@ class TestDatabaseMigration:
             capture_output=True,
             text=True
         )
+        if result.returncode != 0 and "gaierror" in result.stderr:
+            pytest.skip("数据库不可用，跳过 alembic current 测试")
         assert result.returncode == 0, f"alembic current failed: {result.stderr}"
 
     def test_all_required_tables_exist(self):
@@ -31,7 +85,6 @@ class TestDatabaseMigration:
             from sqlalchemy import inspect
 
             async with engine.connect() as conn:
-                inspector = inspect(conn)
                 tables = await conn.run_sync(lambda sync_conn: inspect(sync_conn).get_table_names())
                 return set(tables)
 
@@ -40,16 +93,7 @@ class TestDatabaseMigration:
         except Exception as e:
             pytest.skip(f"无法连接数据库: {e}")
 
-        required_tables = [
-            "users", "brands", "competitors", "queries",
-            "citations", "alerts", "alert_settings", "suggestions",
-            "organizations", "subscriptions", "subscription_plans",
-            "agent_configs", "agent_executions", "knowledge_bases",
-            "content_versions", "platform_rules", "analytics_events",
-            "distribution_schedules", "client_brands"
-        ]
-
-        missing_tables = [t for t in required_tables if t not in existing_tables]
+        missing_tables = [t for t in REQUIRED_TABLES if t not in existing_tables]
         assert not missing_tables, f"缺失表: {missing_tables}"
 
     def test_migration_head_matches_models(self):
@@ -60,6 +104,8 @@ class TestDatabaseMigration:
             capture_output=True,
             text=True
         )
+        if result.returncode != 0 and "gaierror" in result.stderr:
+            pytest.skip("数据库不可用，跳过 alembic check 测试")
         assert result.returncode == 0, f"alembic check failed: {result.stderr}"
 
     def test_no_duplicate_migration_versions(self):
@@ -72,15 +118,11 @@ class TestDatabaseMigration:
         version_numbers = []
 
         for f in version_files:
-            # 版本号通常是目录名中下划线前的部分
-            # 例如: 059724556401_add_missing_sentiment_fields.py
             name = f.stem
-            # 取第一个下划线前的部分作为版本号
             if "_" in name:
                 version_num = name.split("_")[0]
                 version_numbers.append((version_num, f.name))
 
-        # 检查重复
         seen = {}
         duplicates = []
         for version_num, filename in version_numbers:
@@ -114,7 +156,6 @@ class TestDatabaseMigration:
 
         fk_columns = [fk['constrained_columns'] for fk in foreign_keys]
 
-        # 检查 brands.user_id 外键存在
         has_user_fk = any('user_id' in cols for cols in fk_columns)
         assert has_user_fk, "brands 表缺少 user_id 外键"
 
@@ -126,5 +167,24 @@ class TestDatabaseMigration:
             capture_output=True,
             text=True
         )
+        if result.returncode != 0 and "gaierror" in result.stderr:
+            pytest.skip("数据库不可用，跳过 alembic history 测试")
         assert result.returncode == 0, f"alembic history failed: {result.stderr}"
         assert result.stdout.strip(), "alembic history 为空"
+
+    @pytest.mark.skipif(
+        os.environ.get("CI") is None,
+        reason="alembic check drift detection 仅在 CI 环境中运行（需要完整数据库）"
+    )
+    def test_alembic_check_detects_drift(self):
+        """alembic check 应检测到模型与迁移之间的漂移"""
+        result = subprocess.run(
+            ["alembic", "check"],
+            cwd=PROJECT_ROOT,
+            capture_output=True,
+            text=True
+        )
+        # returncode 0 表示无漂移，非零表示存在漂移
+        assert result.returncode == 0, (
+            f"检测到迁移漂移！模型定义与迁移文件不一致:\n{result.stderr}\n{result.stdout}"
+        )