"""Basic parity check between CLI analyzer and MCP tool output."""
from __future__ import annotations
import asyncio
import sys
from pathlib import Path
from typing import Any, Dict
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from mcp_server import analyze_dps_logs
from server import DEFAULT_SAMPLE_PATH, analyze_logs
import json
import argparse
def normalize_payload(obj):
"""Recursively remove volatile keys from dicts/lists."""
if isinstance(obj, dict):
return {k: normalize_payload(v) for k, v in obj.items() if k != "generated_at"}
if isinstance(obj, list):
return [normalize_payload(x) for x in obj]
return obj
def main() -> None:
parser = argparse.ArgumentParser(description="Compare CLI and MCP analyzer outputs deterministically.")
parser.add_argument("--sample", action="store_true", help="Use the bundled sample log.")
parser.add_argument("--input-path", type=str, help="File or directory containing TL logs.")
parser.add_argument("--limit-runs", type=int, help="Newest N files if input-path is a directory.")
args = parser.parse_args()
# Determine input path and limit
if args.sample:
input_path = DEFAULT_SAMPLE_PATH
limit_runs = None
else:
if not args.input_path:
parser.error("Must provide --sample or --input-path.")
input_path = Path(args.input_path)
limit_runs = args.limit_runs
# Run CLI analyzer
cli_payload: Dict[str, Any] = analyze_logs(input_path, limit_runs=limit_runs)
# Run MCP tool
tool_payload: Dict[str, Any] = asyncio.run(analyze_dps_logs(
log_dir=str(input_path),
limit_runs=limit_runs
))
norm_cli = normalize_payload(cli_payload)
norm_tool = normalize_payload(tool_payload)
if norm_cli != norm_tool:
print("Payloads differ after normalization. CLI:")
print(json.dumps(norm_cli, sort_keys=True, indent=2))
print("\nMCP tool:")
print(json.dumps(norm_tool, sort_keys=True, indent=2))
raise AssertionError("MCP tool payload diverges from CLI analyzer output (normalized)")
print("MCP tool matches CLI analyzer output for the given input.")
if __name__ == "__main__":
main()
