From c349e40fde7d4fa04d2043a37b2a71740f2abd85 Mon Sep 17 00:00:00 2001 From: chiguyong Date: Thu, 4 Jun 2026 22:17:02 +0800 Subject: [PATCH] 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 | 2 +- .../test_database_migration.py | 92 +++++++++++++++---- 2 files changed, 77 insertions(+), 17 deletions(-) diff --git a/backend/alembic.ini b/backend/alembic.ini index cb920b1..8eb783a 100644 --- a/backend/alembic.ini +++ b/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] diff --git a/backend/tests/test_infrastructure/test_database_migration.py b/backend/tests/test_infrastructure/test_database_migration.py index d412b2f..f658a8e 100644 --- a/backend/tests/test_infrastructure/test_database_migration.py +++ b/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}" + )