firehose and audit bucket fixture

Author: Karthikeyannhs

src/eligibility_signposting_api/config/config.py

@@ -21,6 +21,7 @@
 def config() -> dict[str, Any]:
     person_table_name = TableName(os.getenv("PERSON_TABLE_NAME", "test_eligibility_datastore"))
     rules_bucket_name = BucketName(os.getenv("RULES_BUCKET_NAME", "test-rules-bucket"))
+    audit_bucket_name = BucketName(os.getenv("AUDIT_BUCKET_NAME", "test-audit-bucket"))
     aws_default_region = AwsRegion(os.getenv("AWS_DEFAULT_REGION", "eu-west-1"))
     log_level = LOG_LEVEL
 
@@ -33,6 +34,8 @@ def config() -> dict[str, Any]:
             "person_table_name": person_table_name,
             "s3_endpoint": None,
             "rules_bucket_name": rules_bucket_name,
+            "audit_bucket_name": audit_bucket_name,
+            "firehose_endpoint": None,
             "log_level": log_level,
         }
 
@@ -44,6 +47,8 @@ def config() -> dict[str, Any]:
         "person_table_name": person_table_name,
         "s3_endpoint": URL(os.getenv("S3_ENDPOINT", "http://localhost:4566")),
         "rules_bucket_name": rules_bucket_name,
+        "audit_bucket_name": audit_bucket_name,
+        "firehose_endpoint": URL(os.getenv("FIREHOSE_ENDPOINT", "http://localhost:4566")),
         "log_level": log_level,
     }
 

src/eligibility_signposting_api/config/contants.py

@@ -1,3 +1,6 @@
 MAGIC_COHORT_LABEL = "elid_all_people"
 RULE_STOP_DEFAULT = False
-NHS_NUMBER_HEADER_NAME = "nhs-login-nhs-number"
+NHS_NUMBER_HEADER = "nhs-login-nhs-number"
+
+# Infra
+ELIGIBILITY_SIGNPOSTING_AUDIT_STREAM = "eligibility-signposting-audit-stream"

src/eligibility_signposting_api/repos/factory.py

@@ -35,3 +35,11 @@ def dynamodb_resource_factory(
 def s3_service_factory(session: Session, s3_endpoint: Annotated[URL, Inject(param="s3_endpoint")]) -> BaseClient:
     endpoint_url = str(s3_endpoint) if s3_endpoint is not None else None
     return session.client("s3", endpoint_url=endpoint_url)
+
+
+@service(qualifier="firehose")
+def firehose_client_factory(
+    session: Session, firehose_endpoint: Annotated[URL, Inject(param="firehose_endpoint")]
+) -> BaseClient:
+    endpoint_url = str(firehose_endpoint) if firehose_endpoint is not None else None
+    return session.client("firehose", endpoint_url=endpoint_url)

src/eligibility_signposting_api/services/audit_service.py

@@ -0,0 +1,34 @@
+import json
+import logging
+from typing import Annotated
+
+from botocore.client import BaseClient
+from wireup import Inject, service
+
+from eligibility_signposting_api.config.contants import ELIGIBILITY_SIGNPOSTING_AUDIT_STREAM
+
+logger = logging.getLogger(__name__)
+
+
+@service
+class AuditService:
+    def __init__(self, firehose: Annotated[BaseClient, Inject(qualifier="firehose")]) -> None:
+        super().__init__()
+        self.firehose = firehose
+        self.delivery_stream_name = ELIGIBILITY_SIGNPOSTING_AUDIT_STREAM
+
+    def audit(self, audit_record: dict) -> None:
+        """
+        Sends an audit record to the configured Firehose delivery stream.
+
+        Args:
+            audit_record (dict): The audit data to send.
+
+        Returns:
+            str: The Firehose record ID.
+        """
+        response = self.firehose.put_record(
+            DeliveryStreamName=self.delivery_stream_name,
+            Record={"Data": (json.dumps(audit_record) + "\n").encode("utf-8")},
+        )
+        logger.info("Successfully sent to the Firehose", extra={"firehose_record_id": response["RecordId"]})

src/eligibility_signposting_api/services/eligibility_services.py

@@ -4,6 +4,7 @@
 
 from eligibility_signposting_api.model import eligibility
 from eligibility_signposting_api.repos import CampaignRepo, NotFoundError, PersonRepo
+from eligibility_signposting_api.services.audit_service import AuditService
 from eligibility_signposting_api.services.calculators import eligibility_calculator as calculator
 
 logger = logging.getLogger(__name__)
@@ -23,11 +24,13 @@ def __init__(
         self,
         person_repo: PersonRepo,
         campaign_repo: CampaignRepo,
+        audit_service: AuditService,
         calculator_factory: calculator.EligibilityCalculatorFactory,
     ) -> None:
         super().__init__()
         self.person_repo = person_repo
         self.campaign_repo = campaign_repo
+        self.audit_service = audit_service
         self.calculator_factory = calculator_factory
 
     def get_eligibility_status(
@@ -51,6 +54,7 @@ def get_eligibility_status(
                 raise UnknownPersonError from e
             else:
                 calc: calculator.EligibilityCalculator = self.calculator_factory.get(person_data, campaign_configs)
+                self.audit_service.audit({"test_audit": "check if audit works"})
                 return calc.evaluate_eligibility(include_actions_flag=include_actions_flag)
 
         raise UnknownPersonError  # pragma: no cover

src/eligibility_signposting_api/wrapper.py

@@ -4,7 +4,7 @@
 
 from mangum.types import LambdaContext, LambdaEvent
 
-from eligibility_signposting_api.config.contants import NHS_NUMBER_HEADER_NAME
+from eligibility_signposting_api.config.contants import NHS_NUMBER_HEADER
 
 logger = logging.getLogger(__name__)
 
@@ -20,7 +20,7 @@ def wrapper(event: LambdaEvent, context: LambdaContext) -> dict[str, int | str]:
             headers = event.get("headers", {})
             path_params = event.get("pathParameters", {})
 
-            header_nhs = headers.get(NHS_NUMBER_HEADER_NAME)
+            header_nhs = headers.get(NHS_NUMBER_HEADER)
             path_nhs = path_params.get("id")
 
             logger.info("nhs numbers from the request", extra={"header_nhs": header_nhs, "path_nhs": path_nhs})

tests/docker-compose.yml

@@ -9,7 +9,7 @@ services:
       # LocalStack configuration: https://docs.localstack.cloud/references/configuration/
       - DEBUG=${LOCALSTACK_DEBUG:-0}
       - DEFAULT_REGION=${AWS_DEFAULT_REGION:-eu-west-1}
-      - LAMBDA_EXECUTOR=local
+      - LAMBDA_EXECUTOR=docker
     volumes:
       - "${LOCALSTACK_VOLUME_DIR:-../volume}:/var/lib/localstack"
       - "/var/run/docker.sock:/var/run/docker.sock"

tests/integration/conftest.py

@@ -97,6 +97,16 @@ def s3_client(boto3_session: Session, localstack: URL) -> BaseClient:
     return boto3_session.client("s3", endpoint_url=str(localstack))
 
 
+@pytest.fixture(scope="session")
+def firehose_client(boto3_session: Session, localstack: URL) -> BaseClient:
+    return boto3_session.client("firehose", endpoint_url=str(localstack))
+
+
+@pytest.fixture(autouse=True)
+def firehose_delivery_stream(firehose_client: BaseClient) -> dict[str, Any]:
+    return firehose_client.create_delivery_stream(DeliveryStreamName="eligibility-signposting-audit-stream")
+
+
 @pytest.fixture(scope="session")
 def iam_role(iam_client: BaseClient) -> Generator[str]:
     role_name = "LambdaExecutionRole"
@@ -190,6 +200,7 @@ def flask_function(lambda_client: BaseClient, iam_role: str, lambda_zip: Path) -
                 "Variables": {
                     "DYNAMODB_ENDPOINT": os.getenv("LOCALSTACK_INTERNAL_ENDPOINT", "http://localstack:4566/"),
                     "S3_ENDPOINT": os.getenv("LOCALSTACK_INTERNAL_ENDPOINT", "http://localstack:4566/"),
+                    "FIREHOSE_ENDPOINT": os.getenv("LOCALSTACK_INTERNAL_ENDPOINT", "http://localstack:4566/"),
                     "AWS_REGION": AWS_REGION,
                     "LOG_LEVEL": "DEBUG",
                 }
@@ -372,15 +383,23 @@ def persisted_person_pc_sw19(person_table: Any, faker: Faker) -> Generator[eligi
 
 
 @pytest.fixture(scope="session")
-def bucket(s3_client: BaseClient) -> Generator[BucketName]:
+def rules_bucket(s3_client: BaseClient) -> Generator[BucketName]:
     bucket_name = BucketName(os.getenv("RULES_BUCKET_NAME", "test-rules-bucket"))
     s3_client.create_bucket(Bucket=bucket_name, CreateBucketConfiguration={"LocationConstraint": AWS_REGION})
     yield bucket_name
     s3_client.delete_bucket(Bucket=bucket_name)
 
 
+@pytest.fixture(scope="session")
+def audit_bucket(s3_client: BaseClient) -> Generator[BucketName]:
+    bucket_name = BucketName(os.getenv("AUDIT_BUCKET_NAME", "test-audit-bucket"))
+    s3_client.create_bucket(Bucket=bucket_name, CreateBucketConfiguration={"LocationConstraint": AWS_REGION})
+    yield bucket_name
+    s3_client.delete_bucket(Bucket=bucket_name)
+
+
 @pytest.fixture(scope="class")
-def campaign_config(s3_client: BaseClient, bucket: BucketName) -> Generator[rules.CampaignConfig]:
+def campaign_config(s3_client: BaseClient, rules_bucket: BucketName) -> Generator[rules.CampaignConfig]:
     campaign: rules.CampaignConfig = rule.CampaignConfigFactory.build(
         target="RSV",
         iterations=[
@@ -402,14 +421,16 @@ def campaign_config(s3_client: BaseClient, bucket: BucketName) -> Generator[rule
     )
     campaign_data = {"CampaignConfig": campaign.model_dump(by_alias=True)}
     s3_client.put_object(
-        Bucket=bucket, Key=f"{campaign.name}.json", Body=json.dumps(campaign_data), ContentType="application/json"
+        Bucket=rules_bucket, Key=f"{campaign.name}.json", Body=json.dumps(campaign_data), ContentType="application/json"
     )
     yield campaign
-    s3_client.delete_object(Bucket=bucket, Key=f"{campaign.name}.json")
+    s3_client.delete_object(Bucket=rules_bucket, Key=f"{campaign.name}.json")
 
 
 @pytest.fixture(scope="class")
-def campaign_config_with_magic_cohort(s3_client: BaseClient, bucket: BucketName) -> Generator[rules.CampaignConfig]:
+def campaign_config_with_magic_cohort(
+    s3_client: BaseClient, rules_bucket: BucketName
+) -> Generator[rules.CampaignConfig]:
     campaign: rules.CampaignConfig = rule.CampaignConfigFactory.build(
         target="COVID",
         iterations=[
@@ -424,15 +445,15 @@ def campaign_config_with_magic_cohort(s3_client: BaseClient, bucket: BucketName)
     )
     campaign_data = {"CampaignConfig": campaign.model_dump(by_alias=True)}
     s3_client.put_object(
-        Bucket=bucket, Key=f"{campaign.name}.json", Body=json.dumps(campaign_data), ContentType="application/json"
+        Bucket=rules_bucket, Key=f"{campaign.name}.json", Body=json.dumps(campaign_data), ContentType="application/json"
     )
     yield campaign
-    s3_client.delete_object(Bucket=bucket, Key=f"{campaign.name}.json")
+    s3_client.delete_object(Bucket=rules_bucket, Key=f"{campaign.name}.json")
 
 
 @pytest.fixture(scope="class")
 def campaign_config_with_missing_descriptions_missing_rule_text(
-    s3_client: BaseClient, bucket: BucketName
+    s3_client: BaseClient, rules_bucket: BucketName
 ) -> Generator[rules.CampaignConfig]:
     campaign: rules.CampaignConfig = rule.CampaignConfigFactory.build(
         target="FLU",
@@ -456,7 +477,7 @@ def campaign_config_with_missing_descriptions_missing_rule_text(
     )
     campaign_data = {"CampaignConfig": campaign.model_dump(by_alias=True)}
     s3_client.put_object(
-        Bucket=bucket, Key=f"{campaign.name}.json", Body=json.dumps(campaign_data), ContentType="application/json"
+        Bucket=rules_bucket, Key=f"{campaign.name}.json", Body=json.dumps(campaign_data), ContentType="application/json"
     )
     yield campaign
-    s3_client.delete_object(Bucket=bucket, Key=f"{campaign.name}.json")
+    s3_client.delete_object(Bucket=rules_bucket, Key=f"{campaign.name}.json")

tests/integration/lambda/test_app_running_as_lambda.py

@@ -1,6 +1,7 @@
 import base64
 import json
 import logging
+import time
 from http import HTTPStatus
 
 import httpx
@@ -167,6 +168,8 @@ def test_given_nhs_number_in_path_matches_with_nhs_number_in_headers(
         timeout=10,
     )
 
+    time.sleep(40)
+
     # Then
     assert_that(
         response,

tests/integration/repo/test_campaign_repo.py

@@ -12,19 +12,19 @@
 
 
 @pytest.fixture(scope="module")
-def campaign_config(s3_client: BaseClient, bucket: BucketName) -> Generator[CampaignConfig]:
+def campaign_config(s3_client: BaseClient, rules_bucket: BucketName) -> Generator[CampaignConfig]:
     campaign: CampaignConfig = CampaignConfigFactory.build()
     campaign_data = {"CampaignConfig": campaign.model_dump(by_alias=True)}
     s3_client.put_object(
-        Bucket=bucket, Key=f"{campaign.name}.json", Body=json.dumps(campaign_data), ContentType="application/json"
+        Bucket=rules_bucket, Key=f"{campaign.name}.json", Body=json.dumps(campaign_data), ContentType="application/json"
     )
     yield campaign
-    s3_client.delete_object(Bucket=bucket, Key=f"{campaign.name}.json")
+    s3_client.delete_object(Bucket=rules_bucket, Key=f"{campaign.name}.json")
 
 
-def test_get_campaign_config(s3_client: BaseClient, bucket: BucketName, campaign_config: CampaignConfig):
+def test_get_campaign_config(s3_client: BaseClient, rules_bucket: BucketName, campaign_config: CampaignConfig):
     # Given
-    repo = CampaignRepo(s3_client, bucket)
+    repo = CampaignRepo(s3_client, rules_bucket)
 
     # When
     actual = list(repo.get_campaign_configs())