#!/usr/bin/env python3 """Tests for Story 1.04 (Track B): ExecutionTelemetry Interceptor Black Box Tests: BB1: pre_execute passes payload through unchanged + logs dispatch_start BB2: post_execute logs dispatch_complete with positive duration_ms BB3: on_error logs dispatch_error with error_type field BB4: on_correction logs dispatch_correction with attempt number White Box Tests: WB1: _start_times populated during pre_execute WB2: _start_times consumed (removed) in post_execute WB3: All events are append-only (multiple writes visible in file) WB4: isinstance check + correct priority metadata """ import asyncio import json import os import tempfile import sys sys.path.insert(0, '/mnt/e/genesis-system') def _read_events(path): events = [] with open(path, 'r') as f: for line in f: line = line.strip() if line: events.append(json.loads(line)) return events def test_execution_telemetry(): # Use temp file to avoid polluting real events log with tempfile.NamedTemporaryFile(mode='w', suffix='.jsonl', delete=False) as f: test_log = f.name os.environ["EVENTS_LOG_PATH"] = test_log # Reload module to pick up env var change import importlib if 'core.interceptors.execution_telemetry' in sys.modules: importlib.reload(sys.modules['core.interceptors.execution_telemetry']) from core.interceptors.execution_telemetry import ExecutionTelemetryInterceptor from core.interceptors.base_interceptor import BaseInterceptor telemetry = ExecutionTelemetryInterceptor() async def run_tests(): # --- BB1: pre_execute passes payload through unchanged + logs dispatch_start --- payload = {"task_id": "test-001", "type": "build", "prompt": "test", "tier": "pro"} result = await telemetry.pre_execute(payload) assert result is payload, "pre_execute MUST return the same dict object (pass-through)" events = _read_events(test_log) assert len(events) >= 1, "Expected at least 1 event after pre_execute" start_evt = events[-1] assert start_evt["event"] == "dispatch_start", f"Expected dispatch_start, got {start_evt['event']}" assert start_evt["task_id"] == "test-001" assert start_evt["task_type"] == "build" assert start_evt["tier"] == "pro" assert "timestamp" in start_evt print("BB1 PASSED: pre_execute pass-through + dispatch_start logged") # --- BB2: post_execute logs dispatch_complete with positive duration_ms --- import time time.sleep(0.01) # Small delay to ensure measurable monotonic duration await telemetry.post_execute({"status": "completed"}, payload) events = _read_events(test_log) complete_events = [e for e in events if e["event"] == "dispatch_complete"] assert len(complete_events) >= 1, "Expected at least 1 dispatch_complete event" c_evt = complete_events[-1] assert c_evt["task_id"] == "test-001" assert c_evt["status"] == "completed" assert c_evt["duration_ms"] > 0, f"duration_ms should be > 0, got {c_evt['duration_ms']}" assert "timestamp" in c_evt print(f"BB2 PASSED: dispatch_complete logged with duration_ms={c_evt['duration_ms']}") # --- BB3: on_error logs dispatch_error with error_type --- await telemetry.on_error(ValueError("test error"), {"task_id": "test-002"}) events = _read_events(test_log) error_events = [e for e in events if e["event"] == "dispatch_error"] assert len(error_events) >= 1, "Expected at least 1 dispatch_error event" e_evt = error_events[-1] assert e_evt["task_id"] == "test-002" assert e_evt["error_type"] == "ValueError" assert e_evt["error_msg"] == "test error" assert "timestamp" in e_evt print("BB3 PASSED: dispatch_error logged with correct error_type") # --- BB4: on_correction logs dispatch_correction with attempt number --- corr_result = await telemetry.on_correction({"task_id": "test-003", "attempt": 2, "data": "x"}) events = _read_events(test_log) corr_events = [e for e in events if e["event"] == "dispatch_correction"] assert len(corr_events) >= 1, "Expected at least 1 dispatch_correction event" corr_evt = corr_events[-1] assert corr_evt["task_id"] == "test-003" assert corr_evt["attempt"] == 2 assert "timestamp" in corr_evt # on_correction must pass correction_payload through assert corr_result["data"] == "x", "on_correction must return the correction_payload" print("BB4 PASSED: dispatch_correction logged with attempt=2, payload passed through") # --- WB1: _start_times populated in pre_execute --- payload2 = {"task_id": "test-004"} await telemetry.pre_execute(payload2) assert "test-004" in telemetry._start_times, "_start_times should contain task_id after pre_execute" print("WB1 PASSED: _start_times populated during pre_execute") # --- WB2: _start_times consumed in post_execute --- await telemetry.post_execute({"status": "ok"}, payload2) assert "test-004" not in telemetry._start_times, "_start_times should not contain task_id after post_execute" print("WB2 PASSED: _start_times consumed in post_execute") # --- WB3: Append-only — multiple events accumulate in file --- events = _read_events(test_log) assert len(events) >= 6, f"Expected >= 6 events total (append-only), got {len(events)}" event_types = [e["event"] for e in events] assert "dispatch_start" in event_types assert "dispatch_complete" in event_types assert "dispatch_error" in event_types assert "dispatch_correction" in event_types print(f"WB3 PASSED: Append-only confirmed — {len(events)} events in file") # --- WB4: isinstance check + metadata --- assert isinstance(telemetry, BaseInterceptor), "Must be instance of BaseInterceptor" assert telemetry.metadata.name == "execution_telemetry" assert telemetry.metadata.priority == 0, f"Priority should be 0 (first), got {telemetry.metadata.priority}" assert telemetry.metadata.enabled is True print("WB4 PASSED: isinstance BaseInterceptor, priority=0, enabled=True") asyncio.run(run_tests()) # Cleanup os.unlink(test_log) if "EVENTS_LOG_PATH" in os.environ: del os.environ["EVENTS_LOG_PATH"] print("\nALL 8 TESTS PASSED — Story 1.04 (Track B): ExecutionTelemetry Interceptor") if __name__ == "__main__": test_execution_telemetry()