Fixed linting

Author: shweta-nhs

src/eligibility_signposting_api/audit_context.py

@@ -2,6 +2,7 @@
 from dataclasses import asdict
 from datetime import UTC, datetime
 from operator import attrgetter
+from uuid import UUID
 
 from flask import Request, g
 
@@ -36,7 +37,7 @@ class AuditContext:
     def add_request_details(request: Request) -> None:
         g.audit_log = AuditEvent()
         resource_id = None
-        if "nhs_number" in request.view_args:
+        if request.view_args and request.view_args["nhs_number"]:
             resource_id = request.view_args["nhs_number"]
         g.audit_log.request = RequestAuditData(
             nhs_number=resource_id,
@@ -62,7 +63,7 @@ def add_request_details(request: Request) -> None:
     def append_audit_condition(
         suggested_actions: SuggestedActions | None,
         condition_name: ConditionName,
-        best_results: tuple[Iteration, IterationResult, dict[str, CohortGroupResult]],
+        best_results: tuple[Iteration | None, IterationResult | None, dict[str, CohortGroupResult] | None],
         campaign_details: tuple[CampaignID | None, CampaignVersion | None],
         redirect_rule_details: tuple[RulePriority | None, RuleName | None],
     ) -> None:
@@ -72,35 +73,37 @@ def append_audit_condition(
         best_candidate = best_results[1]
         best_cohort_results = best_results[2]
 
-        for value in sorted(best_cohort_results.values(), key=attrgetter("cohort_code")):
-            audit_eligibility_cohorts.append(
-                AuditEligibilityCohorts(cohort_code=value.cohort_code, cohort_status=value.status.name)
-            )
-
-            audit_eligibility_cohort_groups.append(
-                AuditEligibilityCohortGroups(
-                    cohort_code=value.cohort_code, cohort_status=value.status.name, cohort_text=value.description
+        if best_cohort_results:
+            for value in sorted(best_cohort_results.values(), key=attrgetter("cohort_code")):
+                cohort_status_name = value.status.name if value.status else None
+                audit_eligibility_cohorts.append(
+                    AuditEligibilityCohorts(cohort_code=value.cohort_code, cohort_status=cohort_status_name)
                 )
-            )
 
-            if value.audit_rules:
-                if best_candidate.status.name == Status.not_eligible.name:
-                    audit_filter_rule = AuditFilterRule(
-                        rule_priority=int(value.audit_rules[0].rule_priority), rule_name=value.audit_rules[0].rule_name
-                    )
-                if best_candidate.status.name == Status.not_actionable.name:
-                    audit_suitability_rule = AuditSuitabilityRule(
-                        rule_priority=int(value.audit_rules[0].rule_priority),
-                        rule_name=value.audit_rules[0].rule_name,
-                        rule_message=value.audit_rules[0].rule_description,
+                audit_eligibility_cohort_groups.append(
+                    AuditEligibilityCohortGroups(
+                        cohort_code=value.cohort_code, cohort_status=cohort_status_name, cohort_text=value.description
                     )
+                )
 
-        if best_candidate.status.name == Status.actionable.name:
+                if value.audit_rules and best_candidate:  # Check best_candidate here as well
+                    if best_candidate.status and best_candidate.status.name == Status.not_eligible.name:
+                        audit_filter_rule = AuditFilterRule(
+                            rule_priority=int(value.audit_rules[0].rule_priority),
+                            rule_name=value.audit_rules[0].rule_name,
+                        )
+                    if best_candidate.status and best_candidate.status.name == Status.not_actionable.name:
+                        audit_suitability_rule = AuditSuitabilityRule(
+                            rule_priority=int(value.audit_rules[0].rule_priority),
+                            rule_name=value.audit_rules[0].rule_name,
+                            rule_message=value.audit_rules[0].rule_description,
+                        )
+
+        if best_candidate and best_candidate.status and best_candidate.status.name == Status.actionable.name:
             audit_redirect_rule = AuditRedirectRule(redirect_rule_details[0], redirect_rule_details[1])
 
-        if suggested_actions is None or suggested_actions == []:
-            audit_actions = suggested_actions
-
+        if suggested_actions is None:
+            audit_actions = None
         elif len(suggested_actions.actions) > 0:
             for action in suggested_actions.actions:
                 audit_actions.append(
@@ -116,11 +119,11 @@ def append_audit_condition(
         audit_condition = AuditCondition(
             campaign_id=campaign_details[0],
             campaign_version=campaign_details[1],
-            iteration_id=best_active_iteration.id,
-            iteration_version=best_active_iteration.version,
+            iteration_id=best_active_iteration.id if best_active_iteration else None,
+            iteration_version=best_active_iteration.version if best_active_iteration else None,
             condition_name=condition_name,
-            status=best_candidate.status.name,
-            status_text=best_candidate.status.name,
+            status=best_candidate.status.name if best_candidate and best_candidate.status else None,
+            status_text=best_candidate.status.name if best_candidate and best_candidate.status else None,
             eligibility_cohorts=audit_eligibility_cohorts,
             eligibility_cohort_groups=audit_eligibility_cohort_groups,
             filter_rules=audit_filter_rule,
@@ -132,11 +135,10 @@ def append_audit_condition(
         g.audit_log.response.condition.append(audit_condition)
 
     @staticmethod
-    def add_response_details(response) -> None:
-        g.audit_log.response.response_id = response.response_id
-        g.audit_log.response.last_updated = response.meta.last_updated
+    def add_response_details(response_id: UUID, last_updated: datetime) -> None:
+        g.audit_log.response.response_id = response_id
+        g.audit_log.response.last_updated = last_updated
 
     @staticmethod
-    def write_to_firehose(service: AuditService, response) -> None:
-        AuditContext.add_response_details(response)
+    def write_to_firehose(service: AuditService) -> None:
         service.audit(asdict(g.audit_log))

src/eligibility_signposting_api/audit_models.py

@@ -5,10 +5,10 @@
 
 @dataclass
 class RequestAuditHeader:
-    x_request_id: str = None
-    x_correlation_id: str = None
-    nhsd_end_user_organisation_ods: str = None
-    nhsd_application_id: str = None
+    x_request_id: str | None = None
+    x_correlation_id: str | None = None
+    nhsd_end_user_organisation_ods: str | None = None
+    nhsd_application_id: str | None = None
 
 
 @dataclass
@@ -70,19 +70,19 @@ class AuditAction:
 
 @dataclass
 class AuditCondition:
-    campaign_id: str = None
-    campaign_version: str = None
-    iteration_id: str = None
-    iteration_version: str = None
-    condition_name: str = None
-    status: str = None
-    status_text: str = None
+    campaign_id: str | None = None
+    campaign_version: str | None = None
+    iteration_id: str | None = None
+    iteration_version: str | None = None
+    condition_name: str | None = None
+    status: str | None = None
+    status_text: str | None = None
     eligibility_cohorts: list[AuditEligibilityCohorts] | None = None
     eligibility_cohort_groups: list[AuditEligibilityCohortGroups] | None = None
     filter_rules: AuditFilterRule | None = None
     suitability_rules: AuditSuitabilityRule | None = None
     action_rule: AuditRedirectRule | None = None
-    actions: list[AuditAction] = field(default_factory=list)
+    actions: list[AuditAction] | None = field(default_factory=list)
 
 
 @dataclass

src/eligibility_signposting_api/services/calculators/eligibility_calculator.py

@@ -144,6 +144,12 @@ def evaluate_eligibility(self, *, include_actions_flag: bool = True) -> eligibil
         redirect_rule_priority, redirect_rule_name = None, None
 
         for condition_name, campaign_group in self.campaigns_grouped_by_condition_name:
+            best_active_iteration: Iteration | None
+            best_candidate: IterationResult
+            best_campaign_id: CampaignID | None
+            best_campaign_version: CampaignVersion | None
+            best_cohort_results: dict[str, CohortGroupResult] | None
+
             iteration_results: dict[
                 str, tuple[Iteration, IterationResult, CampaignID, CampaignVersion, dict[str, CohortGroupResult]]
             ] = {}

src/eligibility_signposting_api/services/calculators/rule_calculator.py

@@ -24,7 +24,7 @@ def evaluate_exclusion(self) -> tuple[eligibility.Status, eligibility.Reason]:
         reason = eligibility.Reason(
             rule_name=eligibility.RuleName(self.rule.name),
             rule_type=eligibility.RuleType(self.rule.type),
-            rule_priority=eligibility.RulePriority(self.rule.priority),
+            rule_priority=eligibility.RulePriority(str(self.rule.priority)),
             rule_description=eligibility.RuleDescription(self.rule.description),
             matcher_matched=matcher_matched,
         )

src/eligibility_signposting_api/views/eligibility.py

@@ -49,7 +49,7 @@ def check_eligibility(
         return handle_unknown_person_error(nhs_number)
     else:
         eligibility_response = build_eligibility_response(eligibility_status)
-        AuditContext.write_to_firehose(audit_service, eligibility_response)
+        AuditContext.write_to_firehose(audit_service)
         return make_response(
             eligibility_response.model_dump(by_alias=True, mode="json", exclude_none=True), HTTPStatus.OK
         )
@@ -122,9 +122,14 @@ def build_eligibility_response(eligibility_status: EligibilityStatus) -> eligibi
 
         processed_suggestions.append(suggestions)
 
+    response_id = uuid.uuid4()
+    updated = eligibility.LastUpdated(datetime.now(tz=UTC))
+
+    AuditContext.add_response_details(response_id, updated)
+
     return eligibility.EligibilityResponse(  # pyright: ignore[reportCallIssue]
-        responseId=uuid.uuid4(),  # pyright: ignore[reportCallIssue]
-        meta=eligibility.Meta(lastUpdated=eligibility.LastUpdated(datetime.now(tz=UTC))),
+        responseId=response_id,  # pyright: ignore[reportCallIssue]
+        meta=eligibility.Meta(lastUpdated=updated),
         # pyright: ignore[reportCallIssue]
         processedSuggestions=processed_suggestions,
     )

tests/unit/test_audit_context.py

@@ -28,7 +28,6 @@
 from eligibility_signposting_api.model.rules import CampaignID, CampaignVersion, Iteration, RuleType
 from eligibility_signposting_api.services.audit_service import AuditService
 from tests.fixtures.builders.model.rule import IterationFactory
-from tests.fixtures.builders.views.response_model.eligibility import EligibilityResponseFactory
 
 
 @pytest.fixture
@@ -141,35 +140,30 @@ def test_append_audit_condition_adds_condition_to_audit_log_on_g(app):
 
 
 def test_add_response_details_adds_to_audit_log_on_g(app):
-    eligibility_response = EligibilityResponseFactory.build(
-        response_id=uuid.uuid4(),
-        meta={"last_updated": datetime(2023, 1, 1, 0, 0, tzinfo=UTC)},
-        processed_suggestions=[],
-    )
+    response_id = uuid.uuid4()
+    last_updated = datetime(2023, 1, 1, 0, 0, tzinfo=UTC)
 
     with app.app_context():
         g.audit_log = AuditEvent()
 
-        AuditContext.add_response_details(eligibility_response)
+        AuditContext.add_response_details(response_id, last_updated)
 
-        assert g.audit_log.response.response_id == eligibility_response.response_id
-        assert g.audit_log.response.last_updated is eligibility_response.meta.last_updated
+        assert g.audit_log.response.response_id == response_id
+        assert g.audit_log.response.last_updated is last_updated
 
 
 def test_write_to_firehose_calls_audit_service_with_correct_data_from_g(app):
     mock_audit_service = Mock(spec=AuditService)
-    eligibility_response = EligibilityResponseFactory.build(
-        response_id=(uuid.uuid4()),
-        meta={"last_updated": (datetime(2023, 1, 1, 0, 0, tzinfo=UTC))},
-        processed_suggestions=[],
-    )
+    response_id = uuid.uuid4()
+    last_updated = datetime(2023, 1, 1, 0, 0, tzinfo=UTC)
 
     with app.app_context():
         g.audit_log = AuditEvent()
 
-        AuditContext.write_to_firehose(mock_audit_service, eligibility_response)
+        AuditContext.add_response_details(response_id, last_updated)
+        AuditContext.write_to_firehose(mock_audit_service)
 
-        assert g.audit_log.response.response_id == eligibility_response.response_id
-        assert g.audit_log.response.last_updated == eligibility_response.meta.last_updated
+        assert g.audit_log.response.response_id == response_id
+        assert g.audit_log.response.last_updated == last_updated
 
         mock_audit_service.audit.assert_called_once_with(asdict(g.audit_log))