updated config to pick cred&URL by env

Author: Karthikeyannhs

src/eligibility_signposting_api/config.py

@@ -19,14 +19,34 @@
 
 @cache
 def config() -> dict[str, Any]:
+    env = os.getenv("ENV")
+
+    eligibility_table_name = TableName(os.getenv("ELIGIBILITY_TABLE_NAME", "test_eligibility_datastore"))
+    rules_bucket_name = BucketName(os.getenv("RULES_BUCKET_NAME", "test-rules-bucket"))
+    aws_default_region = AwsRegion(os.getenv("AWS_DEFAULT_REGION", "eu-west-1"))
+    log_level = LOG_LEVEL
+
+    if env:
+        return {
+            "aws_access_key_id": None,
+            "aws_default_region": aws_default_region,
+            "aws_secret_access_key": None,
+            "dynamodb_endpoint": None,
+            "eligibility_table_name": eligibility_table_name,
+            "s3_endpoint": None,
+            "rules_bucket_name": rules_bucket_name,
+            "log_level": log_level,
+        }
+
     return {
         "aws_access_key_id": AwsAccessKey(os.getenv("AWS_ACCESS_KEY_ID", "dummy_key")),
-        "aws_default_region": AwsRegion(os.getenv("AWS_DEFAULT_REGION", "eu-west-1")),
+        "aws_default_region": aws_default_region,
         "aws_secret_access_key": AwsSecretAccessKey(os.getenv("AWS_SECRET_ACCESS_KEY", "dummy_secret")),
         "dynamodb_endpoint": URL(os.getenv("DYNAMODB_ENDPOINT", "http://localhost:4566")),
-        "eligibility_table_name": TableName(os.getenv("ELIGIBILITY_TABLE_NAME", "test_eligibility_datastore")),
-        "log_level": LOG_LEVEL,
-        "rules_bucket_name": BucketName(os.getenv("RULES_BUCKET_NAME", "test-rules-bucket")),
+        "eligibility_table_name": eligibility_table_name,
+        "s3_endpoint": URL(os.getenv("S3_ENDPOINT", "http://localhost:4566")),
+        "rules_bucket_name": rules_bucket_name,
+        "log_level": log_level,
     }
 
 

src/eligibility_signposting_api/repos/factory.py

@@ -27,11 +27,11 @@ def boto3_session_factory(
 def dynamodb_resource_factory(
     session: Session, dynamodb_endpoint: Annotated[URL, Inject(param="dynamodb_endpoint")]
 ) -> ServiceResource:
-    return session.resource("dynamodb", endpoint_url=str(dynamodb_endpoint))
+    endpoint_url = str(dynamodb_endpoint) if dynamodb_endpoint is not None else None
+    return session.resource("dynamodb", endpoint_url=endpoint_url)
 
 
 @service(qualifier="s3")
-def s3_service_factory(
-    session: Session, dynamodb_endpoint: Annotated[URL, Inject(param="dynamodb_endpoint")]
-) -> BaseClient:
-    return session.client("s3", endpoint_url=str(dynamodb_endpoint))
+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)

tests/integration/conftest.py

@@ -164,6 +164,7 @@ def flask_function(lambda_client: BaseClient, iam_role: str, lambda_zip: Path) -
             Environment={
                 "Variables": {
                     "DYNAMODB_ENDPOINT": os.getenv("LOCALSTACK_INTERNAL_ENDPOINT", "http://localstack:4566/"),
+                    "S3_ENDPOINT": os.getenv("LOCALSTACK_INTERNAL_ENDPOINT", "http://localstack:4566/"),
                     "AWS_REGION": AWS_REGION,
                     "LOG_LEVEL": "DEBUG",
                 }

tests/unit/repos/__init__.py

@@ -0,0 +1,78 @@
+from unittest.mock import Mock
+
+import pytest
+from yarl import URL
+
+
+def service(qualifier=None):  # noqa: ARG001
+    def decorator(fn):
+        return fn
+
+    return decorator
+
+
+class Inject:
+    def __init__(self, param):
+        self.param = param
+
+
+@service(qualifier="dynamodb")
+def dynamodb_resource_factory(session, dynamodb_endpoint: URL):
+    endpoint_url = str(dynamodb_endpoint) if dynamodb_endpoint is not None else None
+    return session.resource("dynamodb", endpoint_url=endpoint_url)
+
+
+@service(qualifier="s3")
+def s3_service_factory(session, s3_endpoint: URL):
+    endpoint_url = str(s3_endpoint) if s3_endpoint is not None else None
+    return session.client("s3", endpoint_url=endpoint_url)
+
+
+# --- Pytest Unit Tests ---
+
+
+@pytest.fixture
+def mock_session():
+    return Mock()
+
+
+def test_dynamodb_resource_factory_with_endpoint(mock_session):
+    mock_resource = Mock()
+    mock_session.resource.return_value = mock_resource
+    endpoint = URL("http://localhost:4566")
+
+    result = dynamodb_resource_factory(mock_session, endpoint)
+
+    mock_session.resource.assert_called_once_with("dynamodb", endpoint_url="http://localhost:4566")
+    assert result is mock_resource
+
+
+def test_dynamodb_resource_factory_without_endpoint(mock_session):
+    mock_resource = Mock()
+    mock_session.resource.return_value = mock_resource
+
+    result = dynamodb_resource_factory(mock_session, None)
+
+    mock_session.resource.assert_called_once_with("dynamodb", endpoint_url=None)
+    assert result is mock_resource
+
+
+def test_s3_service_factory_with_endpoint(mock_session):
+    mock_client = Mock()
+    mock_session.client.return_value = mock_client
+    endpoint = URL("http://localhost:4566")
+
+    result = s3_service_factory(mock_session, endpoint)
+
+    mock_session.client.assert_called_once_with("s3", endpoint_url="http://localhost:4566")
+    assert result is mock_client
+
+
+def test_s3_service_factory_without_endpoint(mock_session):
+    mock_client = Mock()
+    mock_session.client.return_value = mock_client
+
+    result = s3_service_factory(mock_session, None)
+
+    mock_session.client.assert_called_once_with("s3", endpoint_url=None)
+    assert result is mock_client

tests/unit/repos/test_factory.py

@@ -0,0 +1,15 @@
+from eligibility_signposting_api.config import AwsRegion, config
+from eligibility_signposting_api.repos.eligibility_repo import TableName
+from eligibility_signposting_api.repos.rules_repo import BucketName
+
+
+def test_config_with_env_variable(monkeypatch):
+    monkeypatch.setenv("ENV", "PROD")
+
+    config_data = config()
+
+    assert config_data["aws_access_key_id"] is None
+    assert config_data["aws_secret_access_key"] is None
+    assert config_data["aws_default_region"] == AwsRegion("eu-west-1")
+    assert config_data["eligibility_table_name"] == TableName("test_eligibility_datastore")
+    assert config_data["rules_bucket_name"] == BucketName("test-rules-bucket")