活動頁

import json import os import subprocess import sys from pathlib import Path 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.platform_seed_db_diff import build_platform_seed_db_diff_plan from services.market_intel.schema_db_probe import build_schema_db_probe_plan 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 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["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" 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-migration" in template assert "data-market-intel-migration-tables" 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_migration_blueprint" 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_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"]["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 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["schema_db_probe"]["read_only_query_executed"] is False assert readiness["platform_seed_db_diff"]["read_only_query_executed"] is False 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"] == 7 assert blueprint["forward_has_destructive_sql"] is False assert blueprint["safety_checks"]["forward_sql_additive_only"] is True assert blueprint["safety_checks"]["does_not_write_seed_rows"] 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_not_implemented" 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 "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 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-token", ) assert status["mode"] == "seed_writer_cli_blocked_skeleton" assert status["execute_requested"] is True assert status["approval_token_present"] is True 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["database_commit_executed"] is False assert status["external_network_executed"] is False assert status["scheduler_attached"] is False assert status["exit_code"] == 2 assert "execute_request_blocked_by_skeleton" in status["blocked_reasons"] assert "real_write_implementation_enabled" 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_in_this_phase"] is True assert status["safety_contract"]["keeps_crawler_disabled_for_seed_write"] is True 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_skeleton" assert data["execute_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