From ebddd4e063845f15ca0acb01ba0896475b1fe3d4 Mon Sep 17 00:00:00 2001 From: OoO Date: Tue, 19 May 2026 21:59:19 +0800 Subject: [PATCH] =?UTF-8?q?[V10.279]=20=E8=A3=9C=20Phase80=20AI=20summary?= =?UTF-8?q?=20preflight=20=E5=96=AE=E5=85=83=E6=B8=AC=E8=A9=A6=20|=20test?= =?UTF-8?q?=5Fmarket=5Fintel=5Fskeleton.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_market_intel_skeleton.py | 235 ++++++++++++++++++++++++++++ 1 file changed, 235 insertions(+) diff --git a/tests/test_market_intel_skeleton.py b/tests/test_market_intel_skeleton.py index ef1a07e..d231866 100644 --- a/tests/test_market_intel_skeleton.py +++ b/tests/test_market_intel_skeleton.py @@ -5280,6 +5280,172 @@ def test_candidate_queue_review_archive_summary_blocks_llm_and_writes(): ) +def test_candidate_queue_review_ai_summary_preflight_is_ollama_first_preview(): + from services.market_intel.candidate_queue_review_ai_summary_preflight import ( + build_candidate_queue_review_ai_summary_preflight, + ) + from services.market_intel.candidate_queue_review_archive_summary import ( + build_candidate_queue_review_archive_summary, + ) + from services.market_intel.candidate_queue_review_completion_archive import ( + build_candidate_queue_review_completion_archive, + ) + from services.market_intel.candidate_queue_review_decision_post_closeout_inventory import ( + build_candidate_queue_review_decision_post_closeout_inventory, + ) + from services.market_intel.candidate_queue_review_decision_writer_run_closeout import ( + build_candidate_queue_review_decision_writer_run_closeout, + ) + + fixture = _build_review_decision_writer_receipt_fixture( + "sample-batch-review-ai-summary-preflight" + ) + preflight_evidence = { + **fixture["operator_evidence"], + "review_state_receipt_artifact_path": ( + "artifacts/market_intel/review-state-receipt.json" + ), + "review_state_closeout_artifact_path": ( + "artifacts/market_intel/review-state-closeout.json" + ), + "post_closeout_inventory_artifact_path": ( + "artifacts/market_intel/post-closeout-inventory.json" + ), + "review_completion_archive_path": ( + "artifacts/market_intel/review-completion-archive.json" + ), + "archive_summary_artifact_path": ( + "artifacts/market_intel/review-archive-summary.json" + ), + "ai_summary_preflight_artifact_path": ( + "artifacts/market_intel/ai-summary-preflight.json" + ), + "operator_confirmed_review_state_closeout_next": True, + "operator_confirmed_post_closeout_inventory_read_only": True, + "operator_confirmed_review_completion_archive": True, + "operator_confirmed_archive_is_read_only": True, + "operator_confirmed_ai_summary_review": True, + "operator_confirmed_summary_is_read_only": True, + "operator_confirmed_ai_summary_preflight": True, + "operator_confirmed_ollama_first": True, + "operator_confirmed_gemini_backup_only": True, + "operator_confirmed_no_llm_call": True, + "operator_confirmed_no_telegram_dispatch": True, + "operator_confirmed_no_scheduler_attach": True, + "operator_confirmed_no_api_db_write": True, + } + closeout = build_candidate_queue_review_decision_writer_run_closeout( + transaction_preview=fixture["transaction"], + run_receipt=fixture["receipt"], + operator_evidence=preflight_evidence, + ) + inventory = build_candidate_queue_review_decision_post_closeout_inventory( + transaction_preview=fixture["transaction"], + run_closeout=closeout, + postwrite_smoke=fixture["postwrite_smoke_result"], + live_db_inventory={ + "mode": "live_db_inventory_loaded", + "summary_ready": True, + "read_only_query_executed": True, + "database_connection_opened": True, + "database_write_executed": False, + "database_commit_executed": False, + "migration_executed": False, + "external_network_executed": False, + "scheduler_attached": False, + "table_statuses": [ + { + "table": "market_alert_review_queue", + "exists": True, + "row_count": 1, + "status": "loaded", + } + ], + }, + operator_evidence=preflight_evidence, + execute_requested=True, + ) + archive = build_candidate_queue_review_completion_archive( + transaction_preview=fixture["transaction"], + run_receipt=fixture["receipt"], + run_closeout=closeout, + post_closeout_inventory=inventory, + operator_evidence=preflight_evidence, + execute_requested=True, + ) + archive_summary = build_candidate_queue_review_archive_summary( + review_completion_archive=archive, + operator_evidence=preflight_evidence, + execute_requested=True, + ) + preflight = build_candidate_queue_review_ai_summary_preflight( + archive_summary=archive_summary, + operator_evidence=preflight_evidence, + execute_requested=True, + ) + token_leak = build_candidate_queue_review_ai_summary_preflight( + archive_summary=archive_summary, + operator_evidence={ + **preflight_evidence, + "approval_token": TEST_APPROVAL_TOKEN, + }, + execute_requested=True, + ) + + assert preflight["mode"] == "candidate_queue_review_ai_summary_preflight_preview" + assert preflight["target_operation"] == "preflight_ollama_first_ai_summary" + assert preflight["ai_summary_preflight_ready"] is True + assert preflight["ready_for_manual_ollama_summary_run"] is True + assert preflight["ready_for_ai_summary_generation"] is False + assert preflight["ready_for_llm_call"] is False + assert preflight["ready_for_telegram_dispatch"] is False + assert preflight["api_reads_approval_token"] is False + assert preflight["api_writes_file"] is False + assert preflight["api_writes_database"] is False + assert preflight["api_updates_review_state"] is False + assert preflight["summary_file_written"] is False + assert preflight["summary_record_written"] is False + assert preflight["summary_manifest_written"] is False + assert preflight["ai_summary_generated"] is False + assert preflight["llm_call_executed"] is False + assert preflight["ollama_call_executed"] is False + assert preflight["gemini_call_executed"] is False + assert preflight["telegram_dispatched"] is False + assert preflight["database_write_executed"] is False + assert preflight["database_commit_executed"] is False + assert preflight["scheduler_attached"] is False + assert preflight["blocked_reasons"] == [] + assert preflight["model_route_policy"]["primary_policy"] == "ollama_first" + assert len(preflight["model_route_policy"]["primary_cascade"]) == 3 + assert ( + preflight["model_route_policy"]["fallback_policy"] + == "gemini_backup_only_after_ollama_cascade_failure" + ) + assert ( + preflight["model_route_policy"]["app_host_forbidden_as_ollama_node"] + == "192.168.0.188" + ) + assert "ollama_cascade_must_use_gcp_a_then_gcp_b_then_111" in preflight[ + "safe_boundaries" + ] + assert "host_188_forbidden_as_ollama_node" in preflight["safe_boundaries"] + assert "do_not_call_llm_from_ai_summary_preflight" in preflight[ + "safe_boundaries" + ] + assert token_leak["ai_summary_preflight_ready"] is False + assert token_leak["operator_ai_summary_preflight"][ + "approval_token_submitted_to_api" + ] is True + assert "ai_summary_preflight_no_approval_token_submitted_to_api" in token_leak[ + "blocked_reasons" + ] + assert TEST_APPROVAL_TOKEN not in json.dumps( + token_leak, + ensure_ascii=False, + sort_keys=True, + ) + + def test_candidate_queue_writer_preflight_route_is_post_only_and_no_write(): from routes.market_intel_routes import market_intel_bp @@ -7217,6 +7383,75 @@ def test_candidate_queue_review_archive_summary_route_is_post_only_and_no_write( assert TEST_APPROVAL_TOKEN not in payload +def test_candidate_queue_review_ai_summary_preflight_route_is_post_only_and_no_write(): + from routes.market_intel_routes import market_intel_bp + from routes.market_intel_review_routes import market_intel_review_bp + + app = Flask(__name__) + app.secret_key = "test-secret" + app.register_blueprint(market_intel_bp) + app.register_blueprint(market_intel_review_bp) + client = app.test_client() + with client.session_transaction() as session: + session["logged_in"] = True + + get_response = client.get( + "/api/market_intel/manual_sample_review/" + "candidate_queue_review_ai_summary_preflight" + ) + response = client.post( + "/api/market_intel/manual_sample_review/" + "candidate_queue_review_ai_summary_preflight" + "?execute=true&apply_real_write=true", + json={"sample_result": {}}, + ) + data = response.get_json() + payload = json.dumps(data, ensure_ascii=False, sort_keys=True) + + assert get_response.status_code == 405 + assert response.status_code == 200 + assert data["mode"] == "candidate_queue_review_ai_summary_preflight_preview" + assert data["phase"] == "phase_80_candidate_queue_review_ai_summary_preflight" + assert data["ai_summary_preflight_ready"] is False + assert data["ready_for_manual_ollama_summary_run"] is False + assert data["ready_for_ai_summary_generation"] is False + assert data["ready_for_llm_call"] is False + assert data["ready_for_telegram_dispatch"] is False + assert data["ready_for_api_review_state_update"] is False + assert data["ready_for_api_database_write"] is False + assert data["ready_for_scheduler_attach"] is False + assert data["api_writes_file"] is False + assert data["api_executes_cli"] is False + assert data["api_reads_approval_token"] is False + assert data["api_writes_database"] is False + assert data["api_updates_review_state"] is False + assert data["summary_file_written"] is False + assert data["summary_record_written"] is False + assert data["summary_manifest_written"] is False + assert data["ai_summary_generated"] is False + assert data["llm_call_executed"] is False + assert data["ollama_call_executed"] is False + assert data["gemini_call_executed"] is False + assert data["telegram_dispatched"] is False + assert data["database_write_executed"] is False + assert data["database_commit_executed"] is False + assert data["review_state_update_executed"] is False + assert data["read_only_query_executed"] is False + assert data["scheduler_attached"] is False + assert data["expected_dedupe_keys"] == [] + assert "archive_summary_ready" in data["blocked_reasons"] + assert "do_not_call_llm_from_ai_summary_preflight" in data["safe_boundaries"] + assert ( + data["model_route_policy"]["primary_policy"] + == "ollama_first" + ) + assert ( + data["model_route_policy"]["fallback_policy"] + == "gemini_backup_only_after_ollama_cascade_failure" + ) + assert TEST_APPROVAL_TOKEN not in payload + + def test_candidate_queue_writer_run_receipt_route_is_post_only_and_no_write(): from routes.market_intel_routes import market_intel_bp