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,7 @@ 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,
             "log_level": log_level,
         }
 
@@ -44,6 +46,7 @@ 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,
         "log_level": log_level,
     }
 

tests/integration/conftest.py

@@ -96,6 +96,10 @@ def iam_client(boto3_session: Session, localstack: URL) -> BaseClient:
 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:
+    session = Session()
+    return session.client("firehose", endpoint_url=str(localstack))
 
 @pytest.fixture(scope="session")
 def iam_role(iam_client: BaseClient) -> Generator[str]:
@@ -372,15 +376,21 @@ 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 +412,15 @@ 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 +435,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 +467,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

@@ -155,7 +155,8 @@ def get_log_messages(flask_function: str, logs_client: BaseClient) -> list[str]:
 def test_given_nhs_number_in_path_matches_with_nhs_number_in_headers(
     lambda_client: BaseClient,  # noqa:ARG001
     persisted_person: NHSNumber,
-    campaign_config: CampaignConfig,  # noqa:ARG001
+    campaign_config: CampaignConfig, # noqa:ARG001
+    firehose_client: BaseClient, # noqa:ARG001
     api_gateway_endpoint: URL,
 ):
     # Given

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())