Coverage for narrative_harm_classifier/api/routes/tracking.py: 100%

1""" 

2api/routes/tracking.py — Escalation-chain tracking endpoints. 

3""" 

4 

5from fastapi import APIRouter, Depends, HTTPException 

6 

7from narrative_harm_classifier.core.models import ClassifyRequest 

8from narrative_harm_classifier.core.config import get_settings, Settings 

9from narrative_harm_classifier.classifier.factory import build_tracker 

10from narrative_harm_classifier.classifier.tracking.models import SourceProfile, Observation, ChainVerification 

11from narrative_harm_classifier.classifier.tracking.tracker import EscalationTracker 

12 

13router = APIRouter() 

14 

15 

16def get_tracker(settings: Settings = Depends(get_settings)) -> EscalationTracker: 

17 return build_tracker(settings) 

18 

19 

20@router.post( 

21 "/{source_id}/observe", 

22 response_model=Observation, 

23 summary="Classify a text and record it against a tracked source", 

24 description=( 

25 "Classifies the text and appends it to the source's observation history, " 

26 "used to compute escalation trend and risk level over time." 

27 ), 

28) 

29def observe( 

30 source_id: str, 

31 request: ClassifyRequest, 

32 tracker: EscalationTracker = Depends(get_tracker), 

33) -> Observation: 

34 return tracker.observe(source_id, request) 

35 

36 

37@router.get( 

38 "/{source_id}", 

39 response_model=SourceProfile, 

40 summary="Get a tracked source's escalation profile", 

41) 

42def get_profile( 

43 source_id: str, 

44 window: int = 20, 

45 tracker: EscalationTracker = Depends(get_tracker), 

46) -> SourceProfile: 

47 profile = tracker.profile(source_id, window=window) 

48 if profile.observation_count == 0: 

49 raise HTTPException(status_code=404, detail=f"No observations recorded for source '{source_id}'") 

50 return profile 

51 

52 

53@router.get( 

54 "", 

55 response_model=list[SourceProfile], 

56 summary="List all tracked sources, sorted by risk (highest first)", 

57) 

58def list_profiles( 

59 window: int = 20, 

60 tracker: EscalationTracker = Depends(get_tracker), 

61) -> list[SourceProfile]: 

62 return tracker.list_profiles(window=window) 

63 

64 

65@router.get( 

66 "/{source_id}/verify", 

67 response_model=ChainVerification, 

68 summary="Verify the tamper-evident hash chain for a source's observation history", 

69 description=( 

70 "Recomputes the hash chain over the full stored history and confirms it's intact. " 

71 "Detects tampering with any historical record; does not prevent it." 

72 ), 

73) 

74def verify_chain( 

75 source_id: str, 

76 tracker: EscalationTracker = Depends(get_tracker), 

77) -> ChainVerification: 

78 result = tracker.verify_chain(source_id) 

79 if result.observation_count == 0: 

80 raise HTTPException(status_code=404, detail=f"No observations recorded for source '{source_id}'") 

81 return result