Coverage for backend/django/Economics/results/services/comparison/results.py: 88%

1"""Current-result lookup for Economics study comparisons.""" 

2 

3from __future__ import annotations 

4 

5from authentication.user.models import User 

6from core.managers import has_flowsheet_read_access 

7from core.validation import flowsheet_context 

8from django.core.exceptions import ObjectDoesNotExist 

9 

10from Economics.results.models import EconomicsResultLine, EconomicsResultRun 

11from Economics.results.services.lifecycle.runs import classify_result_run 

12from Economics.results.services.resource_metrics import resource_metric_values_for_lines 

13from Economics.shared.choices import ResultRunStatus 

14from Economics.studies.models import EconomicsStudy 

15from Economics.studies.services.baseline_access import stable_flowsheet_id 

16 

17from .contracts import ComparisonMetricValue, ResultContext 

18 

19 

20def result_context(*, study: EconomicsStudy, user: User) -> ResultContext | None: 

21 """Load financial metric result lines after checking access to the study flowsheet.""" 

22 flowsheet_id = stable_flowsheet_id(study) 

23 if not has_flowsheet_read_access(user, flowsheet_id): 23 ↛ 24line 23 didn't jump to line 24 because the condition on line 23 was never true

24 return None 

25 with flowsheet_context(flowsheet_id, user): 

26 return _result_context_for_accessible_study(study) 

27 

28 

29def _result_context_for_accessible_study(study: EconomicsStudy) -> ResultContext | None: 

30 """Load the current result run for a study in its own authorized flowsheet context.""" 

31 run = ( 

32 EconomicsResultRun.objects.filter( 

33 flowsheet_state=study.flowsheet_state, 

34 study=study, 

35 status=ResultRunStatus.CURRENT, 

36 ) 

37 .order_by("-created_at", "-pk") 

38 .first() 

39 ) 

40 if run is None: 

41 return None 

42 financial_lines = { 

43 line.row_key: ComparisonMetricValue(amount=line.amount, unit=line.unit) 

44 for line in EconomicsResultLine.objects.filter( 

45 flowsheet_state=study.flowsheet_state, 

46 result_run=run, 

47 group="financial_metrics", 

48 ) 

49 } 

50 operating_lines = list( 

51 EconomicsResultLine.objects.filter( 

52 flowsheet_state=study.flowsheet_state, 

53 result_run=run, 

54 group="operating_lines", 

55 ) 

56 ) 

57 return ResultContext( 

58 run=run, 

59 classification=_classification(run), 

60 lines={**financial_lines, **resource_metric_values_for_lines(operating_lines)}, 

61 ) 

62 

63 

64def _classification(run: EconomicsResultRun) -> str: 

65 try: 

66 return classify_result_run(run) 

67 except (ValueError, ObjectDoesNotExist): 

68 return run.status