feat(ai): add external mcp rag integration readback
This commit is contained in:
73
tests/test_external_mcp_rag_integration_service.py
Normal file
73
tests/test_external_mcp_rag_integration_service.py
Normal file
@@ -0,0 +1,73 @@
|
||||
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
|
||||
Reference in New Issue
Block a user