File size: 2,571 Bytes
d2585c1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
"""결과 리포트 생성 — JSON + 터미널 요약."""

from __future__ import annotations

import json
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional

from .config import BASE_URL, VALID_TOOLS
from .flow_tracker import LatencyAggregator
from .http_client import get_http_backend
from .logger import E2ELogger


def write_json_report(
    results: List[dict],
    output_path: str,
    run_id: str,
    cold_start_wait: float = 0.0,
    observed_tools: Optional[set] = None,
    latency_aggregator: Optional[LatencyAggregator] = None,
) -> None:
    """JSON 결과 파일을 출력한다."""
    _observed = observed_tools or set()
    passed = sum(1 for r in results if r.get("status") == "passed")
    failed = sum(1 for r in results if r.get("status") == "failed")
    skipped = sum(1 for r in results if r.get("status") == "skipped")

    tool_ratio = len(_observed) / len(VALID_TOOLS) if VALID_TOOLS else 0

    output: Dict[str, Any] = {
        "meta": {
            "run_id": run_id,
            "timestamp_utc": datetime.now(timezone.utc).isoformat(),
            "target_url": BASE_URL,
            "cold_start_wait_seconds": cold_start_wait,
            "http_backend": get_http_backend(),
        },
        "summary": {
            "total": len(results),
            "passed": passed,
            "failed": failed,
            "skipped": skipped,
            "tool_coverage": {
                "observed": sorted(_observed),
                "ratio": round(tool_ratio, 2),
            },
        },
        "scenarios": results,
    }

    if latency_aggregator:
        output["latency_stats"] = latency_aggregator.all_stats()

    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(output, f, ensure_ascii=False, indent=2)


def print_summary(results: List[dict], logger: E2ELogger, observed_tools: set) -> None:
    """터미널에 결과 요약을 출력한다."""
    passed = sum(1 for r in results if r.get("status") == "passed")
    failed = sum(1 for r in results if r.get("status") == "failed")
    skipped = sum(1 for r in results if r.get("status") == "skipped")
    total = len(results)

    logger.info("=" * 60)
    logger.info(f"결과: {passed}/{total} 통과, {failed} 실패, {skipped} 스킵")

    tool_ratio = len(observed_tools) / len(VALID_TOOLS) if VALID_TOOLS else 0
    logger.info(f"도구 커버리지: {len(observed_tools)}/{len(VALID_TOOLS)} ({tool_ratio:.0%})")
    if observed_tools:
        logger.info(f"  관측된 도구: {sorted(observed_tools)}")