Coverage for backend/django/Economics/results/services/lifecycle/common.py: 84%

21 statements  

« prev     ^ index     » next       coverage.py v7.10.7, created at 2026-06-23 21:51 +0000

1"""Shared contracts and boundary helpers for Economics result lifecycle internals. 

2 

3This module is deliberately small: it owns serialization, rounding, timing, and 

4study-assumption lookup helpers that are reused across the lifecycle package. 

5It must not query result runs or materialize result rows, so higher-level 

6modules can depend on it without creating lifecycle import cycles. 

7""" 

8 

9from __future__ import annotations 

10 

11import time 

12from datetime import date, datetime 

13from typing import Any 

14 

15from pydantic import BaseModel, ConfigDict 

16 

17from Economics.settings_profiles.models import EconomicsSettingsProfile 

18 

19from Economics.studies.models import EconomicsStudy 

20from Economics.settings_profiles.services.settings_profiles import get_settings_profile 

21from Economics.shared.payloads import json_ready, warning_payload 

22 

23 

24class EconomicsContract(BaseModel): 

25 """Frozen Pydantic base used for lifecycle JSON-boundary contracts.""" 

26 

27 model_config = ConfigDict( 

28 frozen=True, 

29 arbitrary_types_allowed=True, 

30 ) 

31 

32 

33def get_assumptions(study: EconomicsStudy) -> EconomicsSettingsProfile | None: 

34 """Return the study's selected settings profile when configured.""" 

35 return get_settings_profile(study) 

36 

37 

38def version(value: Any) -> str: 

39 """Normalize source versions used in dependency audit rows.""" 

40 if value is None: 40 ↛ 41line 40 didn't jump to line 41 because the condition on line 40 was never true

41 return "" 

42 if isinstance(value, (datetime, date)): 42 ↛ 44line 42 didn't jump to line 44 because the condition on line 42 was always true

43 return value.isoformat() 

44 return str(value) 

45 

46 

47def duration_ms(started_at: float) -> int: 

48 """Return elapsed wall-clock time in milliseconds for diagnostic payloads.""" 

49 return int((time.perf_counter() - started_at) * 1000)