Added unit tests for AuditContext

Author: shweta-nhs

src/eligibility_signposting_api/audit_context.py

@@ -132,8 +132,8 @@ def append_audit_condition(
 
     @staticmethod
     def add_response_details(response) -> None:
-        g.audit_log.response.response_id = str(response.response_id)
-        g.audit_log.response.last_updated = str(response.meta.last_updated)
+        g.audit_log.response.response_id = response.response_id
+        g.audit_log.response.last_updated = response.meta.last_updated
 
     @staticmethod
     def write_to_firehose(service: AuditService, response) -> None:

src/eligibility_signposting_api/audit_models.py

@@ -1,5 +1,6 @@
 from dataclasses import dataclass, field
 from datetime import datetime
+from uuid import UUID
 
 
 @dataclass
@@ -19,7 +20,7 @@ class RequestAuditQueryParams:
 
 @dataclass
 class RequestAuditData:
-    request_timestamp: datetime = field(default_factory=datetime.utcnow)
+    request_timestamp: datetime = field(default_factory=datetime.utcnow) # TODO: fix use of deprecated datetime.now()
     headers: RequestAuditHeader = field(default_factory=RequestAuditHeader)
     query_params: RequestAuditQueryParams = field(default_factory=RequestAuditQueryParams)
     nhs_number: str | None = None
@@ -86,7 +87,7 @@ class AuditCondition:
 
 @dataclass
 class ResponseAuditData:
-    response_id: str | None = None
+    response_id: UUID | None = None
     last_updated: str | None = None
     condition: list[AuditCondition] = field(default_factory=list)
 

tests/integration/lambda/test_app_running_as_lambda.py

@@ -1,7 +1,6 @@
 import base64
 import json
 import logging
-import time
 from http import HTTPStatus
 
 import httpx

tests/unit/test_audit_context.py

@@ -0,0 +1,128 @@
+import uuid
+import pytest
+from datetime import datetime
+
+from flask import g, request, Flask
+
+from eligibility_signposting_api.audit_context import AuditContext
+from eligibility_signposting_api.audit_models import AuditEvent
+from eligibility_signposting_api.model.eligibility import SuggestedActions, ConditionName, IterationResult, \
+    CohortGroupResult, SuggestedAction, ActionCode, ActionType, ActionDescription, UrlLink, UrlLabel, Reason, RuleName, \
+    RuleDescription, RulePriority, Status
+from eligibility_signposting_api.model.rules import Iteration, CampaignID, CampaignVersion, \
+    RuleType
+from fixtures.builders.model.rule import IterationFactory
+from fixtures.builders.views.response_model.eligibility import EligibilityResponseFactory
+
+
+@pytest.fixture
+def app():
+    app = Flask(__name__)
+    return app
+
+
+def test_add_request_details_sets_audit_log_on_g(app):
+    headers = {
+        "X-Request-ID": "test-x-request-id",
+        "X-Correlation-ID": "test-x-correlation-id",
+        "NHSD-End-User-Organisation-ODS": "test-org",
+        "nhsd-application-id": "test-app-id",
+    }
+
+    nhs_number = "1234567890"
+    url = "/patient-check?includeActions=Y"
+
+    with app.test_request_context(url, headers=headers, method="GET"):
+        request.view_args = {'nhs_number': nhs_number}
+        AuditContext.add_request_details(request)
+
+        assert hasattr(g, "audit_log")
+        audit_req = g.audit_log.request
+        assert audit_req.nhs_number == nhs_number
+        assert audit_req.headers.x_request_id == "test-x-request-id"
+        assert audit_req.headers.x_correlation_id == "test-x-correlation-id"
+        assert audit_req.headers.nhsd_end_user_organisation_ods == "test-org"
+        assert audit_req.headers.nhsd_application_id == "test-app-id"
+        assert audit_req.query_params.include_actions == "Y"
+        assert isinstance(audit_req.request_timestamp, datetime)
+
+
+def test_add_request_details_when_headers_are_empty_sets_audit_log_on_g(app):
+    nhs_number = "1234567890"
+    url = "/patient-check?includeActions=Y"
+
+    with app.test_request_context(url, method="GET"):
+        request.view_args = {'nhs_number': nhs_number}
+        AuditContext.add_request_details(request)
+
+        assert hasattr(g, "audit_log")
+        audit_req = g.audit_log.request
+        assert audit_req.nhs_number == nhs_number
+        assert audit_req.headers.x_request_id is None
+        assert audit_req.headers.x_correlation_id is None
+        assert audit_req.headers.nhsd_end_user_organisation_ods is None
+        assert audit_req.headers.nhsd_application_id is None
+        assert audit_req.query_params.include_actions == "Y"
+        assert isinstance(audit_req.request_timestamp, datetime)
+
+
+def test_append_audit_condition_adds_condition_to_audit_log(app):
+    suggested_actions: SuggestedActions | None
+    condition_name: ConditionName
+    best_results: tuple[Iteration, IterationResult, dict[str, CohortGroupResult]]
+    campaign_details: tuple[CampaignID | None, CampaignVersion | None]
+    redirect_rule_details: tuple[RulePriority | None, RuleName | None]
+
+    suggested_actions = SuggestedActions(actions=[SuggestedAction(
+        action_code=ActionCode("ActionCode1"),
+        action_type=ActionType("ActionType1"),
+        action_description=ActionDescription("ActionDescription1"),
+        url_link=UrlLink("https://www.example.com"),
+        url_label=UrlLabel("ActionLabel1")
+    )])
+
+    condition_name = ConditionName("Condition1")
+    iteration = IterationFactory.build()
+    audit_rules = [Reason(
+        rule_type=RuleType.filter,
+        rule_name=RuleName("FilterRuleName1"),
+        rule_description=RuleDescription("FilterRuleDescription1"),
+        matcher_matched=True,
+        rule_priority=RulePriority("1")
+    )]
+    cohort_group_result = CohortGroupResult(status=Status.actionable, cohort_code="CohortCode1",
+                                            description="CohortDescription1", audit_rules=audit_rules,
+                                            reasons=audit_rules)
+    iteration_result = IterationResult(status=Status.actionable, cohort_results=[cohort_group_result],
+                                       actions=suggested_actions)
+    best_results = (iteration, iteration_result, {"CohortCode1": cohort_group_result})
+    campaign_details = (CampaignID("CampaignID1"), CampaignVersion("CampaignVersion1"))
+    redirect_rule_details = (RulePriority("1"), RuleName("RedirectRuleName1"))
+
+    with app.app_context():
+        g.audit_log = AuditEvent()
+
+        AuditContext.append_audit_condition(suggested_actions, condition_name, best_results, campaign_details,
+                                            redirect_rule_details)
+
+        assert g.audit_log.response.condition, condition_name
+        cond = g.audit_log.response.condition[0]
+        assert cond.condition_name == condition_name
+        assert cond.campaign_id == campaign_details[0]
+        assert cond.status == best_results[1].status.name
+        assert cond.status_text == best_results[1].status.name
+
+def test_add_response_details_adds_to_audit_log(app):
+    eligibility_response = EligibilityResponseFactory.build(response_id=uuid.uuid4(),
+                                                            meta={"last_updated": datetime(2023, 1, 1, 0, 0)},
+                                                            processed_suggestions=[])
+
+    with app.app_context():
+        g.audit_log = AuditEvent()
+
+        AuditContext.add_response_details(eligibility_response)
+
+        assert g.audit_log.response.response_id == eligibility_response.response_id
+        assert g.audit_log.response.last_updated is eligibility_response.meta.last_updated
+
+