74 lines
2.8 KiB
Python
74 lines
2.8 KiB
Python
import json
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
def test_external_mcp_rag_readback_reports_partial_truth():
|
|
from services.external_mcp_rag_integration_service import (
|
|
POLICY,
|
|
build_external_mcp_rag_integration_readback,
|
|
)
|
|
|
|
payload = build_external_mcp_rag_integration_readback()
|
|
inventory = {item["id"]: item for item in payload["inventory"]}
|
|
|
|
assert payload["policy"] == POLICY
|
|
assert payload["status"] == "partially_integrated"
|
|
assert payload["completion"]["all_absorbed"] is False
|
|
assert payload["completion"]["unresolved_count"] >= 1
|
|
assert "不是全部完成" in payload["answer_to_owner"]
|
|
assert inventory["rag.pixelrag.visual_evidence"]["status"] == "integrated_phase1_visual_receipts"
|
|
assert inventory["rag.qwen3_vl_embedding"]["status"] == "deferred_until_ollama_first_verified"
|
|
assert inventory["rag.faiss_visual_index"]["status"] == "rejected_for_production_without_adr"
|
|
assert inventory["mcp.gemini_grounding"]["status"] == "disabled_by_default_fallback_only"
|
|
assert payload["controlled_apply"]["db_write"] is False
|
|
assert payload["controlled_apply"]["network_call"] is False
|
|
|
|
|
|
def test_external_mcp_rag_readback_can_filter_family():
|
|
from services.external_mcp_rag_integration_service import (
|
|
build_external_mcp_rag_integration_readback,
|
|
)
|
|
|
|
payload = build_external_mcp_rag_integration_readback(target_family="external_rag")
|
|
|
|
assert payload["completion"]["total_capabilities"] == 3
|
|
assert {item["family"] for item in payload["inventory"]} == {"external_rag"}
|
|
|
|
|
|
def test_external_mcp_rag_cli_outputs_machine_readable_json():
|
|
completed = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"scripts/ops/report_external_mcp_rag_integration.py",
|
|
"--family",
|
|
"external_rag",
|
|
],
|
|
capture_output=True,
|
|
check=False,
|
|
text=True,
|
|
)
|
|
|
|
assert completed.returncode == 0
|
|
payload = json.loads(completed.stdout)
|
|
assert payload["success"] is True
|
|
assert payload["completion"]["total_capabilities"] == 3
|
|
assert payload["inventory"][0]["family"] == "external_rag"
|
|
|
|
|
|
def test_external_mcp_rag_ai_automation_route_returns_readback():
|
|
from flask import Flask
|
|
from routes import system_public_routes as routes
|
|
|
|
app = Flask(__name__)
|
|
with app.test_request_context(
|
|
"/api/ai-automation/external-mcp-rag-integration?family=external_mcp"
|
|
):
|
|
response = routes.ai_automation_external_mcp_rag_integration_api.__wrapped__()
|
|
payload = response.get_json()
|
|
|
|
assert payload["policy"] == "read_only_external_mcp_rag_integration_readback_v1"
|
|
assert payload["success"] is True
|
|
assert {item["family"] for item in payload["inventory"]} == {"external_mcp"}
|
|
assert payload["runtime"]["mcp"]["caller_count"] >= 1
|