Coverage for projects/04-llm-adapter-shadow/tests/test_shadow.py: 100%

1import json 

2 

3from src.llm_adapter.providers.mock import MockProvider 

4from src.llm_adapter.runner import Runner 

5from src.llm_adapter.provider_spi import ProviderRequest 

6 

7 

8def test_shadow_exec_records_metrics(tmp_path): 

9 primary = MockProvider("primary", base_latency_ms=5, error_markers=set()) 

10 shadow = MockProvider("shadow", base_latency_ms=5, error_markers=set()) 

11 runner = Runner([primary]) 

12 

13 metrics_path = tmp_path / "metrics.jsonl" 

14 response = runner.run( 

15 ProviderRequest(prompt="hello"), 

16 shadow=shadow, 

17 shadow_metrics_path=metrics_path, 

18 ) 

19 

20 assert response.text.startswith("echo(primary):") 

21 assert metrics_path.exists() 

22 

23 payloads = [json.loads(line) for line in metrics_path.read_text().splitlines() if line.strip()] 

24 diff_event = next(item for item in payloads if item["event"] == "shadow_diff") 

25 success_event = next(item for item in payloads if item["event"] == "provider_success") 

26 

27 assert diff_event["primary_provider"] == "primary" 

28 assert diff_event["shadow_provider"] == "shadow" 

29 assert diff_event["shadow_ok"] is True 

30 assert diff_event["primary_text_len"] == len(response.text) 

31 assert diff_event["primary_token_usage_total"] == response.token_usage.total 

32 assert diff_event["request_fingerprint"] 

33 

34 assert success_event["provider"] == "primary" 

35 assert success_event["attempt"] == 1 

36 assert success_event["shadow_used"] is True 

37 assert success_event["latency_ms"] == response.latency_ms 

38 

39 expected_tokens = max(1, len("hello") // 4) + 16 

40 assert diff_event["shadow_token_usage_total"] == expected_tokens 

41 assert diff_event["shadow_text_len"] == len("echo(shadow): hello") 

42 

43 

44def test_shadow_error_records_metrics(tmp_path): 

45 primary = MockProvider("primary", base_latency_ms=5, error_markers=set()) 

46 shadow = MockProvider("shadow", base_latency_ms=5, error_markers={"[TIMEOUT]"}) 

47 runner = Runner([primary]) 

48 

49 metrics_path = tmp_path / "metrics.jsonl" 

50 runner.run( 

51 ProviderRequest(prompt="[TIMEOUT] hello"), 

52 shadow=shadow, 

53 shadow_metrics_path=metrics_path, 

54 ) 

55 

56 payloads = [json.loads(line) for line in metrics_path.read_text().splitlines() if line.strip()] 

57 diff_event = next(item for item in payloads if item["event"] == "shadow_diff") 

58 

59 assert diff_event["shadow_ok"] is False 

60 assert diff_event["shadow_error"] == "TimeoutError" 

61 assert diff_event["shadow_error_message"] == "simulated timeout" 

62 assert diff_event["shadow_duration_ms"] >= 0 

63 

64 

65def test_request_hash_includes_max_tokens(tmp_path): 

66 provider = MockProvider("primary", base_latency_ms=1, error_markers=set()) 

67 runner = Runner([provider]) 

68 

69 metrics_path = tmp_path / "metrics.jsonl" 

70 

71 runner.run( 

72 ProviderRequest(prompt="hello", max_tokens=32), 

73 shadow_metrics_path=metrics_path, 

74 ) 

75 runner.run( 

76 ProviderRequest(prompt="hello", max_tokens=64), 

77 shadow_metrics_path=metrics_path, 

78 ) 

79 

80 payloads = [json.loads(line) for line in metrics_path.read_text().splitlines() if line.strip()] 

81 success_events = [item for item in payloads if item["event"] == "provider_success"] 

82 

83 assert len(success_events) == 2 

84 request_hashes = {event["request_hash"] for event in success_events} 

85 fingerprints = {event["request_fingerprint"] for event in success_events} 

86 

87 assert len(request_hashes) == 2 

88 assert len(fingerprints) == 2