ewoooc/tests/test_ai_agent_product_integration_service.py

import json
import subprocess
import sys


def _telemetry(*, all_active):
    from services.ai_agent_product_integration_service import AGENT_SPECS

    agents = {}
    for index, key in enumerate(AGENT_SPECS):
        active = all_active or index < 2
        agents[key] = {
            "calls": 5 if active else 0,
            "errors": 0,
            "rag_hits": 1 if active else 0,
            "fallbacks": 0,
            "mcp_calls": 1 if all_active else 0,
            "rag_queries": 1 if all_active else 0,
            "rag_query_hits": 1 if all_active else 0,
            "ai_insights": 1 if active else 0,
            "action_plans": 1 if active else 0,
            "executed_action_plans": 1 if all_active else 0,
            "last_called": "2026-07-17T00:00:00+00:00" if active else None,
        }
    return {
        "agents": agents,
        "unmatched_ai_callers": [],
        "totals": {
            "ai_calls": sum(item["calls"] for item in agents.values()),
            "ai_errors": 0,
            "mcp_calls": 4 if all_active else 0,
            "rag_queries": 4 if all_active else 0,
            "rag_query_hits": 4 if all_active else 0,
            "ai_insights": 4,
            "action_plans": 4,
            "executed_action_plans": 4 if all_active else 0,
            "action_outcomes": 2 if all_active else 0,
            "heal_logs": 2 if all_active else 0,
            "heal_success": 2 if all_active else 0,
            "verified_retry_or_rollback_incidents": 1 if all_active else 0,
        },
        "embedding_retry_queue": {"done": 8},
        "read_errors": [],
    }


def _external_runtime(*, enabled):
    return {
        "runtime": {
            "mcp": {"enabled": enabled},
            "rag": {"enabled": enabled},
            "pixelrag": {"enabled": True, "platform_count": 8},
        }
    }


def _canary(*, passed):
    return {"latest_execution": {"canary_passed": passed}}


def test_ai_agent_integration_reports_production_like_partial_truth(monkeypatch):
    from services import ai_agent_product_integration_service as service

    monkeypatch.setattr(
        service,
        "_collect_runtime_telemetry",
        lambda _since: _telemetry(all_active=False),
    )
    monkeypatch.setattr(
        service,
        "build_external_mcp_rag_integration_readback",
        lambda: _external_runtime(enabled=False),
    )
    monkeypatch.setattr(
        service,
        "run_internal_rag_candidate_canary",
        lambda **_kwargs: _canary(passed=False),
    )
    payload = service.build_ai_agent_product_integration_readback(window_hours=168)

    assert payload["status"] == "partially_integrated"
    assert payload["completion"]["source_wired_agents"] == 4
    assert payload["completion"]["runtime_active_agents"] == 2
    assert payload["completion"]["full_product_integration"] is False
    assert "mcp_router_runtime_disabled" in payload["blockers"]
    assert "rag_runtime_disabled" in payload["blockers"]
    assert "尚未完整整合" in payload["answer_to_owner"]


def test_ai_agent_integration_requires_full_runtime_closure(monkeypatch):
    from services import ai_agent_product_integration_service as service

    monkeypatch.setattr(
        service,
        "_collect_runtime_telemetry",
        lambda _since: _telemetry(all_active=True),
    )
    monkeypatch.setattr(
        service,
        "build_external_mcp_rag_integration_readback",
        lambda: _external_runtime(enabled=True),
    )
    monkeypatch.setattr(
        service,
        "run_internal_rag_candidate_canary",
        lambda **_kwargs: _canary(passed=True),
    )
    payload = service.build_ai_agent_product_integration_readback()

    assert payload["status"] == "fully_integrated"
    assert payload["completion"]["runtime_active_agents"] == 4
    assert payload["completion"]["closure_stage_passed"] == 9
    assert payload["completion"]["full_product_integration"] is True
    assert payload["blockers"] == []


def test_ai_agent_integration_route_returns_truth(monkeypatch):
    from flask import Flask
    from routes import system_public_routes as routes
    from services import ai_agent_product_integration_service as service

    monkeypatch.setattr(
        service,
        "build_ai_agent_product_integration_readback",
        lambda **kwargs: {
            "success": True,
            "policy": service.POLICY,
            "status": "partially_integrated",
            "window_hours": kwargs["window_hours"],
        },
    )
    app = Flask(__name__)
    with app.test_request_context(
        "/api/ai-automation/agent-product-integration?window_hours=72"
    ):
        response = routes.ai_automation_agent_product_integration_api.__wrapped__()
    payload = response.get_json()

    assert payload["policy"] == "runtime_truth_ai_agent_product_integration_v1"
    assert payload["window_hours"] == 72


def test_ai_agent_integration_cli_outputs_json():
    completed = subprocess.run(
        [
            sys.executable,
            "scripts/ops/report_ai_agent_product_integration.py",
            "--window-hours",
            "1",
        ],
        capture_output=True,
        check=False,
        text=True,
    )

    assert completed.returncode == 0
    payload = json.loads(completed.stdout)
    assert payload["policy"] == "runtime_truth_ai_agent_product_integration_v1"
    assert payload["controlled_apply"]["database_write"] is False