import json import os import subprocess import sys from pathlib import Path from flask import Flask from sqlalchemy import create_engine, text from database.manager import Base from services.market_intel import MarketIntelService from services.market_intel.service import MARKET_INTEL_TABLES from services.market_intel.adapters import get_adapter, get_adapter_summaries from services.market_intel.candidate_preview import build_candidate_preview_from_discovery from services.market_intel.discovery_runner import ManualDiscoveryRunner from services.market_intel.html_diagnostics import parse_html_diagnostics from services.market_intel.mcp_activation_runbook import build_mcp_activation_runbook_preview from services.market_intel.mcp_contract import build_mcp_tool_contract_preview from services.market_intel.mcp_deploy_preflight import build_mcp_deploy_preflight_plan from services.market_intel.mcp_fetch_gate import build_mcp_fetch_gate_preview from services.market_intel.mcp_readiness import build_mcp_readiness_plan from services.market_intel.live_db_inventory import build_live_db_inventory_preview from services.market_intel.platform_seed_db_diff import build_platform_seed_db_diff_plan from services.market_intel.schema_db_probe import build_schema_db_probe_plan TEST_APPROVAL_TOKEN = "test-market-intel-approval-token" def test_market_intel_defaults_are_safe(): service = MarketIntelService() status = service.get_runtime_status().to_dict() plan = service.build_dry_run_plan("momo") assert status["enabled"] is False assert status["crawler_enabled"] is False assert status["write_enabled"] is False assert status["database_write_allowed"] is False assert plan["would_discover_campaigns"] is False assert plan["would_write_database"] is False def test_market_intel_adapter_registry_is_read_only(): summaries = get_adapter_summaries() codes = {item["platform_code"] for item in summaries} assert {"momo", "pchome", "coupang", "shopee"} <= codes for summary in summaries: policy = summary["safety_policy"] assert policy["allow_login"] is False assert policy["allow_database_write"] is False assert policy["allow_scheduler_attach"] is False def test_market_intel_discovery_plan_does_not_allow_network_or_write(): service = MarketIntelService() plan = service.build_discovery_plan("momo") assert plan["found"] is True assert len(plan["plans"]) == 1 momo_plan = plan["plans"][0] assert momo_plan["network_request_allowed"] is False assert momo_plan["database_write_allowed"] is False assert momo_plan["scheduler_attach_allowed"] is False assert momo_plan["sources"] def test_unknown_adapter_returns_diagnostic_error(): assert get_adapter("unknown") is None plan = MarketIntelService().build_discovery_plan("unknown") assert plan["found"] is False assert plan["error"] == "未知平台 adapter" def test_manual_discovery_default_does_not_call_network(): called = {"count": 0} def fake_get(*args, **kwargs): called["count"] += 1 raise AssertionError("預設 dry-run 不應發 HTTP request") result = MarketIntelService().run_manual_discovery("momo", fetch=False, http_get=fake_get) assert called["count"] == 0 assert result["found"] is True assert result["runs"][0]["status"] == "planned" assert result["runs"][0]["sources_fetched"] == 0 assert all(item["network_executed"] is False for item in result["runs"][0]["results"]) def test_manual_discovery_fetch_is_blocked_when_flags_are_off(): called = {"count": 0} def fake_get(*args, **kwargs): called["count"] += 1 raise AssertionError("flags 關閉時不應發 HTTP request") result = MarketIntelService().run_manual_discovery("momo", fetch=True, http_get=fake_get) assert called["count"] == 0 assert result["runs"][0]["status"] == "blocked" assert result["runs"][0]["network_allowed"] is False assert result["runs"][0]["database_write_allowed"] is False assert result["mcp_fetch_gate"]["network_request_allowed"] is False assert result["runs"][0]["network_gate"]["manual_fetch_gate_open"] is False def test_manual_runner_fetch_can_be_blocked_by_mcp_gate_even_when_flags_allow(): class RuntimeStatus: enabled = True crawler_enabled = True called = {"count": 0} def fake_get(*args, **kwargs): called["count"] += 1 raise AssertionError("MCP gate 關閉時不應發 HTTP request") gate = { "manual_fetch_gate_open": False, "network_request_allowed": False, "blocked_reasons": ["mcp_router_enabled"], "operator_message": "人工 fetch 仍被 MCP gate 阻擋", } adapter = get_adapter("momo") runner = ManualDiscoveryRunner( runtime_status=RuntimeStatus(), http_get=fake_get, network_allowed_override=False, network_gate=gate, ) result = runner.run(adapter, fetch=True).to_dict() assert called["count"] == 0 assert result["status"] == "blocked" assert result["network_allowed"] is False assert result["network_gate"]["blocked_reasons"] == ["mcp_router_enabled"] assert "MCP gate" in result["error_message"] def test_manual_runner_fetch_uses_injected_http_get_when_allowed(): class RuntimeStatus: enabled = True crawler_enabled = True class Response: status_code = 200 text = "活動頁OK" called = {"count": 0} def fake_get(url, **kwargs): called["count"] += 1 return Response() adapter = get_adapter("momo") runner = ManualDiscoveryRunner(runtime_status=RuntimeStatus(), http_get=fake_get) result = runner.run(adapter, fetch=True).to_dict() assert called["count"] == len(adapter.campaign_sources()) assert result["status"] == "success" assert result["database_write_allowed"] is False assert result["scheduler_attached"] is False assert result["sources_fetched"] == len(adapter.campaign_sources()) assert result["results"][0]["title"] == "活動頁" assert result["results"][0]["diagnostics"]["title"] == "活動頁" def test_html_diagnostics_extracts_campaign_link_candidates(): html = """ 五月品牌日 品牌日活動 客服中心 外部 promo """ diagnostics = parse_html_diagnostics(html, base_url="https://shop.example").to_dict() assert diagnostics["title"] == "五月品牌日" assert diagnostics["link_count"] == 3 assert diagnostics["same_host_link_count"] == 1 assert diagnostics["campaign_link_candidates"][0]["href"] == "https://shop.example/event/brand-day" assert diagnostics["campaign_link_candidates"][0]["score"] > 0 assert "generic_score" in diagnostics["campaign_link_candidates"][0] assert "platform_score" in diagnostics["campaign_link_candidates"][0] assert "confidence_band" in diagnostics["campaign_link_candidates"][0] assert "confidence_reason" in diagnostics["campaign_link_candidates"][0] def test_manual_runner_returns_parser_diagnostics_when_fetch_succeeds(): class RuntimeStatus: enabled = True crawler_enabled = True class Response: status_code = 200 text = """ PChome 優惠活動 限時優惠 """ adapter = get_adapter("pchome") runner = ManualDiscoveryRunner(runtime_status=RuntimeStatus(), http_get=lambda *args, **kwargs: Response()) result = runner.run(adapter, fetch=True).to_dict() diagnostics = result["results"][0]["diagnostics"] assert diagnostics["title"] == "PChome 優惠活動" assert diagnostics["campaign_link_candidates"] assert diagnostics["campaign_link_candidates"][0]["is_same_host"] is True def test_momo_platform_scorer_prioritizes_momo_campaign_links(): adapter = get_adapter("momo") html = """ MOMO 活動 外部 promo MOMO 品牌日活動 """ diagnostics = parse_html_diagnostics( html, base_url=adapter.base_url, score_link=adapter.score_campaign_link, ).to_dict() first = diagnostics["campaign_link_candidates"][0] assert first["href"] == "https://www.momoshop.com.tw/edm/cmmedm.jsp" assert first["platform_score"] > 0 assert first["confidence_band"] == "high" def test_pchome_platform_scorer_prioritizes_region_campaign_links(): adapter = get_adapter("pchome") html = """ PChome 活動 外部 event PChome 24h 美妝優惠 """ diagnostics = parse_html_diagnostics( html, base_url=adapter.base_url, score_link=adapter.score_campaign_link, ).to_dict() first = diagnostics["campaign_link_candidates"][0] assert first["href"] == "https://24h.pchome.com.tw/region/DDAB" assert first["platform_score"] > 0 assert first["confidence_band"] == "high" def test_coupang_platform_scorer_prioritizes_official_campaign_links(): adapter = get_adapter("coupang") html = """ Coupang 活動 外部 event 酷澎 火箭跨境優惠 """ diagnostics = parse_html_diagnostics( html, base_url=adapter.base_url, score_link=adapter.score_campaign_link, ).to_dict() first = diagnostics["campaign_link_candidates"][0] assert first["href"] == "https://www.tw.coupang.com/np/coupangglobal" assert first["platform_score"] > 0 assert first["confidence_band"] == "high" def test_shopee_platform_scorer_prioritizes_mall_campaign_links(): adapter = get_adapter("shopee") html = """ Shopee 活動 外部 event 蝦皮商城 品牌限時優惠 """ diagnostics = parse_html_diagnostics( html, base_url=adapter.base_url, score_link=adapter.score_campaign_link, ).to_dict() first = diagnostics["campaign_link_candidates"][0] assert first["href"] == "https://shopee.tw/mall" assert first["platform_score"] > 0 assert first["confidence_band"] == "high" def test_confidence_bands_cover_high_medium_low(): adapter = get_adapter("momo") html = """ 信心帶測試 MOMO 限時品牌日活動優惠 活動 清單 """ diagnostics = parse_html_diagnostics( html, base_url=adapter.base_url, score_link=adapter.score_campaign_link, ).to_dict() bands = {item["href"]: item["confidence_band"] for item in diagnostics["campaign_link_candidates"]} assert bands["https://www.momoshop.com.tw/edm/cmmedm.jsp"] == "high" assert bands["https://neutral.example/event-light"] == "medium" assert bands["https://other.example/sale"] == "low" for item in diagnostics["campaign_link_candidates"]: assert item["confidence_reason"] def test_candidate_preview_default_is_empty_and_does_not_call_network(): called = {"count": 0} def fake_get(*args, **kwargs): called["count"] += 1 raise AssertionError("candidate preview 預設不應發 HTTP request") preview = MarketIntelService().build_candidate_preview("momo", fetch=False, http_get=fake_get) assert called["count"] == 0 assert preview["candidate_count"] == 0 assert preview["database_write_allowed"] is False assert preview["scheduler_attached"] is False assert preview["mcp_fetch_gate_open"] is False assert preview["mcp_fetch_gate"]["network_request_allowed"] is False assert preview["run_statuses"][0]["status"] == "planned" def test_candidate_preview_fetch_is_blocked_when_flags_are_off(): called = {"count": 0} def fake_get(*args, **kwargs): called["count"] += 1 raise AssertionError("flags 關閉時 candidate preview 不應發 HTTP request") preview = MarketIntelService().build_candidate_preview("momo", fetch=True, http_get=fake_get) assert called["count"] == 0 assert preview["candidate_count"] == 0 assert preview["run_statuses"][0]["status"] == "blocked" assert preview["mcp_fetch_gate_open"] is False assert "market_intel_enabled" in preview["mcp_fetch_gate"]["blocked_reasons"] def test_candidate_preview_aggregates_and_filters_by_band(): discovery = { "platform_code": "all", "fetch_requested": True, "manual_fetch_allowed": True, "runs": [ { "platform_code": "momo", "status": "success", "sources_planned": 1, "sources_fetched": 1, "errors": 0, "results": [ { "source_key": "momo_edm", "name": "MOMO EDM", "url": "https://www.momoshop.com.tw/edm/cmmedm.jsp", "status": "fetched", "diagnostics": { "title": "MOMO 活動", "page_hash": "hash-a", "campaign_link_candidates": [ { "href": "https://www.momoshop.com.tw/edm/a", "text": "品牌日", "is_same_host": True, "score": 20, "generic_score": 6, "platform_score": 14, "confidence_band": "high", "confidence_reason": "same_host", }, { "href": "https://other.example/sale", "text": "清單", "is_same_host": False, "score": 2, "generic_score": 2, "platform_score": 0, "confidence_band": "low", "confidence_reason": "external_host", }, ], }, } ], } ], } preview = build_candidate_preview_from_discovery(discovery, min_band="medium", limit=10) assert preview["candidate_count"] == 1 assert preview["candidates"][0]["confidence_band"] == "high" assert preview["candidates"][0]["platform_code"] == "momo" assert preview["candidates"][0]["source_key"] == "momo_edm" def test_market_intel_preview_template_uses_safe_fetch_false_endpoint(): template = Path("templates/market_intel/disabled.html").read_text(encoding="utf-8") assert "data-market-intel-preview" in template assert "data-market-intel-writer" in template assert "data-market-intel-cli" in template assert "data-market-intel-cli-body" in template assert "data-market-intel-db-probe" in template assert "data-market-intel-db-probe-body" in template assert "data-market-intel-seed-diff" in template assert "data-market-intel-seed-diff-body" in template assert "data-market-intel-mcp-readiness" in template assert "data-market-intel-mcp-servers" in template assert "data-market-intel-mcp-checks" in template assert "data-market-intel-mcp-tools" in template assert "market_intel_contract" in template assert "data-market-intel-mcp-preflight" in template assert "data-market-intel-mcp-preflight-env" in template assert "data-market-intel-mcp-preflight-services" in template assert "data-market-intel-mcp-preflight-ports" in template assert "data-market-intel-mcp-preflight-fallback" in template assert "data-market-intel-mcp-activation" in template assert "data-market-intel-mcp-activation-stages" in template assert "data-market-intel-mcp-activation-safety" in template assert "data-market-intel-mcp-activation-fallback" in template assert "data-market-intel-mcp-fetch-gate" in template assert "data-market-intel-mcp-fetch-gate-checks" in template assert "data-market-intel-mcp-fetch-gate-sequence" in template assert "data-market-intel-mcp-fetch-gate-readiness" in template assert "data-market-intel-scheduler" in template assert "data-market-intel-scheduler-checks" in template assert "data-market-intel-scheduler-jobs" in template assert "data-market-intel-scheduler-sequence" in template assert "data-market-intel-scheduler-fallback" in template assert "data-market-intel-match-review" in template assert "data-market-intel-match-review-checks" in template assert "data-market-intel-match-review-signals" in template assert "data-market-intel-match-review-actions" in template assert "data-market-intel-match-review-sequence" in template assert "data-market-intel-opportunity" in template assert "data-market-intel-opportunity-checks" in template assert "data-market-intel-opportunity-rules" in template assert "data-market-intel-opportunity-severity" in template assert "data-market-intel-opportunity-sequence" in template assert "data-market-intel-opportunity-scoring" in template assert "data-market-intel-opportunity-scoring-checks" in template assert "data-market-intel-opportunity-scoring-dimensions" in template assert "data-market-intel-opportunity-scoring-thresholds" in template assert "data-market-intel-opportunity-scoring-evidence" in template assert "data-market-intel-opportunity-scoring-sequence" in template assert "data-market-intel-opportunity-evidence" in template assert "data-market-intel-opportunity-evidence-checks" in template assert "data-market-intel-opportunity-evidence-sections" in template assert "data-market-intel-opportunity-evidence-escalation" in template assert "data-market-intel-opportunity-evidence-tables" in template assert "data-market-intel-opportunity-evidence-sequence" in template assert "data-market-intel-opportunity-alert" in template assert "data-market-intel-opportunity-alert-checks" in template assert "data-market-intel-opportunity-alert-channels" in template assert "data-market-intel-opportunity-alert-escalation" in template assert "data-market-intel-opportunity-alert-payload" in template assert "data-market-intel-opportunity-alert-review" in template assert "data-market-intel-opportunity-alert-actions" in template assert "data-market-intel-opportunity-alert-queue-contract" in template assert "data-market-intel-opportunity-alert-priority-lanes" in template assert "data-market-intel-opportunity-alert-queue-indexes" in template assert "data-market-intel-opportunity-alert-approval" in template assert "data-market-intel-opportunity-alert-sequence" in template assert "data-market-intel-migration" in template assert "data-market-intel-migration-tables" in template assert "data-market-intel-migration-drill" in template assert "data-market-intel-migration-drill-checks" in template assert "data-market-intel-migration-drill-preapply" in template assert "data-market-intel-migration-drill-rollback" in template assert "data-market-intel-catalog-review" in template assert "data-market-intel-catalog-review-checks" in template assert "data-market-intel-catalog-review-findings" in template assert "data-market-intel-catalog-review-tables" in template assert "data-market-intel-live-smoke" in template assert "data-market-intel-live-smoke-checks" in template assert "data-market-intel-live-smoke-findings" in template assert "data-market-intel-live-smoke-targets" in template assert "data-market-intel-live-inventory" in template assert "data-market-intel-live-inventory-checks" in template assert "data-market-intel-live-inventory-tables" in template assert "data-market-intel-live-inventory-platforms" in template assert "data-market-intel-live-inventory-alerts" in template assert "data-market-intel-manual-sample" in template assert "data-market-intel-manual-sample-checks" in template assert "data-market-intel-manual-sample-platforms" in template assert "data-market-intel-manual-sample-sequence" in template assert "data-market-intel-sample-acceptance" in template assert "data-market-intel-sample-acceptance-checks" in template assert "data-market-intel-sample-acceptance-fields" in template assert "data-market-intel-sample-acceptance-rules" in template assert "data-market-intel-approval" in template assert "data-market-intel-approval-gates" in template assert "data-market-intel-deploy" in template assert "data-market-intel-deploy-steps" in template assert "data-market-intel-deploy-fallback" in template assert "market_intel.market_intel_candidate_preview" in template assert "market_intel.market_intel_platform_seed_writer_plan" in template assert "market_intel.market_intel_seed_writer_cli_status" in template assert "market_intel.market_intel_schema_db_probe" in template assert "market_intel.market_intel_platform_seed_db_diff" in template assert "market_intel.market_intel_legacy_source_bridge" in template assert "market_intel.market_intel_mcp_readiness" in template assert "market_intel.market_intel_mcp_deploy_preflight" in template assert "market_intel.market_intel_mcp_activation_runbook" in template assert "market_intel.market_intel_mcp_fetch_gate" in template assert "market_intel.market_intel_manual_sample_plan" in template assert "market_intel.market_intel_manual_sample_acceptance" in template assert "market_intel.market_intel_scheduler_plan" in template assert "market_intel.market_intel_match_review_plan" in template assert "market_intel.market_intel_opportunity_plan" in template assert "market_intel.market_intel_opportunity_scoring_plan" in template assert "market_intel.market_intel_opportunity_evidence_plan" in template assert "market_intel.market_intel_opportunity_alert_plan" in template assert "market_intel.market_intel_migration_blueprint" in template assert "market_intel.market_intel_migration_apply_drill" in template assert "market_intel.market_intel_migration_catalog_review" in template assert "market_intel.market_intel_migration_live_smoke" in template assert "market_intel.market_intel_live_db_inventory" in template assert "market_intel.market_intel_write_approval_runbook" in template assert "market_intel.market_intel_deployment_readiness" in template assert "required_manual_steps" in template assert "fallback_plan" in template assert "approval_gates" in template assert "備援方案" in template assert "fetch=false" in template assert "fetch=true" not in template assert "execute=false" in template assert "execute=true" not in template assert "writes=executed" not in template assert "API 不執行推版" in template def test_market_intel_schema_metadata_contains_all_market_tables(): metadata_tables = set(Base.metadata.tables) assert set(MARKET_INTEL_TABLES) <= metadata_tables def test_legacy_source_bridge_default_is_planned_only(): bridge = MarketIntelService().build_legacy_source_bridge() assert bridge["mode"] == "legacy_source_bridge_planned" assert bridge["phase"] == "phase_48_manual_sample_acceptance_contract" assert bridge["execute_requested"] is False assert bridge["read_only_query_executed"] is False assert bridge["database_connection_opened"] is False assert bridge["database_session_created"] is False assert bridge["database_write_executed"] is False assert bridge["database_commit_executed"] is False assert bridge["external_network_executed"] is False assert bridge["scheduler_attached"] is False assert bridge["writes_executed"] is False assert bridge["would_write_database"] is False assert bridge["source_count"] == 3 assert bridge["existing_source_count"] == 0 assert "execute_false_planned_only" in bridge["blocked_reasons"] assert {item["table"] for item in bridge["source_summaries"]} == { "promo_products", "competitor_prices", "competitor_price_history", } def test_legacy_source_bridge_read_only_sqlite_counts_sources(): engine = create_engine("sqlite:///:memory:") with engine.begin() as conn: conn.execute( text( """ CREATE TABLE promo_products ( id INTEGER PRIMARY KEY, batch_id TEXT, crawled_at TEXT, time_slot TEXT, activity_time_text TEXT, session_time_text TEXT, i_code TEXT, name TEXT, price INTEGER, discount_text TEXT, url TEXT, image_url TEXT, previous_price INTEGER, remain_qty INTEGER, status_change TEXT, page_type TEXT ) """ ) ) conn.execute( text( """ CREATE TABLE competitor_prices ( id INTEGER PRIMARY KEY, sku TEXT, source TEXT, price NUMERIC, original_price NUMERIC, discount_pct INTEGER, competitor_product_id TEXT, competitor_product_name TEXT, match_score NUMERIC, crawled_at TEXT ) """ ) ) conn.execute( text( """ CREATE TABLE competitor_price_history ( id INTEGER PRIMARY KEY, sku TEXT, source TEXT, momo_product_id INTEGER, momo_price NUMERIC, price NUMERIC, original_price NUMERIC, discount_pct INTEGER, competitor_product_id TEXT, competitor_product_name TEXT, match_score NUMERIC, crawled_at TEXT ) """ ) ) conn.execute( text( """ INSERT INTO promo_products (batch_id, crawled_at, time_slot, i_code, name, price, discount_text, url, page_type) VALUES ('batch-1', '2026-05-18 10:00:00', '10:00', 'A001', 'MOMO 商品 A', 399, '8折', 'https://momo/A001', 'edm'), ('batch-1', '2026-05-18 10:01:00', '10:00', 'A002', 'MOMO 商品 B', 499, '9折', 'https://momo/A002', 'edm') """ ) ) conn.execute( text( """ INSERT INTO competitor_prices (sku, source, price, original_price, competitor_product_id, competitor_product_name, match_score, crawled_at) VALUES ('A001', 'pchome', 379, 459, 'PC-A001', 'PChome 商品 A', 0.82, '2026-05-18 11:00:00') """ ) ) conn.execute( text( """ INSERT INTO competitor_price_history (sku, source, momo_product_id, momo_price, price, original_price, competitor_product_id, competitor_product_name, match_score, crawled_at) VALUES ('A001', 'pchome', 1, 399, 379, 459, 'PC-A001', 'PChome 商品 A', 0.82, '2026-05-18 11:00:00') """ ) ) bridge = MarketIntelService().build_legacy_source_bridge( execute_requested=True, engine=engine, database_type="sqlite", ) assert bridge["mode"] == "legacy_source_bridge_read_only" assert bridge["execute_requested"] is True assert bridge["read_only_query_executed"] is True assert bridge["database_connection_opened"] is True assert bridge["database_session_created"] is False assert bridge["database_write_executed"] is False assert bridge["database_commit_executed"] is False assert bridge["external_network_executed"] is False assert bridge["scheduler_attached"] is False assert bridge["writes_executed"] is False assert bridge["would_write_database"] is False assert bridge["source_tables_ready"] is True assert bridge["existing_source_count"] == 3 assert bridge["total_existing_rows"] == 4 assert [item["table"] for item in bridge["source_summaries"]] == [ "promo_products", "competitor_prices", "competitor_price_history", ] assert bridge["source_summaries"][0]["row_count"] == 2 assert bridge["source_summaries"][0]["distinct_entity_count"] == 2 assert bridge["source_summaries"][1]["row_count"] == 1 assert bridge["source_summaries"][2]["row_count"] == 1 assert len(bridge["bridge_operations"]) == 4 assert all( item["write_status"] != "executed" for item in bridge["bridge_operations"] ) def test_mcp_tool_contract_preview_is_read_only_and_whitelisted(): contract = MarketIntelService().build_mcp_tool_contract() assert contract["mode"] == "mcp_tool_contract_preview" assert contract["phase"] == "phase_48_manual_sample_acceptance_contract" assert contract["caller"] == "market_intel" assert contract["contract_ready"] is True assert contract["blocked_reasons"] == [] assert contract["tool_count"] == 3 assert contract["database_session_created"] is False assert contract["database_write_executed"] is False assert contract["database_commit_executed"] is False assert contract["external_network_executed"] is False assert contract["scheduler_attached"] is False assert contract["writes_executed"] is False assert contract["would_write_database"] is False assert contract["checks"]["market_intel_caller_registered"] is True assert contract["checks"]["all_router_tools_whitelisted"] is True assert contract["checks"]["all_tools_read_only"] is True assert contract["checks"]["no_database_write_allowed"] is True assert contract["checks"]["no_scheduler_attach_allowed"] is True assert {item["name"] for item in contract["tools"]} == { "market_campaign_search", "market_campaign_scrape", "market_product_match_lookup", } assert all(item["router_tools_whitelisted"] is True for item in contract["tools"]) assert all(item["write_risk"] is False for item in contract["tools"]) def test_mcp_tool_contract_detects_missing_router_whitelist(): contract = build_mcp_tool_contract_preview(tool_registry={}) assert contract["contract_ready"] is False assert "market_intel_caller_registered" in contract["blocked_reasons"] assert "all_router_tools_whitelisted" in contract["blocked_reasons"] def test_mcp_tool_contract_route_is_preview_only(): from routes.market_intel_routes import market_intel_bp app = Flask(__name__) app.secret_key = "test-secret" app.register_blueprint(market_intel_bp) client = app.test_client() with client.session_transaction() as session: session["logged_in"] = True response = client.get("/api/market_intel/mcp_tool_contract") data = response.get_json() assert response.status_code == 200 assert data["mode"] == "mcp_tool_contract_preview" assert data["contract_ready"] is True assert data["database_session_created"] is False assert data["database_write_executed"] is False assert data["database_commit_executed"] is False assert data["external_network_executed"] is False assert data["scheduler_attached"] is False def test_mcp_activation_runbook_blocks_until_env_and_health_pass(): preflight = build_mcp_deploy_preflight_plan(env={}) runbook = build_mcp_activation_runbook_preview(preflight=preflight) assert runbook["mode"] == "mcp_activation_runbook_preview" assert runbook["ready_for_operator_activation"] is False assert "required_env_ready" in runbook["blocked_reasons"] assert "external_health_smoke_not_executed" in runbook["blocked_reasons"] assert runbook["deployment_actions_executed"] is False assert runbook["docker_command_executed"] is False assert runbook["ssh_command_executed"] is False assert runbook["database_session_created"] is False assert runbook["database_write_executed"] is False assert runbook["database_commit_executed"] is False assert runbook["external_network_executed"] is False assert runbook["scheduler_attached"] is False assert runbook["would_write_database"] is False assert runbook["safety_contract"]["does_not_execute_docker"] is True assert runbook["safety_contract"]["does_not_enable_router"] is True assert {item["key"] for item in runbook["stages"]} == { "configure_required_env", "create_mcp_readonly_role", "start_external_mcp_stack", "run_mcp_health_smoke", "enable_mcp_router_flag", "run_market_intel_mcp_smoke", } def test_mcp_activation_runbook_can_be_ready_with_mocked_gates(): preflight = build_mcp_deploy_preflight_plan( env={ "MCP_POSTGRES_PASSWORD": "secret", "TAVILY_API_KEY": "tavily", "EXA_API_KEY": "exa", "MCP_ROUTER_ENABLED": "false", } ) readiness = { "readiness_checks": { "external_server_health_checked": True, "external_servers_all_healthy": True, "market_intel_tool_contract_ready": True, } } runbook = build_mcp_activation_runbook_preview( preflight=preflight, readiness=readiness, ) assert runbook["ready_for_operator_activation"] is True assert runbook["blocked_reasons"] == [] assert runbook["checks"]["required_env_ready"] is True assert runbook["checks"]["all_health_passed"] is True assert { item["key"]: item["status"] for item in runbook["stages"] }["enable_mcp_router_flag"] == "ready" def test_mcp_activation_runbook_route_is_preview_only(): from routes.market_intel_routes import market_intel_bp app = Flask(__name__) app.secret_key = "test-secret" app.register_blueprint(market_intel_bp) client = app.test_client() with client.session_transaction() as session: session["logged_in"] = True response = client.get("/api/market_intel/mcp_activation_runbook") data = response.get_json() assert response.status_code == 200 assert data["mode"] == "mcp_activation_runbook_preview" assert data["phase"] == "phase_48_manual_sample_acceptance_contract" assert data["deployment_actions_executed"] is False assert data["docker_command_executed"] is False assert data["ssh_command_executed"] is False assert data["database_write_executed"] is False assert data["external_network_executed"] is False assert data["scheduler_attached"] is False def test_mcp_fetch_gate_default_blocks_external_fetch(): gate = MarketIntelService().build_mcp_fetch_gate(fetch_requested=True) assert gate["mode"] == "mcp_fetch_gate_planned" assert gate["phase"] == "phase_48_manual_sample_acceptance_contract" assert gate["fetch_requested"] is True assert gate["manual_fetch_gate_open"] is False assert gate["network_request_allowed"] is False assert gate["would_use_external_network"] is False assert gate["database_session_created"] is False assert gate["database_write_executed"] is False assert gate["database_commit_executed"] is False assert gate["external_network_executed"] is False assert gate["scheduler_attached"] is False assert gate["writes_executed"] is False assert gate["would_write_database"] is False assert "market_intel_enabled" in gate["blocked_reasons"] assert "market_intel_crawler_enabled" in gate["blocked_reasons"] assert "mcp_readiness_executed" in gate["blocked_reasons"] assert "mcp_router_enabled" in gate["blocked_reasons"] def test_mcp_fetch_gate_can_open_with_mocked_ready_state(): class RuntimeStatus: enabled = True crawler_enabled = True database_write_allowed = False scheduler_attached = False readiness = { "mode": "mcp_readiness_read_only", "execute_requested": True, "router_enabled": True, "external_mcp_complete": True, "internal_mcp_complete": True, "market_intel_mcp_integrated": True, "blocked_reasons": [], "readiness_checks": { "market_intel_tool_contract_ready": True, "external_servers_all_healthy": True, }, } gate = build_mcp_fetch_gate_preview( RuntimeStatus(), fetch_requested=True, execute_readiness=True, readiness=readiness, ) assert gate["mode"] == "mcp_fetch_gate_read_only" assert gate["manual_fetch_prerequisites_met"] is True assert gate["manual_fetch_gate_open"] is True assert gate["network_request_allowed"] is True assert gate["blocked_reasons"] == [] assert gate["database_write_executed"] is False assert gate["scheduler_attached"] is False def test_mcp_fetch_gate_route_is_preview_only(): from routes.market_intel_routes import market_intel_bp app = Flask(__name__) app.secret_key = "test-secret" app.register_blueprint(market_intel_bp) client = app.test_client() with client.session_transaction() as session: session["logged_in"] = True response = client.get("/api/market_intel/mcp_fetch_gate") data = response.get_json() assert response.status_code == 200 assert data["mode"] == "mcp_fetch_gate_planned" assert data["phase"] == "phase_48_manual_sample_acceptance_contract" assert data["fetch_requested"] is False assert data["network_request_allowed"] is False assert data["external_network_executed"] is False assert data["database_write_executed"] is False assert data["scheduler_attached"] is False def test_manual_sample_plan_preview_blocks_fetch_and_write(): plan = MarketIntelService().build_manual_sample_plan() assert plan["mode"] == "manual_sample_fetch_plan_preview" assert plan["phase"] == "phase_48_manual_sample_acceptance_contract" assert plan["ready_for_manual_sample_fetch"] is False assert plan["sample_fetch_executed"] is False assert plan["external_network_executed"] is False assert plan["database_session_created"] is False assert plan["database_write_executed"] is False assert plan["database_commit_executed"] is False assert plan["scheduler_attached"] is False assert plan["writes_executed"] is False assert plan["would_write_database"] is False assert plan["platform_count"] == 4 assert plan["sample_source_total"] == 4 assert plan["sample_policy"]["max_platforms_per_manual_batch"] == 1 assert plan["sample_policy"]["first_batch_platform"] == "pchome" assert [item["platform_code"] for item in plan["sample_platforms"]] == [ "pchome", "momo", "coupang", "shopee", ] assert all( item["sample_source_count"] == 1 and item["network_status"] == "not_executed" and item["write_status"] == "blocked_preview_only" for item in plan["sample_platforms"] ) assert "market_intel_enabled" in plan["blocked_reasons"] assert "mcp_fetch_gate_open" in plan["blocked_reasons"] assert "sample_fetch_not_executed_by_api" in plan["blocked_reasons"] assert "do_not_fetch_from_api_preview" in plan["safe_boundaries"] assert "do_not_bypass_anti_bot" in plan["safe_boundaries"] def test_manual_sample_plan_route_is_preview_only(): from routes.market_intel_routes import market_intel_bp app = Flask(__name__) app.secret_key = "test-secret" app.register_blueprint(market_intel_bp) client = app.test_client() with client.session_transaction() as session: session["logged_in"] = True response = client.get("/api/market_intel/manual_sample_plan") data = response.get_json() assert response.status_code == 200 assert data["mode"] == "manual_sample_fetch_plan_preview" assert data["phase"] == "phase_48_manual_sample_acceptance_contract" assert data["sample_fetch_executed"] is False assert data["external_network_executed"] is False assert data["database_write_executed"] is False assert data["scheduler_attached"] is False def test_manual_sample_acceptance_preview_blocks_candidate_import(): acceptance = MarketIntelService().build_manual_sample_acceptance() assert acceptance["mode"] == "manual_sample_acceptance_preview" assert acceptance["phase"] == "phase_48_manual_sample_acceptance_contract" assert acceptance["contract_ready"] is True assert acceptance["sample_result_loaded"] is False assert acceptance["sample_result_accepted"] is False assert acceptance["candidate_import_allowed"] is False assert acceptance["external_network_executed"] is False assert acceptance["database_connection_opened"] is False assert acceptance["database_session_created"] is False assert acceptance["database_write_executed"] is False assert acceptance["database_commit_executed"] is False assert acceptance["scheduler_attached"] is False assert acceptance["writes_executed"] is False assert acceptance["would_write_database"] is False assert "sample_result_not_loaded" in acceptance["blocked_reasons"] assert "manual_review_required_before_import" in acceptance["blocked_reasons"] assert "status_code" in acceptance["required_result_fields"] assert "campaign_link_candidates" in acceptance["required_diagnostic_fields"] assert acceptance["acceptance_thresholds"]["minimum_campaign_candidates"] == 1 assert acceptance["acceptance_thresholds"]["accepted_candidate_bands"] == ["high", "medium"] assert {item["key"] for item in acceptance["reject_conditions"]} >= { "anti_bot_or_login_wall", "no_campaign_candidates", } assert "do_not_import_candidates_without_human_review" in acceptance["safe_boundaries"] assert "do_not_write_market_tables_from_acceptance_preview" in acceptance["safe_boundaries"] def test_manual_sample_acceptance_route_is_preview_only(): from routes.market_intel_routes import market_intel_bp app = Flask(__name__) app.secret_key = "test-secret" app.register_blueprint(market_intel_bp) client = app.test_client() with client.session_transaction() as session: session["logged_in"] = True response = client.get("/api/market_intel/manual_sample_acceptance") data = response.get_json() assert response.status_code == 200 assert data["mode"] == "manual_sample_acceptance_preview" assert data["phase"] == "phase_48_manual_sample_acceptance_contract" assert data["sample_result_loaded"] is False assert data["candidate_import_allowed"] is False assert data["external_network_executed"] is False assert data["database_write_executed"] is False assert data["scheduler_attached"] is False def test_scheduler_plan_preview_blocks_job_attachment(): plan = MarketIntelService().build_scheduler_plan() assert plan["mode"] == "scheduler_attach_plan_preview" assert plan["phase"] == "phase_48_manual_sample_acceptance_contract" assert plan["ready_to_attach_scheduler"] is False assert plan["scheduler_attached"] is False assert plan["scheduler_registration_executed"] is False assert plan["crawler_job_started"] is False assert plan["external_network_executed"] is False assert plan["database_session_created"] is False assert plan["database_write_executed"] is False assert plan["database_commit_executed"] is False assert plan["writes_executed"] is False assert plan["would_write_database"] is False assert plan["job_count"] == 3 assert {item["key"] for item in plan["jobs"]} == { "campaign_discovery_daily", "campaign_product_probe", "product_match_review_seed", } assert "market_intel_enabled" in plan["blocked_reasons"] assert "mcp_fetch_gate_open" in plan["blocked_reasons"] assert "manual_operator_approval" in plan["blocked_reasons"] assert "do_not_attach_scheduler_from_api" in plan["safe_boundaries"] def test_scheduler_plan_route_is_preview_only(): from routes.market_intel_routes import market_intel_bp app = Flask(__name__) app.secret_key = "test-secret" app.register_blueprint(market_intel_bp) client = app.test_client() with client.session_transaction() as session: session["logged_in"] = True response = client.get("/api/market_intel/scheduler_plan") data = response.get_json() assert response.status_code == 200 assert data["mode"] == "scheduler_attach_plan_preview" assert data["phase"] == "phase_48_manual_sample_acceptance_contract" assert data["scheduler_registration_executed"] is False assert data["crawler_job_started"] is False assert data["external_network_executed"] is False assert data["database_write_executed"] is False def test_match_review_plan_preview_blocks_auto_confirm(): plan = MarketIntelService().build_match_review_plan() assert plan["mode"] == "match_review_plan_preview" assert plan["phase"] == "phase_48_manual_sample_acceptance_contract" assert plan["ready_for_review_queue"] is False assert plan["review_queue_created"] is False assert plan["auto_match_executed"] is False assert plan["auto_confirm_executed"] is False assert plan["database_session_created"] is False assert plan["database_write_executed"] is False assert plan["database_commit_executed"] is False assert plan["external_network_executed"] is False assert plan["scheduler_attached"] is False assert plan["writes_executed"] is False assert plan["would_write_database"] is False assert plan["thresholds"]["auto_confirm_enabled"] is False assert len(plan["scoring_signals"]) >= 5 assert "needs_review" in plan["planned_status_flow"] assert "confirmed" in plan["planned_status_flow"] assert "manual_operator_approval" in plan["blocked_reasons"] assert "do_not_auto_confirm_matches" in plan["safe_boundaries"] assert all(item["requires_operator"] for item in plan["review_actions"]) def test_match_review_plan_route_is_preview_only(): from routes.market_intel_routes import market_intel_bp app = Flask(__name__) app.secret_key = "test-secret" app.register_blueprint(market_intel_bp) client = app.test_client() with client.session_transaction() as session: session["logged_in"] = True response = client.get("/api/market_intel/match_review_plan") data = response.get_json() assert response.status_code == 200 assert data["mode"] == "match_review_plan_preview" assert data["phase"] == "phase_48_manual_sample_acceptance_contract" assert data["review_queue_created"] is False assert data["auto_confirm_executed"] is False assert data["external_network_executed"] is False assert data["database_write_executed"] is False def test_opportunity_plan_preview_blocks_alerts_and_ai_summary(): plan = MarketIntelService().build_opportunity_plan() assert plan["mode"] == "opportunity_plan_preview" assert plan["phase"] == "phase_48_manual_sample_acceptance_contract" assert plan["ready_for_opportunity_queue"] is False assert plan["opportunity_queue_created"] is False assert plan["threat_alert_dispatched"] is False assert plan["telegram_dispatched"] is False assert plan["ai_summary_generated"] is False assert plan["database_session_created"] is False assert plan["database_write_executed"] is False assert plan["database_commit_executed"] is False assert plan["external_network_executed"] is False assert plan["scheduler_attached"] is False assert plan["writes_executed"] is False assert plan["would_write_database"] is False assert plan["rule_count"] == 4 assert {item["key"] for item in plan["rules"]} == { "competitor_price_threat", "promotion_gap", "deep_discount_overlap", "campaign_ending_watch", } assert "match_review_candidates_available" in plan["blocked_reasons"] assert "manual_operator_approval" in plan["blocked_reasons"] assert "do_not_dispatch_alerts_from_preview" in plan["safe_boundaries"] assert "do_not_auto_adjust_price" in plan["safe_boundaries"] def test_opportunity_plan_route_is_preview_only(): from routes.market_intel_routes import market_intel_bp app = Flask(__name__) app.secret_key = "test-secret" app.register_blueprint(market_intel_bp) client = app.test_client() with client.session_transaction() as session: session["logged_in"] = True response = client.get("/api/market_intel/opportunity_plan") data = response.get_json() assert response.status_code == 200 assert data["mode"] == "opportunity_plan_preview" assert data["phase"] == "phase_48_manual_sample_acceptance_contract" assert data["opportunity_queue_created"] is False assert data["threat_alert_dispatched"] is False assert data["ai_summary_generated"] is False assert data["database_write_executed"] is False def test_opportunity_scoring_plan_preview_blocks_scoring_and_alerts(): plan = MarketIntelService().build_opportunity_scoring_plan() assert plan["mode"] == "opportunity_scoring_plan_preview" assert plan["phase"] == "phase_48_manual_sample_acceptance_contract" assert plan["ready_for_scoring_job"] is False assert plan["scoring_job_created"] is False assert plan["score_calculation_executed"] is False assert plan["sample_scores_generated"] is False assert plan["opportunity_queue_created"] is False assert plan["review_queue_created"] is False assert plan["threat_alert_dispatched"] is False assert plan["telegram_dispatched"] is False assert plan["ai_summary_generated"] is False assert plan["database_session_created"] is False assert plan["database_write_executed"] is False assert plan["database_commit_executed"] is False assert plan["external_network_executed"] is False assert plan["scheduler_attached"] is False assert plan["writes_executed"] is False assert plan["would_write_database"] is False assert plan["dimension_count"] == 5 assert plan["total_weight"] == 100 assert {item["key"] for item in plan["dimensions"]} == { "price_gap_pct", "discount_depth", "match_confidence", "campaign_urgency", "data_freshness", } assert "opportunity_queue_ready" in plan["blocked_reasons"] assert "confirmed_match_source_ready" in plan["blocked_reasons"] assert "manual_operator_approval" in plan["blocked_reasons"] assert "do_not_write_scoring_results_from_preview" in plan["safe_boundaries"] assert "do_not_auto_adjust_price" in plan["safe_boundaries"] def test_opportunity_scoring_plan_route_is_preview_only(): from routes.market_intel_routes import market_intel_bp app = Flask(__name__) app.secret_key = "test-secret" app.register_blueprint(market_intel_bp) client = app.test_client() with client.session_transaction() as session: session["logged_in"] = True response = client.get("/api/market_intel/opportunity_scoring_plan") data = response.get_json() assert response.status_code == 200 assert data["mode"] == "opportunity_scoring_plan_preview" assert data["phase"] == "phase_48_manual_sample_acceptance_contract" assert data["scoring_job_created"] is False assert data["score_calculation_executed"] is False assert data["sample_scores_generated"] is False assert data["database_write_executed"] is False assert data["external_network_executed"] is False def test_opportunity_evidence_plan_preview_blocks_queries_and_alerts(): plan = MarketIntelService().build_opportunity_evidence_plan() assert plan["mode"] == "opportunity_evidence_plan_preview" assert plan["phase"] == "phase_48_manual_sample_acceptance_contract" assert plan["ready_for_evidence_bundle"] is False assert plan["evidence_bundle_created"] is False assert plan["evidence_query_executed"] is False assert plan["sample_evidence_generated"] is False assert plan["alert_candidate_created"] is False assert plan["telegram_dispatched"] is False assert plan["ai_summary_generated"] is False assert plan["database_session_created"] is False assert plan["database_write_executed"] is False assert plan["database_commit_executed"] is False assert plan["external_network_executed"] is False assert plan["scheduler_attached"] is False assert plan["writes_executed"] is False assert plan["would_write_database"] is False assert plan["section_count"] == 5 assert {item["key"] for item in plan["sections"]} == { "campaign_context", "market_product_snapshot", "match_review_state", "momo_reference", "scoring_trace", } assert "confirmed_match_evidence_available" in plan["blocked_reasons"] assert "scoring_trace_available" in plan["blocked_reasons"] assert "manual_operator_approval" in plan["blocked_reasons"] assert "do_not_query_database_from_evidence_preview" in plan["safe_boundaries"] assert "do_not_generate_placeholder_evidence" in plan["safe_boundaries"] assert plan["bundle_contract"]["freshness_window_hours"] == 24 def test_opportunity_evidence_plan_route_is_preview_only(): from routes.market_intel_routes import market_intel_bp app = Flask(__name__) app.secret_key = "test-secret" app.register_blueprint(market_intel_bp) client = app.test_client() with client.session_transaction() as session: session["logged_in"] = True response = client.get("/api/market_intel/opportunity_evidence_plan") data = response.get_json() assert response.status_code == 200 assert data["mode"] == "opportunity_evidence_plan_preview" assert data["phase"] == "phase_48_manual_sample_acceptance_contract" assert data["evidence_bundle_created"] is False assert data["evidence_query_executed"] is False assert data["sample_evidence_generated"] is False assert data["alert_candidate_created"] is False assert data["database_write_executed"] is False assert data["external_network_executed"] is False def test_opportunity_alert_plan_preview_blocks_dispatch_and_llm_calls(): plan = MarketIntelService().build_opportunity_alert_plan() assert plan["mode"] == "opportunity_alert_plan_preview" assert plan["phase"] == "phase_48_manual_sample_acceptance_contract" assert plan["ready_for_alert_candidates"] is False assert plan["alert_candidate_created"] is False assert plan["alert_queue_created"] is False assert plan["review_queue_created"] is False assert plan["review_queue_contract_defined"] is True assert plan["review_queue_contract_written"] is False assert plan["review_queue_table_created"] is False assert plan["review_action_executed"] is False assert plan["approval_record_written"] is False assert plan["telegram_dispatched"] is False assert plan["ai_summary_generated"] is False assert plan["llm_call_executed"] is False assert plan["notification_sent"] is False assert plan["database_session_created"] is False assert plan["database_write_executed"] is False assert plan["database_commit_executed"] is False assert plan["external_network_executed"] is False assert plan["scheduler_attached"] is False assert plan["writes_executed"] is False assert plan["would_write_database"] is False assert plan["channel_count"] == 4 assert {item["key"] for item in plan["channels"]} == { "operator_review", "daily_digest", "telegram_candidate", "ai_briefing_candidate", } assert plan["review_state_count"] == 6 assert {item["key"] for item in plan["review_states"]} == { "draft", "needs_review", "approved_for_digest", "approved_for_telegram", "rejected", "deferred", } assert {item["key"] for item in plan["review_actions"]} == { "approve_digest", "approve_telegram", "reject", "defer", } assert plan["review_queue_contract"]["table_name"] == "market_alert_review_queue" assert "alert_candidate_id" in plan["review_queue_contract"]["required_fields"] assert "reviewer_identity" in plan["review_queue_contract"]["audit_fields"] assert "payment_or_shipping_data" in plan["review_queue_contract"]["forbidden_fields"] assert {item["key"] for item in plan["review_priority_lanes"]} == { "critical", "high", "medium", "watch", } assert {item["key"] for item in plan["review_queue_indexes"]} == { "idx_market_alert_review_queue_state_priority", "ux_market_alert_review_queue_dedupe", "idx_market_alert_review_queue_bundle", } assert plan["approval_policy"]["manual_approval_required_for_dispatch"] is True assert plan["approval_policy"]["reviewer_identity_required"] is True assert "evidence_bundle_ready" in plan["blocked_reasons"] assert "scoring_job_ready" in plan["blocked_reasons"] assert "operator_approval" in plan["blocked_reasons"] assert "do_not_dispatch_telegram_from_alert_preview" in plan["safe_boundaries"] assert "do_not_create_review_queue_table_from_preview" in plan["safe_boundaries"] assert "do_not_write_review_queue_contract_from_preview" in plan["safe_boundaries"] assert "do_not_call_llm_from_alert_preview" in plan["safe_boundaries"] assert "customer_personal_data" in plan["payload_contract"]["forbidden_fields"] def test_opportunity_alert_plan_route_is_preview_only(): from routes.market_intel_routes import market_intel_bp app = Flask(__name__) app.secret_key = "test-secret" app.register_blueprint(market_intel_bp) client = app.test_client() with client.session_transaction() as session: session["logged_in"] = True response = client.get("/api/market_intel/opportunity_alert_plan") data = response.get_json() assert response.status_code == 200 assert data["mode"] == "opportunity_alert_plan_preview" assert data["phase"] == "phase_48_manual_sample_acceptance_contract" assert data["alert_candidate_created"] is False assert data["alert_queue_created"] is False assert data["review_queue_created"] is False assert data["review_queue_contract_defined"] is True assert data["review_queue_contract_written"] is False assert data["review_queue_table_created"] is False assert data["review_action_executed"] is False assert data["approval_record_written"] is False assert data["telegram_dispatched"] is False assert data["llm_call_executed"] is False assert data["database_write_executed"] is False assert data["external_network_executed"] is False def test_mcp_deploy_preflight_blocks_without_required_env(): preflight = build_mcp_deploy_preflight_plan(env={}) assert preflight["mode"] == "mcp_external_deploy_preflight_preview" assert preflight["compose_file_present"] is True assert preflight["ready_to_start_stack"] is False assert preflight["checks"]["compose_file_present"] is True assert preflight["checks"]["all_expected_services_declared"] is True assert preflight["checks"]["all_expected_containers_declared"] is True assert preflight["checks"]["all_public_mcp_ports_localhost_only"] is True assert preflight["checks"]["required_env_vars_present"] is False assert preflight["checks"]["router_flag_still_off_before_health"] is True assert "required_env_vars_present" in preflight["blocked_reasons"] assert preflight["operator_command_preview"] == "docker compose -f docker-compose.mcp.yml up -d" assert preflight["deployment_actions_executed"] is False assert preflight["docker_command_executed"] is False assert preflight["ssh_command_executed"] is False assert preflight["database_session_created"] is False assert preflight["database_write_executed"] is False assert preflight["database_commit_executed"] is False assert preflight["external_network_executed"] is False assert preflight["scheduler_attached"] is False assert preflight["would_write_database"] is False def test_mcp_compose_is_mounted_read_only_for_preflight(): compose = Path("docker-compose.yml").read_text(encoding="utf-8") assert "./docker-compose.mcp.yml:/app/docker-compose.mcp.yml:ro" in compose def test_mcp_deploy_preflight_ready_when_env_contract_is_present(): preflight = build_mcp_deploy_preflight_plan( env={ "MCP_POSTGRES_PASSWORD": "secret", "TAVILY_API_KEY": "tavily", "EXA_API_KEY": "exa", "MCP_ROUTER_ENABLED": "false", } ) assert preflight["ready_to_start_stack"] is True assert preflight["blocked_reasons"] == [] assert preflight["checks"]["required_env_vars_present"] is True assert preflight["checks"]["router_flag_still_off_before_health"] is True assert all(item["localhost_only"] for item in preflight["port_statuses"]) def test_mcp_deploy_preflight_route_is_preview_only(): from routes.market_intel_routes import market_intel_bp app = Flask(__name__) app.secret_key = "test-secret" app.register_blueprint(market_intel_bp) client = app.test_client() with client.session_transaction() as session: session["logged_in"] = True response = client.get("/api/market_intel/mcp_deploy_preflight") data = response.get_json() assert response.status_code == 200 assert data["mode"] == "mcp_external_deploy_preflight_preview" assert data["phase"] == "phase_48_manual_sample_acceptance_contract" assert data["deployment_actions_executed"] is False assert data["docker_command_executed"] is False assert data["ssh_command_executed"] is False assert data["database_write_executed"] is False assert data["external_network_executed"] is False assert data["scheduler_attached"] is False def test_mcp_readiness_default_is_planned_only(monkeypatch): monkeypatch.delenv("MCP_ROUTER_ENABLED", raising=False) readiness = MarketIntelService().build_mcp_readiness() assert readiness["mode"] == "mcp_readiness_planned" assert readiness["phase"] == "phase_48_manual_sample_acceptance_contract" assert readiness["execute_requested"] is False assert readiness["router_enabled"] is False assert readiness["external_mcp_complete"] is False assert readiness["internal_mcp_complete"] is False assert readiness["market_intel_mcp_integrated"] is True assert readiness["mcp_tool_contract"]["contract_ready"] is True assert readiness["database_session_created"] is False assert readiness["database_write_executed"] is False assert readiness["database_commit_executed"] is False assert readiness["external_network_executed"] is False assert readiness["scheduler_attached"] is False assert readiness["writes_executed"] is False assert readiness["would_write_database"] is False assert readiness["telemetry"]["read_only_query_executed"] is False assert readiness["telemetry"]["database_session_created"] is False assert readiness["telemetry"]["database_write_executed"] is False assert readiness["readiness_checks"]["base_callers_registered"] is True assert readiness["readiness_checks"]["market_intel_caller_registered"] is True assert readiness["readiness_checks"]["market_intel_tools_registered"] is True assert readiness["readiness_checks"]["market_intel_tool_contract_ready"] is True assert len(readiness["server_statuses"]) == 4 assert all(item["health_checked"] is False for item in readiness["server_statuses"]) assert all(item["healthy"] is False for item in readiness["server_statuses"]) assert "execute_false_planned_only" in readiness["blocked_reasons"] assert "market_intel_caller_registered" not in readiness["blocked_reasons"] assert "market_intel_tools_registered" not in readiness["blocked_reasons"] def test_mcp_readiness_sqlite_read_only_counts_telemetry(monkeypatch): monkeypatch.delenv("MCP_ROUTER_ENABLED", raising=False) monkeypatch.setattr( "services.market_intel.mcp_readiness._health_check_servers", lambda base_hosts, timeout_sec: [ { "server": "postgres", "base_url": "http://127.0.0.1:3001", "configured": True, "health_checked": True, "healthy": False, "status": "error", "error_message": "test skipped network", }, { "server": "omnisearch", "base_url": "http://127.0.0.1:3003", "configured": True, "health_checked": True, "healthy": False, "status": "error", "error_message": "test skipped network", }, { "server": "firecrawl", "base_url": "http://127.0.0.1:3002", "configured": True, "health_checked": True, "healthy": False, "status": "error", "error_message": "test skipped network", }, { "server": "filesystem", "base_url": "http://127.0.0.1:3004", "configured": True, "health_checked": True, "healthy": False, "status": "error", "error_message": "test skipped network", }, ], ) engine = create_engine("sqlite:///:memory:") with engine.begin() as conn: conn.execute( text( """ CREATE TABLE mcp_calls ( id INTEGER PRIMARY KEY, server TEXT, called_at TEXT ) """ ) ) conn.execute( text( """ INSERT INTO mcp_calls (server, called_at) VALUES ('omnisearch', '2026-05-18 10:00:00'), ('omnisearch', '2026-05-18 10:05:00'), ('postgres', '2026-05-18 10:10:00') """ ) ) readiness = build_mcp_readiness_plan( execute_requested=True, timeout_sec=1, engine=engine, database_type="sqlite", ) assert readiness["mode"] == "mcp_readiness_read_only" assert readiness["execute_requested"] is True assert readiness["router_enabled"] is False assert readiness["external_mcp_complete"] is False assert readiness["internal_mcp_complete"] is True assert readiness["market_intel_mcp_integrated"] is True assert readiness["mcp_tool_contract"]["contract_ready"] is True assert readiness["telemetry"]["mode"] == "mcp_telemetry_read_only" assert readiness["telemetry"]["table_exists"] is True assert readiness["telemetry"]["total_calls"] == 3 assert readiness["telemetry"]["recent_24h_calls"] == 3 assert readiness["telemetry"]["read_only_query_executed"] is True assert readiness["telemetry"]["database_session_created"] is False assert readiness["telemetry"]["database_write_executed"] is False assert readiness["telemetry"]["database_commit_executed"] is False assert readiness["database_session_created"] is False assert readiness["database_write_executed"] is False assert readiness["database_commit_executed"] is False assert readiness["scheduler_attached"] is False assert readiness["writes_executed"] is False assert readiness["would_write_database"] is False assert readiness["server_statuses"][0]["health_checked"] is True assert {item["server"]: item["calls"] for item in readiness["telemetry"]["server_counts"]} == { "omnisearch": 2, "postgres": 1, } assert "mcp_router_enabled" in readiness["blocked_reasons"] assert "market_intel_caller_registered" not in readiness["blocked_reasons"] def test_market_intel_schema_smoke_checks_platform_columns(): smoke = MarketIntelService().build_schema_smoke()["schema_smoke"] assert smoke["passed"] is True assert smoke["missing_tables"] == [] assert smoke["missing_market_platform_columns"] == [] assert "crawl_policy_json" in smoke["market_platform_required_columns"] def test_platform_seed_plan_is_read_only_and_adapter_derived(): plan = MarketIntelService().build_platform_seed_plan() seed_codes = {seed["code"] for seed in plan["seeds"]} assert {"momo", "pchome", "coupang", "shopee"} <= seed_codes assert plan["would_write_database"] is False assert plan["required_gates"]["schema_smoke_required"] is True assert plan["required_gates"]["migration_required"] is True assert plan["required_gates"]["manual_operator_approval_required"] is True for seed in plan["seeds"]: policy = seed["crawl_policy_json"] assert seed["enabled"] is False assert seed["source_count"] == len(seed["sources"]) assert policy["allow_login"] is False assert policy["allow_database_write"] is False assert policy["allow_scheduler_attach"] is False def test_platform_seed_plan_unknown_adapter_returns_diagnostic_error(): plan = MarketIntelService().build_platform_seed_plan("unknown") assert plan["found"] is False assert plan["seed_count"] == 0 assert plan["error"] == "未知平台 adapter" def test_platform_seed_write_guard_blocks_default_environment(): guard = MarketIntelService().build_platform_seed_write_guard() assert guard["ready_to_write"] is False assert guard["would_write_database"] is False assert guard["database_write_allowed"] is False assert guard["seed_count"] == 4 assert "market_intel_enabled" in guard["blocked_reasons"] assert "market_intel_write_enabled" in guard["blocked_reasons"] assert "database_write_allowed" in guard["blocked_reasons"] assert "migration_confirmed" in guard["blocked_reasons"] assert "manual_operator_approval" in guard["blocked_reasons"] assert guard["schema_smoke"]["passed"] is True assert "schema_smoke_confirmed" not in guard["blocked_reasons"] def test_platform_seed_writer_plan_is_dry_run_only(): writer_plan = MarketIntelService().build_platform_seed_writer_plan() assert writer_plan["mode"] == "dry_run" assert writer_plan["ready_to_write"] is False assert writer_plan["writes_executed"] is False assert writer_plan["would_write_database"] is False assert writer_plan["operation_count"] == 4 assert writer_plan["schema_smoke"]["passed"] is True first_operation = writer_plan["operations"][0] assert first_operation["operation"] == "upsert" assert first_operation["table"] == "market_platforms" assert first_operation["write_status"] == "blocked_dry_run_only" assert first_operation["values"]["enabled"] is False assert "ON CONFLICT(code)" in first_operation["sql_shape"] def test_schema_db_probe_planned_does_not_connect_or_write(): probe = MarketIntelService().build_schema_db_probe() assert probe["mode"] == "schema_db_probe_planned" assert probe["execute_requested"] is False assert probe["read_only_query_executed"] is False assert probe["database_connection_opened"] is False assert probe["database_session_created"] is False assert probe["explicit_transaction_opened"] is False assert probe["database_write_executed"] is False assert probe["database_commit_executed"] is False assert probe["migration_executed"] is False assert probe["missing_tables"] == list(MARKET_INTEL_TABLES) assert "execute_false_planned_only" in probe["blocked_reasons"] def test_schema_db_probe_sqlite_read_only_reports_existing_and_missing_tables(): engine = create_engine("sqlite:///:memory:") with engine.begin() as conn: conn.execute(text("CREATE TABLE market_platforms (id INTEGER PRIMARY KEY)")) probe = build_schema_db_probe_plan( ["market_platforms", "market_campaigns"], execute_requested=True, database_type="sqlite", engine=engine, ) assert probe["mode"] == "schema_db_probe_read_only" assert probe["execute_requested"] is True assert probe["read_only_query_executed"] is True assert probe["database_connection_opened"] is True assert probe["database_session_created"] is False assert probe["explicit_transaction_opened"] is False assert probe["database_write_executed"] is False assert probe["database_commit_executed"] is False assert probe["existing_tables"] == ["market_platforms"] assert probe["missing_tables"] == ["market_campaigns"] assert probe["schema_tables_exist"] is False assert "market_tables_missing" in probe["blocked_reasons"] def test_platform_seed_db_diff_planned_does_not_connect_or_write(): diff = MarketIntelService().build_platform_seed_db_diff() assert diff["mode"] == "platform_seed_db_diff_planned" assert diff["execute_requested"] is False assert diff["read_only_query_executed"] is False assert diff["database_connection_opened"] is False assert diff["database_session_created"] is False assert diff["explicit_transaction_opened"] is False assert diff["database_write_executed"] is False assert diff["database_commit_executed"] is False assert diff["seed_write_executed"] is False assert diff["expected_seed_count"] == 4 assert set(diff["missing_codes"]) == {"momo", "pchome", "coupang", "shopee"} assert "execute_false_planned_only" in diff["blocked_reasons"] assert "seed_write_still_blocked" in diff["blocked_reasons"] def test_platform_seed_db_diff_sqlite_read_only_reports_missing_and_matching_seed_rows(): seed_plan = { "seeds": [ { "code": "momo", "name": "MOMO", "base_url": "https://momo.example", "enabled": False, "crawl_policy_json": { "allow_login": False, "allow_database_write": False, "seed_source_keys": ["momo_edm"], }, }, { "code": "pchome", "name": "PChome", "base_url": "https://pchome.example", "enabled": False, "crawl_policy_json": { "allow_login": False, "allow_database_write": False, "seed_source_keys": ["pchome_region"], }, }, ] } engine = create_engine("sqlite:///:memory:") with engine.begin() as conn: conn.execute( text( """ CREATE TABLE market_platforms ( code TEXT PRIMARY KEY, name TEXT, base_url TEXT, enabled BOOLEAN, crawl_policy_json TEXT ) """ ) ) conn.execute( text( """ INSERT INTO market_platforms (code, name, base_url, enabled, crawl_policy_json) VALUES (:code, :name, :base_url, :enabled, :crawl_policy_json) """ ), { "code": "momo", "name": "MOMO", "base_url": "https://momo.example", "enabled": False, "crawl_policy_json": json.dumps( seed_plan["seeds"][0]["crawl_policy_json"], ensure_ascii=False, sort_keys=True, ), }, ) diff = build_platform_seed_db_diff_plan( seed_plan, execute_requested=True, database_type="sqlite", engine=engine, ) assert diff["mode"] == "platform_seed_db_diff_read_only" assert diff["execute_requested"] is True assert diff["read_only_query_executed"] is True assert diff["database_connection_opened"] is True assert diff["database_session_created"] is False assert diff["explicit_transaction_opened"] is False assert diff["database_write_executed"] is False assert diff["database_commit_executed"] is False assert diff["seed_write_executed"] is False assert diff["expected_seed_count"] == 2 assert diff["existing_seed_count"] == 1 assert diff["existing_codes"] == ["momo"] assert diff["missing_codes"] == ["pchome"] assert diff["changed_codes"] == [] assert diff["matching_codes"] == ["momo"] assert diff["seed_rows_ready"] is False assert {item["code"]: item["diff_status"] for item in diff["seed_diffs"]} == { "momo": "matches_expected", "pchome": "missing", } assert "seed_rows_missing" in diff["blocked_reasons"] assert "seed_write_still_blocked" in diff["blocked_reasons"] def test_deployment_readiness_reports_app_only_release_gate(): readiness = MarketIntelService().build_deployment_readiness() step_keys = {step["key"] for step in readiness["required_manual_steps"]} fallback_keys = {item["key"] for item in readiness["fallback_plan"]} boundary_keys = {item["key"] for item in readiness["safe_deploy_boundaries"]} assert readiness["mode"] == "app_only_release_gate" assert readiness["production_deployed"] is False assert readiness["git_committed"] is False assert readiness["git_pushed"] is False assert readiness["ready_for_production_deploy"] is True assert readiness["deployment_actions_executed"] is False assert readiness["execution_boundary"]["api_executes_scp"] is False assert readiness["execution_boundary"]["api_recreates_container"] is False assert readiness["execution_boundary"]["api_runs_migration"] is False assert readiness["execution_boundary"]["api_writes_database"] is False assert readiness["requires_backup_before_major_update"] is True assert readiness["backup_command"] == "python backup_system.py" assert readiness["checks"]["schema_smoke_passed"] is True assert readiness["checks"]["schema_db_probe_planned_safe"] is True assert readiness["checks"]["platform_seed_db_diff_planned_safe"] is True assert readiness["checks"]["legacy_source_bridge_planned_safe"] is True assert readiness["checks"]["mcp_readiness_planned_safe"] is True assert readiness["checks"]["mcp_tool_contract_ready"] is True assert readiness["checks"]["mcp_deploy_preflight_preview_safe"] is True assert readiness["checks"]["mcp_activation_runbook_preview_safe"] is True assert readiness["checks"]["mcp_fetch_gate_preview_safe"] is True assert readiness["checks"]["scheduler_plan_preview_safe"] is True assert readiness["checks"]["manual_sample_plan_preview_safe"] is True assert readiness["checks"]["manual_sample_acceptance_preview_safe"] is True assert readiness["checks"]["match_review_plan_preview_safe"] is True assert readiness["checks"]["opportunity_plan_preview_safe"] is True assert readiness["checks"]["opportunity_scoring_plan_preview_safe"] is True assert readiness["checks"]["opportunity_evidence_plan_preview_safe"] is True assert readiness["checks"]["opportunity_alert_plan_preview_safe"] is True assert readiness["checks"]["migration_apply_drill_preview_safe"] is True assert readiness["checks"]["migration_catalog_review_preview_safe"] is True assert readiness["checks"]["migration_live_smoke_preview_safe"] is True assert readiness["checks"]["live_db_inventory_preview_safe"] is True assert readiness["checks"]["writer_plan_dry_run_only"] is True assert readiness["writer_plan_summary"]["writes_executed"] is False assert "readiness_checks_not_all_passed" not in readiness["blocked_reasons"] assert "production_deploy_not_executed_by_api" in readiness["blocked_reasons"] assert "backup_must_be_verified_by_operator" in readiness["blocked_reasons"] assert "run_backup_system" in step_keys assert "verify_health_endpoint" in step_keys assert "feature_flag_kill_switch" in fallback_keys assert "database_write_blocked" in fallback_keys assert "no_remove_orphans" in boundary_keys assert "no_momo_db_lifecycle_change" in boundary_keys assert "/health" in readiness["production_smoke_targets"] assert "/api/market_intel/deployment_readiness" in readiness["production_smoke_targets"] assert "/api/market_intel/platform_seed_db_diff" in readiness["production_smoke_targets"] assert "/api/market_intel/legacy_source_bridge" in readiness["production_smoke_targets"] assert "/api/market_intel/mcp_readiness" in readiness["production_smoke_targets"] assert "/api/market_intel/mcp_tool_contract" in readiness["production_smoke_targets"] assert "/api/market_intel/mcp_deploy_preflight" in readiness["production_smoke_targets"] assert "/api/market_intel/mcp_activation_runbook" in readiness["production_smoke_targets"] assert "/api/market_intel/mcp_fetch_gate" in readiness["production_smoke_targets"] assert "/api/market_intel/scheduler_plan" in readiness["production_smoke_targets"] assert "/api/market_intel/manual_sample_plan" in readiness["production_smoke_targets"] assert "/api/market_intel/manual_sample_acceptance" in readiness["production_smoke_targets"] assert "/api/market_intel/match_review_plan" in readiness["production_smoke_targets"] assert "/api/market_intel/opportunity_plan" in readiness["production_smoke_targets"] assert "/api/market_intel/opportunity_scoring_plan" in readiness["production_smoke_targets"] assert "/api/market_intel/opportunity_evidence_plan" in readiness["production_smoke_targets"] assert "/api/market_intel/opportunity_alert_plan" in readiness["production_smoke_targets"] assert "/api/market_intel/migration_apply_drill" in readiness["production_smoke_targets"] assert "/api/market_intel/migration_catalog_review" in readiness["production_smoke_targets"] assert "/api/market_intel/migration_live_smoke" in readiness["production_smoke_targets"] assert "/api/market_intel/live_db_inventory" in readiness["production_smoke_targets"] assert readiness["write_approval_runbook"]["ready_for_real_write"] is False assert readiness["write_approval_runbook"]["writes_executed"] is False assert readiness["migration_blueprint"]["migration_executed"] is False assert readiness["migration_blueprint"]["file_created"] is True assert readiness["migration_blueprint"]["file_matches_blueprint"] is True assert readiness["migration_apply_drill"]["mode"] == "migration_apply_drill_preview" assert readiness["migration_apply_drill"]["migration_executed"] is False assert readiness["migration_apply_drill"]["rollback_executed"] is False assert readiness["migration_apply_drill"]["database_write_executed"] is False assert readiness["migration_apply_drill"]["api_executes_migration"] is False assert readiness["migration_catalog_review"]["mode"] == "migration_catalog_review_preview" assert readiness["migration_catalog_review"]["catalog_state"] == "planned_no_probe" assert readiness["migration_catalog_review"]["migration_executed"] is False assert readiness["migration_catalog_review"]["database_write_executed"] is False assert readiness["migration_catalog_review"]["api_executes_migration"] is False assert readiness["migration_live_smoke"]["mode"] == "migration_live_smoke_preview" assert readiness["migration_live_smoke"]["smoke_result"] == "planned_no_execution" assert readiness["migration_live_smoke"]["migration_executed"] is False assert readiness["migration_live_smoke"]["database_write_executed"] is False assert readiness["migration_live_smoke"]["api_executes_migration"] is False assert readiness["live_db_inventory"]["mode"] == "live_db_inventory_planned" assert readiness["live_db_inventory"]["read_only_query_executed"] is False assert readiness["live_db_inventory"]["database_write_executed"] is False assert readiness["live_db_inventory"]["migration_executed"] is False assert readiness["schema_db_probe"]["read_only_query_executed"] is False assert readiness["platform_seed_db_diff"]["read_only_query_executed"] is False assert readiness["legacy_source_bridge"]["read_only_query_executed"] is False assert readiness["mcp_readiness"]["mode"] == "mcp_readiness_planned" assert readiness["mcp_readiness"]["telemetry"]["read_only_query_executed"] is False assert readiness["mcp_tool_contract"]["contract_ready"] is True assert readiness["mcp_tool_contract"]["database_write_executed"] is False assert readiness["mcp_deploy_preflight"]["deployment_actions_executed"] is False assert readiness["mcp_deploy_preflight"]["docker_command_executed"] is False assert readiness["mcp_activation_runbook"]["deployment_actions_executed"] is False assert readiness["mcp_activation_runbook"]["docker_command_executed"] is False assert readiness["mcp_fetch_gate"]["network_request_allowed"] is False assert readiness["mcp_fetch_gate"]["external_network_executed"] is False assert readiness["manual_sample_plan"]["mode"] == "manual_sample_fetch_plan_preview" assert readiness["manual_sample_plan"]["sample_fetch_executed"] is False assert readiness["manual_sample_plan"]["external_network_executed"] is False assert readiness["manual_sample_plan"]["database_write_executed"] is False assert readiness["manual_sample_plan"]["scheduler_attached"] is False assert readiness["manual_sample_acceptance"]["mode"] == "manual_sample_acceptance_preview" assert readiness["manual_sample_acceptance"]["sample_result_loaded"] is False assert readiness["manual_sample_acceptance"]["candidate_import_allowed"] is False assert readiness["manual_sample_acceptance"]["external_network_executed"] is False assert readiness["manual_sample_acceptance"]["database_write_executed"] is False assert readiness["manual_sample_acceptance"]["scheduler_attached"] is False assert readiness["scheduler_plan"]["scheduler_registration_executed"] is False assert readiness["scheduler_plan"]["crawler_job_started"] is False assert readiness["scheduler_plan"]["database_write_executed"] is False assert readiness["match_review_plan"]["review_queue_created"] is False assert readiness["match_review_plan"]["auto_confirm_executed"] is False assert readiness["match_review_plan"]["database_write_executed"] is False assert readiness["opportunity_plan"]["opportunity_queue_created"] is False assert readiness["opportunity_plan"]["threat_alert_dispatched"] is False assert readiness["opportunity_plan"]["database_write_executed"] is False assert readiness["opportunity_scoring_plan"]["scoring_job_created"] is False assert readiness["opportunity_scoring_plan"]["score_calculation_executed"] is False assert readiness["opportunity_scoring_plan"]["database_write_executed"] is False assert readiness["opportunity_evidence_plan"]["evidence_bundle_created"] is False assert readiness["opportunity_evidence_plan"]["evidence_query_executed"] is False assert readiness["opportunity_evidence_plan"]["database_write_executed"] is False assert readiness["opportunity_alert_plan"]["alert_candidate_created"] is False assert readiness["opportunity_alert_plan"]["review_queue_contract_written"] is False assert readiness["opportunity_alert_plan"]["review_queue_table_created"] is False assert readiness["opportunity_alert_plan"]["review_action_executed"] is False assert readiness["opportunity_alert_plan"]["approval_record_written"] is False assert readiness["opportunity_alert_plan"]["telegram_dispatched"] is False assert readiness["opportunity_alert_plan"]["llm_call_executed"] is False assert readiness["opportunity_alert_plan"]["database_write_executed"] is False def test_market_intel_service_keeps_readiness_modularized(): service_path = Path("services/market_intel/service.py") readiness_path = Path("services/market_intel/deployment_readiness.py") assert service_path.exists() assert readiness_path.exists() assert len(service_path.read_text().splitlines()) < 800 assert "build_deployment_readiness_preview" in readiness_path.read_text() def test_write_approval_runbook_is_read_only_and_blocks_real_write(): runbook = MarketIntelService().build_write_approval_runbook() gate_keys = {gate["key"] for gate in runbook["approval_gates"]} assert runbook["mode"] == "approval_runbook_read_only" assert runbook["ready_for_real_write"] is False assert runbook["writes_executed"] is False assert runbook["would_write_database"] is False assert runbook["database_session_created"] is False assert runbook["database_commit_executed"] is False assert runbook["external_network_executed"] is False assert runbook["scheduler_attached"] is False assert runbook["approval_required"] is True assert runbook["approval_token_present"] is False assert runbook["seed_count"] == 4 assert runbook["writer_operation_count"] == 4 assert runbook["schema_smoke"]["passed"] is True assert "schema_smoke_passed" in gate_keys assert "backup_completed" in gate_keys assert "migration_file_reviewed" in gate_keys assert "manual_operator_approval" in gate_keys assert "database_write_allowed" in runbook["blocked_reasons"] assert "manual_operator_approval" in runbook["blocked_reasons"] assert "no_momo_db_container_lifecycle_change" in runbook["hard_safety_boundaries"] assert "no_remove_orphans" in runbook["hard_safety_boundaries"] def test_migration_blueprint_is_additive_preview_only(): blueprint = MarketIntelService().build_migration_blueprint() forward_sql_lower = blueprint["forward_sql"].lower() migration_file = Path(blueprint["suggested_filename"]) assert blueprint["mode"] == "migration_file_draft_read_only" assert blueprint["suggested_filename"] == "migrations/032_market_intel_core_schema.sql" assert blueprint["file_created"] is True assert blueprint["file_matches_blueprint"] is True assert blueprint["file_status"] == "local_draft_matches_blueprint" assert blueprint["migration_executed"] is False assert blueprint["database_session_created"] is False assert blueprint["database_commit_executed"] is False assert blueprint["external_network_executed"] is False assert blueprint["scheduler_attached"] is False assert blueprint["table_count"] == 8 assert blueprint["forward_has_destructive_sql"] is False assert blueprint["safety_checks"]["forward_sql_additive_only"] is True assert blueprint["safety_checks"]["writes_seed_rows_only_with_cli_apply_flag"] is True assert "migration_not_executed" in blueprint["blocked_reasons"] assert "migration_file_not_created" not in blueprint["blocked_reasons"] assert "seed_writer_real_write_requires_cli_apply_flag" in blueprint["blocked_reasons"] assert migration_file.exists() assert migration_file.read_text(encoding="utf-8").strip() == blueprint["forward_sql"] assert "CREATE TABLE IF NOT EXISTS market_platforms".lower() in forward_sql_lower assert "CREATE TABLE IF NOT EXISTS market_crawler_runs".lower() in forward_sql_lower assert "CREATE TABLE IF NOT EXISTS market_alert_review_queue".lower() in forward_sql_lower assert "ux_market_alert_review_queue_dedupe".lower() in forward_sql_lower assert "drop table" not in forward_sql_lower assert "truncate " not in forward_sql_lower assert "delete from" not in forward_sql_lower assert "MARKET_INTEL_CRAWLER_ENABLED=false" in blueprint["command_plan"]["seed_writer_command"]["command"] assert blueprint["command_plan"]["seed_writer_command"]["script_created"] is True assert blueprint["command_plan"]["seed_writer_command"]["script_path"] == "scripts/market_intel_seed_writer.py" assert blueprint["rollback_requires_manual_approval"] is True assert "DROP TABLE IF EXISTS market_alert_review_queue" in blueprint["rollback_sql"] def test_migration_apply_drill_planned_is_safe_and_manual_only(): drill = MarketIntelService().build_migration_apply_drill() assert drill["mode"] == "migration_apply_drill_preview" assert drill["phase"] == "phase_48_manual_sample_acceptance_contract" assert drill["execute_requested"] is False assert drill["schema_state"] == "planned_no_db_probe" assert drill["drill_ready_for_operator_review"] is True assert drill["ready_for_manual_apply_review"] is False assert drill["ready_to_apply_migration"] is False assert drill["read_only_query_executed"] is False assert drill["database_connection_opened"] is False assert drill["database_session_created"] is False assert drill["database_write_executed"] is False assert drill["database_commit_executed"] is False assert drill["migration_executed"] is False assert drill["rollback_executed"] is False assert drill["api_executes_migration"] is False assert drill["api_executes_rollback"] is False assert drill["checks"]["migration_file_matches_blueprint"] is True assert drill["checks"]["forward_sql_additive_only"] is True assert "read_only_db_probe_not_executed" in drill["blocked_reasons"] assert "api_never_runs_migration" in drill["blocked_reasons"] assert drill["rollback_drill"]["rollback_executed"] is False assert drill["rollback_drill"]["requires_manual_approval"] is True assert "market_alert_review_queue" in drill["rollback_drill"]["rollback_table_order"] assert len(drill["pre_apply_checklist"]) >= 6 assert len(drill["post_apply_verification"]) >= 4 assert len(drill["risk_register"]) >= 4 def test_migration_apply_drill_sqlite_read_only_reports_already_applied(): service = MarketIntelService() seed_rows = service.build_platform_seed_plan()["seeds"] engine = create_engine("sqlite:///:memory:") with engine.begin() as conn: conn.execute( text( """ CREATE TABLE market_platforms ( code TEXT PRIMARY KEY, name TEXT, base_url TEXT, enabled BOOLEAN, crawl_policy_json TEXT ) """ ) ) for table_name in MARKET_INTEL_TABLES: if table_name == "market_platforms": continue conn.execute(text(f"CREATE TABLE {table_name} (id INTEGER PRIMARY KEY)")) for seed in seed_rows: conn.execute( text( """ INSERT INTO market_platforms (code, name, base_url, enabled, crawl_policy_json) VALUES (:code, :name, :base_url, :enabled, :crawl_policy_json) """ ), { "code": seed["code"], "name": seed["name"], "base_url": seed["base_url"], "enabled": seed["enabled"], "crawl_policy_json": json.dumps( seed["crawl_policy_json"], ensure_ascii=False, sort_keys=True, ), }, ) drill = service.build_migration_apply_drill( execute_requested=True, schema_engine=engine, seed_diff_engine=engine, database_type="sqlite", ) assert drill["mode"] == "migration_apply_drill_preview" assert drill["execute_requested"] is True assert drill["read_only_query_executed"] is True assert drill["database_connection_opened"] is True assert drill["schema_state"] == "already_applied" assert drill["schema_db_probe_summary"]["schema_tables_exist"] is True assert drill["schema_db_probe_summary"]["missing_tables"] == [] assert drill["platform_seed_db_diff_summary"]["seed_rows_ready"] is True assert set(drill["platform_seed_db_diff_summary"]["matching_codes"]) == { "momo", "pchome", "coupang", "shopee", } assert drill["database_session_created"] is False assert drill["database_write_executed"] is False assert drill["database_commit_executed"] is False assert drill["migration_executed"] is False assert "market_schema_already_present_apply_not_required" in drill["blocked_reasons"] def test_migration_apply_drill_route_is_preview_only(): from routes.market_intel_routes import market_intel_bp app = Flask(__name__) app.secret_key = "test-secret" app.register_blueprint(market_intel_bp) client = app.test_client() with client.session_transaction() as session: session["logged_in"] = True response = client.get("/api/market_intel/migration_apply_drill?execute=false") data = response.get_json() assert response.status_code == 200 assert data["mode"] == "migration_apply_drill_preview" assert data["phase"] == "phase_48_manual_sample_acceptance_contract" assert data["execute_requested"] is False assert data["migration_executed"] is False assert data["rollback_executed"] is False assert data["database_write_executed"] is False assert data["api_executes_migration"] is False def test_migration_catalog_review_planned_is_safe_and_diagnostic(): review = MarketIntelService().build_migration_catalog_review() assert review["mode"] == "migration_catalog_review_preview" assert review["phase"] == "phase_48_manual_sample_acceptance_contract" assert review["execute_requested"] is False assert review["catalog_state"] == "planned_no_probe" assert review["seed_state"] == "planned_no_probe" assert review["risk_level"] == "info" assert review["apply_path"] == "run_execute_true_read_only_probe_first" assert review["review_ready"] is True assert review["ready_for_manual_migration_review"] is False assert review["ready_to_apply_migration"] is False assert review["read_only_query_executed"] is False assert review["database_connection_opened"] is False assert review["database_session_created"] is False assert review["database_write_executed"] is False assert review["database_commit_executed"] is False assert review["migration_executed"] is False assert review["rollback_executed"] is False assert review["api_executes_migration"] is False assert review["table_catalog"]["expected_count"] == 8 assert review["table_catalog"]["existing_count"] == 0 assert review["table_catalog"]["missing_count"] == 8 assert "execute_false_planned_only" in review["blocked_reasons"] assert "migration_not_executed_by_catalog_review" in review["blocked_reasons"] assert {item["key"] for item in review["findings"]} >= { "catalog_probe_not_executed", "seed_diff_not_executed", } def test_migration_catalog_review_sqlite_read_only_detects_partial_schema(): service = MarketIntelService() seed_rows = service.build_platform_seed_plan()["seeds"] engine = create_engine("sqlite:///:memory:") with engine.begin() as conn: conn.execute( text( """ CREATE TABLE market_platforms ( code TEXT PRIMARY KEY, name TEXT, base_url TEXT, enabled BOOLEAN, crawl_policy_json TEXT ) """ ) ) for seed in seed_rows: conn.execute( text( """ INSERT INTO market_platforms (code, name, base_url, enabled, crawl_policy_json) VALUES (:code, :name, :base_url, :enabled, :crawl_policy_json) """ ), { "code": seed["code"], "name": seed["name"], "base_url": seed["base_url"], "enabled": seed["enabled"], "crawl_policy_json": json.dumps( seed["crawl_policy_json"], ensure_ascii=False, sort_keys=True, ), }, ) review = service.build_migration_catalog_review( execute_requested=True, schema_engine=engine, seed_diff_engine=engine, database_type="sqlite", ) assert review["mode"] == "migration_catalog_review_preview" assert review["execute_requested"] is True assert review["read_only_query_executed"] is True assert review["database_connection_opened"] is True assert review["catalog_state"] == "partial_schema" assert review["seed_state"] == "ready" assert review["risk_level"] == "high" assert review["apply_path"] == "blocked_until_partial_schema_reviewed" assert review["ready_for_manual_migration_review"] is False assert review["table_catalog"]["expected_count"] == 8 assert review["table_catalog"]["existing_count"] == 1 assert review["table_catalog"]["missing_count"] == 7 assert review["seed_catalog"]["seed_rows_ready"] is True assert "partial_schema_requires_manual_reconciliation" in review["blocked_reasons"] assert review["database_session_created"] is False assert review["database_write_executed"] is False assert review["database_commit_executed"] is False assert review["migration_executed"] is False assert {item["key"] for item in review["findings"]} >= { "partial_market_schema_detected", "platform_seed_ready", } def test_migration_catalog_review_route_is_preview_only(): from routes.market_intel_routes import market_intel_bp app = Flask(__name__) app.secret_key = "test-secret" app.register_blueprint(market_intel_bp) client = app.test_client() with client.session_transaction() as session: session["logged_in"] = True response = client.get("/api/market_intel/migration_catalog_review?execute=false") data = response.get_json() assert response.status_code == 200 assert data["mode"] == "migration_catalog_review_preview" assert data["phase"] == "phase_48_manual_sample_acceptance_contract" assert data["execute_requested"] is False assert data["catalog_state"] == "planned_no_probe" assert data["migration_executed"] is False assert data["rollback_executed"] is False assert data["database_write_executed"] is False assert data["api_executes_migration"] is False def test_migration_live_smoke_planned_is_preview_only(): smoke = MarketIntelService().build_migration_live_smoke() assert smoke["mode"] == "migration_live_smoke_preview" assert smoke["phase"] == "phase_48_manual_sample_acceptance_contract" assert smoke["execute_requested"] is False assert smoke["smoke_result"] == "planned_no_execution" assert smoke["live_smoke_passed"] is False assert smoke["catalog_state"] == "planned_no_probe" assert smoke["seed_state"] == "planned_no_probe" assert smoke["read_only_probe_completed"] is False assert smoke["database_connection_opened"] is False assert smoke["database_session_created"] is False assert smoke["database_write_executed"] is False assert smoke["database_commit_executed"] is False assert smoke["migration_executed"] is False assert smoke["rollback_executed"] is False assert smoke["api_executes_migration"] is False assert "execute_false_planned_only" in smoke["blocked_reasons"] assert "migration_not_executed_by_live_smoke" in smoke["blocked_reasons"] def test_migration_live_smoke_sqlite_not_applied_tolerates_seed_table_missing(): service = MarketIntelService() engine = create_engine("sqlite:///:memory:") smoke = service.build_migration_live_smoke( execute_requested=True, schema_engine=engine, seed_diff_engine=engine, database_type="sqlite", ) assert smoke["mode"] == "migration_live_smoke_preview" assert smoke["execute_requested"] is True assert smoke["read_only_query_executed"] is True assert smoke["database_connection_opened"] is True assert smoke["catalog_state"] == "not_applied" assert smoke["seed_state"] == "probe_error" assert smoke["seed_probe_error_tolerated"] is True assert smoke["smoke_result"] == "passed_not_applied_seed_table_missing" assert smoke["live_smoke_passed"] is True assert smoke["read_only_probe_completed"] is True assert smoke["ready_for_manual_migration_review"] is True assert smoke["database_session_created"] is False assert smoke["database_write_executed"] is False assert smoke["database_commit_executed"] is False assert smoke["migration_executed"] is False assert smoke["catalog_review_summary"]["table_catalog"]["missing_count"] == 8 def test_migration_live_smoke_route_is_preview_only(): from routes.market_intel_routes import market_intel_bp app = Flask(__name__) app.secret_key = "test-secret" app.register_blueprint(market_intel_bp) client = app.test_client() with client.session_transaction() as session: session["logged_in"] = True response = client.get("/api/market_intel/migration_live_smoke?execute=false") data = response.get_json() assert response.status_code == 200 assert data["mode"] == "migration_live_smoke_preview" assert data["phase"] == "phase_48_manual_sample_acceptance_contract" assert data["execute_requested"] is False assert data["smoke_result"] == "planned_no_execution" assert data["migration_executed"] is False assert data["rollback_executed"] is False assert data["database_write_executed"] is False assert data["api_executes_migration"] is False def test_live_db_inventory_planned_is_preview_only(): inventory = MarketIntelService().build_live_db_inventory() assert inventory["mode"] == "live_db_inventory_planned" assert inventory["phase"] == "phase_48_manual_sample_acceptance_contract" assert inventory["execute_requested"] is False assert inventory["read_only_query_executed"] is False assert inventory["database_connection_opened"] is False assert inventory["database_session_created"] is False assert inventory["database_write_executed"] is False assert inventory["database_commit_executed"] is False assert inventory["migration_executed"] is False assert inventory["external_network_executed"] is False assert inventory["scheduler_attached"] is False assert inventory["summary_ready"] is False assert inventory["total_rows"] == 0 assert "execute_false_planned_only" in inventory["blocked_reasons"] assert "inventory_not_loaded" in inventory["blocked_reasons"] def test_live_db_inventory_sqlite_read_only_counts_market_tables(): engine = create_engine("sqlite:///:memory:") with engine.begin() as conn: conn.execute(text("CREATE TABLE market_platforms (code TEXT, name TEXT, enabled BOOLEAN)")) conn.execute(text("CREATE TABLE market_campaigns (platform_code TEXT, status TEXT)")) conn.execute(text("CREATE TABLE market_campaign_snapshots (status TEXT)")) conn.execute( text( "CREATE TABLE market_campaign_products " "(platform_code TEXT, is_active BOOLEAN, last_seen_at TEXT, discount_rate FLOAT)" ) ) conn.execute( text( "CREATE TABLE market_product_price_history " "(platform_code TEXT, crawled_at TEXT, price FLOAT)" ) ) conn.execute( text( "CREATE TABLE market_product_matches " "(match_status TEXT, match_score FLOAT, reviewed_at TEXT)" ) ) conn.execute( text( "CREATE TABLE market_crawler_runs " "(platform_code TEXT, status TEXT, dry_run BOOLEAN, started_at TEXT, finished_at TEXT, error_count INTEGER)" ) ) conn.execute( text( "CREATE TABLE market_alert_review_queue " "(review_state TEXT, priority_lane TEXT, total_score FLOAT, updated_at TEXT)" ) ) conn.execute( text( "INSERT INTO market_platforms (code, name, enabled) " "VALUES ('momo', 'MOMO', 0), ('pchome', 'PChome', 0)" ) ) conn.execute( text( "INSERT INTO market_campaigns (platform_code, status) " "VALUES ('momo', 'active'), ('pchome', 'unknown')" ) ) conn.execute(text("INSERT INTO market_campaign_snapshots (status) VALUES ('success')")) conn.execute( text( "INSERT INTO market_campaign_products " "(platform_code, is_active, last_seen_at, discount_rate) " "VALUES ('momo', 1, '2026-05-18 09:00:00', 0.2), " "('pchome', 0, '2026-05-18 08:00:00', 0.1)" ) ) conn.execute( text( "INSERT INTO market_product_price_history (platform_code, crawled_at, price) " "VALUES ('momo', '2026-05-18 09:00:00', 199)" ) ) conn.execute( text( "INSERT INTO market_product_matches (match_status, match_score, reviewed_at) " "VALUES ('needs_review', 0.72, NULL), ('confirmed', 0.91, '2026-05-18 10:00:00')" ) ) conn.execute( text( "INSERT INTO market_crawler_runs " "(platform_code, status, dry_run, started_at, finished_at, error_count) " "VALUES ('momo', 'success', 1, '2026-05-18 09:00:00', '2026-05-18 09:01:00', 0)" ) ) conn.execute( text( "INSERT INTO market_alert_review_queue " "(review_state, priority_lane, total_score, updated_at) " "VALUES ('draft', 'watch', 0.61, '2026-05-18 11:00:00')" ) ) inventory = build_live_db_inventory_preview( MARKET_INTEL_TABLES, execute_requested=True, engine=engine, database_type="sqlite", ) assert inventory["mode"] == "live_db_inventory_read_only" assert inventory["execute_requested"] is True assert inventory["read_only_query_executed"] is True assert inventory["database_connection_opened"] is True assert inventory["database_session_created"] is False assert inventory["database_write_executed"] is False assert inventory["database_commit_executed"] is False assert inventory["migration_executed"] is False assert inventory["external_network_executed"] is False assert inventory["scheduler_attached"] is False assert inventory["missing_tables"] == [] assert inventory["summary_ready"] is True assert inventory["table_counts"]["market_platforms"] == 2 assert inventory["table_counts"]["market_campaign_products"] == 2 assert inventory["total_rows"] == 12 assert {item["code"] for item in inventory["platform_breakdown"]} == {"momo", "pchome"} assert {item["status"] for item in inventory["campaign_status_breakdown"]} == {"active", "unknown"} assert {item["match_status"] for item in inventory["match_status_breakdown"]} == {"confirmed", "needs_review"} assert inventory["alert_review_state_breakdown"][0]["review_state"] == "draft" assert inventory["crawler_run_summary"][0]["status"] == "success" def test_live_db_inventory_route_is_preview_only(): from routes.market_intel_routes import market_intel_bp app = Flask(__name__) app.secret_key = "test-secret" app.register_blueprint(market_intel_bp) client = app.test_client() with client.session_transaction() as session: session["logged_in"] = True response = client.get("/api/market_intel/live_db_inventory?execute=false") data = response.get_json() assert response.status_code == 200 assert data["mode"] == "live_db_inventory_planned" assert data["phase"] == "phase_48_manual_sample_acceptance_contract" assert data["execute_requested"] is False assert data["read_only_query_executed"] is False assert data["database_write_executed"] is False assert data["migration_executed"] is False def test_seed_writer_cli_status_blocks_real_write(): status = MarketIntelService().build_seed_writer_cli_status( platform_code="all", execute_requested=True, approval_token=TEST_APPROVAL_TOKEN, approval_token_secret=TEST_APPROVAL_TOKEN, ) assert status["mode"] == "seed_writer_cli_blocked" assert status["execute_requested"] is True assert status["apply_real_write_requested"] is False assert status["approval_token_present"] is True assert status["approval_token_valid"] is True assert status["approval_token_secret_configured"] is True assert "approval_token_hint" not in status assert status["ready_for_real_write"] is False assert status["writes_executed"] is False assert status["would_write_database"] is False assert status["database_session_created"] is False assert status["explicit_transaction_opened"] is False assert status["database_write_executed"] is False assert status["database_commit_executed"] is False assert status["external_network_executed"] is False assert status["scheduler_attached"] is False assert status["exit_code"] == 2 assert "apply_real_write_requested" in status["blocked_reasons"] assert "approval_token_valid" not in status["blocked_reasons"] preview = status["transaction_preview"] assert preview["mode"] == "seed_transaction_preview_no_session" assert preview["statement_count"] == 4 assert preview["database_session_created"] is False assert preview["transaction_opened"] is False assert preview["database_commit_executed"] is False assert preview["database_snapshot_loaded"] is False assert preview["statements"][0]["table"] == "market_platforms" assert preview["statements"][0]["diff_status"] == "not_loaded_no_db_session" assert "ON CONFLICT (code) DO UPDATE SET" in preview["statements"][0]["sql_template"] assert preview["statements"][0]["parameter_payload_hash"] assert status["safety_contract"]["refuses_execute_without_apply_flag"] is True assert status["safety_contract"]["keeps_crawler_disabled_for_seed_write"] is True assert status["safety_contract"]["uses_core_connection_not_orm_session"] is True def test_seed_writer_cli_status_route_never_leaks_approval_token(monkeypatch): from routes.market_intel_routes import market_intel_bp monkeypatch.setenv("MARKET_INTEL_SEED_WRITE_APPROVAL", TEST_APPROVAL_TOKEN) app = Flask(__name__) app.secret_key = "test-secret" app.register_blueprint(market_intel_bp) client = app.test_client() with client.session_transaction() as session: session["logged_in"] = True response = client.get("/api/market_intel/seed_writer_cli_status?execute=true&platform=all") data = response.get_json() payload = json.dumps(data, ensure_ascii=False, sort_keys=True) assert response.status_code == 200 assert data["mode"] == "seed_writer_cli_blocked" assert data["execute_requested"] is True assert data["apply_real_write_requested"] is False assert data["approval_token_present"] is False assert data["approval_token_valid"] is False assert data["approval_token_secret_configured"] is True assert data["ready_for_real_write"] is False assert data["writes_executed"] is False assert data["would_write_database"] is False assert data["database_session_created"] is False assert data["database_commit_executed"] is False assert "approval_token_present" in data["blocked_reasons"] assert "approval_token_valid" in data["blocked_reasons"] assert "apply_real_write_requested" in data["blocked_reasons"] assert "approval_token_hint" not in payload assert TEST_APPROVAL_TOKEN not in payload assert "APPROVED_MARKET_INTEL_SEED_WRITE" not in payload def test_seed_writer_cli_real_write_sqlite_upserts_seed_rows(): engine = create_engine("sqlite:///:memory:") with engine.begin() as conn: conn.execute( text( """ CREATE TABLE market_platforms ( id INTEGER PRIMARY KEY AUTOINCREMENT, code TEXT NOT NULL UNIQUE, name TEXT NOT NULL, base_url TEXT, enabled BOOLEAN NOT NULL DEFAULT 0, crawl_policy_json TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) """ ) ) status = MarketIntelService().build_seed_writer_cli_status( platform_code="all", execute_requested=True, apply_real_write=True, approval_token=TEST_APPROVAL_TOKEN, approval_token_secret=TEST_APPROVAL_TOKEN, engine=engine, database_type="sqlite", ) with engine.connect() as conn: rows = conn.execute( text("SELECT code, enabled FROM market_platforms ORDER BY code") ).fetchall() assert status["mode"] == "seed_writer_cli_executed" assert status["ready_for_real_write"] is True assert status["writes_executed"] is True assert status["would_write_database"] is True assert status["database_connection_opened"] is True assert status["database_session_created"] is False assert status["explicit_transaction_opened"] is True assert status["database_write_executed"] is True assert status["database_commit_executed"] is True assert status["database_rollback_executed"] is False assert status["external_network_executed"] is False assert status["scheduler_attached"] is False assert status["exit_code"] == 0 assert status["blocked_reasons"] == [] assert status["execution_result"]["inserted_codes"] == [ "momo", "pchome", "coupang", "shopee", ] assert status["execution_result"]["updated_codes"] == [] assert [row[0] for row in rows] == ["coupang", "momo", "pchome", "shopee"] assert all(row[1] in (False, 0) for row in rows) def test_seed_writer_cli_script_outputs_blocked_plan(): env = { **os.environ, "MOMO_ALLOW_INSECURE_CONFIG_FOR_TESTS": "true", "SECRET_KEY": "test", "LOGIN_PASSWORD": "test", } result = subprocess.run( [sys.executable, "scripts/market_intel_seed_writer.py", "--platform", "all"], capture_output=True, check=False, env=env, text=True, ) data = json.loads(result.stdout) assert result.returncode == 0 assert data["mode"] == "seed_writer_cli_blocked" assert data["execute_requested"] is False assert data["apply_real_write_requested"] is False assert data["writes_executed"] is False assert data["database_session_created"] is False assert data["database_commit_executed"] is False assert data["exit_code"] == 0 assert data["transaction_preview"]["statement_count"] == 4 assert data["transaction_preview"]["transaction_opened"] is False