Coverage for src/case_router/deadlines.py: 100%

1"""Response deadlines: how long the support team has to answer a case.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Mapping 

6from dataclasses import dataclass, field 

7from datetime import datetime, timedelta 

8from types import MappingProxyType 

9 

10from .models import Priority, Tier 

11 

12DEFAULT_HOURS: Mapping[Priority, float] = MappingProxyType( 

13 {Priority.CRITICAL: 1, Priority.HIGH: 4, Priority.MEDIUM: 8, Priority.LOW: 24} 

14) 

15 

16 

17@dataclass(frozen=True) 

18class ResponsePolicy: 

19 """Hours to first response by priority. Premium customers get a shorter window.""" 

20 

21 hours: Mapping[Priority, float] = field(default=DEFAULT_HOURS) 

22 premium_factor: float = 0.5 

23 

24 def __post_init__(self) -> None: 

25 missing = [p for p in Priority if p not in self.hours] 

26 if missing: 

27 raise ValueError(f"no response window for {', '.join(missing)}") 

28 if any(h <= 0 for h in self.hours.values()): 

29 raise ValueError("response windows must be positive") 

30 if not 0 < self.premium_factor <= 1: 

31 raise ValueError("premium_factor must be greater than 0 and at most 1") 

32 ordered = [self.hours[p] for p in sorted(Priority, key=lambda p: p.rank)] 

33 if ordered != sorted(ordered, reverse=True): 

34 raise ValueError("a higher priority must never get a longer response window") 

35 

36 def window(self, priority: Priority, tier: Tier) -> timedelta: 

37 hours = self.hours[priority] 

38 if tier is Tier.PREMIUM: 

39 hours *= self.premium_factor 

40 return timedelta(hours=hours) 

41 

42 def respond_by(self, created_at: datetime, priority: Priority, tier: Tier) -> datetime: 

43 if created_at.tzinfo is None: 

44 raise ValueError("created_at must include a time zone") 

45 return created_at + self.window(priority, tier)