Implement settings validation and boundary checking

cwms_batch_events/api/main.py

@@ -11,7 +11,7 @@
     stream=sys.stdout,
 )
 
-app = FastAPI(root_path=settings.root_path)
+app = FastAPI(root_path=settings.fastapi_root_path)
 
 
 origins = r"http://localhost(:\d+)?"

cwms_batch_events/core/auth/user/jwt.py

@@ -21,6 +21,9 @@
 
 
 def get_public_pem():
+    if not settings.auth_environment:
+        raise ValueError("AUTH_ENVIRONMENT is not configured")
+
     try:
         public_key = PUBLIC_KEY[settings.auth_environment]
     except KeyError as e:
@@ -53,6 +56,9 @@ def verify_jwt_by_api(token: str) -> dict:
 
 
 def verify_jwt_by_saved_key(token: str) -> dict:
+    if settings.auth_environment not in ISSUER:
+        raise ValueError(f"Invalid AUTH_ENVIRONMENT: '{settings.auth_environment}'. Must be one of {list(ISSUER.keys())}")
+
     key = get_public_pem()
     payload = jwt.decode(
         token,

cwms_batch_events/core/job_logger/s3.py

@@ -3,23 +3,26 @@
 
 from cwms_batch_events.core.settings import settings
 
-S3_ENDPOINT_URL = settings.s3_endpoint_url
-S3_BUCKET = settings.s3_bucket
-
 
 class S3JobLogger:
     def __init__(self):
+        self.s3_bucket = settings.s3_bucket
+        self.s3_endpoint_url = settings.s3_endpoint_url
+
+        if not self.s3_bucket:
+            raise ValueError("S3_BUCKET must be configured for S3 logging")
+
         self.s3 = boto3.client(
             "s3",
-            endpoint_url=S3_ENDPOINT_URL,
+            endpoint_url=self.s3_endpoint_url,
         )
 
     def get_logs_for_job(self, job_id: UUID) -> str:
         key = f"logs/{job_id}.log"
-        response = self.s3.get_object(Bucket=S3_BUCKET, Key=key)
+        response = self.s3.get_object(Bucket=self.s3_bucket, Key=key)
         body: str = response["Body"].read().decode("utf-8")
         return body
 
     def push_logs_for_job(self, job_id: UUID, logs: str) -> None:
         key = f"logs/{job_id}.log"
-        self.s3.put_object(Bucket=S3_BUCKET, Key=key, Body=logs.encode("utf-8"))
+        self.s3.put_object(Bucket=self.s3_bucket, Key=key, Body=logs.encode("utf-8"))

cwms_batch_events/core/settings.py

@@ -1,28 +1,62 @@
 from functools import lru_cache
+from pydantic import model_validator
 from pydantic_settings import BaseSettings
 
 
 class Settings(BaseSettings):
-    app_key: str = ""
-    auth_environment: str = ""
-    auth_host: str = "http://traefik/auth"
+    app_key: str
+    auth_environment: str | None = None
+    auth_host: str | None = None
     auth_realm: str = "cwms"
     aws_access_key_id: str | None = None
     aws_secret_access_key: str | None = None
     aws_default_region: str | None = None
-    cda_api_root: str = "http://traefik/cwms-data/"
+    cda_api_root: str | None = None
     default_job_runner: str = "batch"
-    dynamodb_host: str = "http://dynamodb:9010"
-    pguser: str = ""
-    pgpassword: str = ""
-    pgdatabase: str = "postgres"
-    pghost: str = "db"
+    dynamodb_host: str | None = None
+    pguser: str | None = None
+    pgpassword: str | None = None
+    pgdatabase: str | None = None
+    pghost: str | None = None
     mock_user: bool = False
-    root_path: str = ""
-    s3_bucket: str = ""
+    root_path: str | None = None
+    s3_bucket: str | None = None
     s3_endpoint_url: str | None = None
     sqs_endpoint_url: str | None = None
 
+    @model_validator(mode="before")
+    @classmethod
+    def normalize_empty_strings(cls, values):
+        if isinstance(values, dict):
+            return {
+                name: value
+                for name, value in values.items()
+                if not (isinstance(value, str) and value.strip() == "")
+            }
+        return values
+
+    @model_validator(mode="after")
+    def validate_boundaries(self):
+        if not self.mock_user and not self.auth_environment:
+            raise ValueError("AUTH_ENVIRONMENT must be configured when MOCK_USER is false")
+
+        if not self.mock_user and not self.cda_api_root:
+            raise ValueError("CDA_API_ROOT must be configured when MOCK_USER is false")
+
+        if self.default_job_runner == "docker-local":
+            if not self.s3_bucket:
+                raise ValueError("S3_BUCKET must be configured when DEFAULT_JOB_RUNNER is docker-local")
+            if not self.s3_endpoint_url:
+                raise ValueError("S3_ENDPOINT_URL must be configured when DEFAULT_JOB_RUNNER is docker-local")
+            if not self.sqs_endpoint_url:
+                raise ValueError("SQS_ENDPOINT_URL must be configured when DEFAULT_JOB_RUNNER is docker-local")
+
+        return self
+
+    @property
+    def fastapi_root_path(self) -> str:
+        return self.root_path or ""
+
 
 @lru_cache
 def get_settings():

tests/conftest.py

@@ -11,3 +11,6 @@
 os.environ.setdefault("ALB_DNS_NAME", "http://events")
 os.environ.setdefault("APP_SECRETS_ARN", "arn:aws:secretsmanager:test")
 os.environ.setdefault("AWS_DEFAULT_REGION", "us-gov-west-1")
+os.environ.setdefault("APP_KEY", "test")
+os.environ.setdefault("AUTH_ENVIRONMENT", "TEST")
+os.environ.setdefault("CDA_API_ROOT", "http://localhost")

tests/cwms_batch_events/core/auth/user/test_jwt.py

@@ -35,6 +35,14 @@ def test_get_public_pem_raises_for_unknown_environment():
             get_public_pem()
 
 
+def test_get_public_pem_raises_when_auth_environment_missing():
+    with mock.patch(
+        "cwms_batch_events.core.auth.user.jwt.settings.auth_environment", None
+    ):
+        with pytest.raises(ValueError, match="AUTH_ENVIRONMENT is not configured"):
+            get_public_pem()
+
+
 def test_verify_jwt_routes_local_to_api_verification():
     with mock.patch(
         "cwms_batch_events.core.auth.user.jwt.settings.auth_environment", "LOCAL"

tests/cwms_batch_events/core/test_job_loggers.py

@@ -17,7 +17,7 @@ def test_s3_job_logger_reads_logs_from_bucket():
     with mock.patch(
         "cwms_batch_events.core.job_logger.s3.boto3.client",
         return_value=s3_client,
-    ), mock.patch("cwms_batch_events.core.job_logger.s3.S3_BUCKET", "bucket"):
+    ), mock.patch("cwms_batch_events.core.job_logger.s3.settings.s3_bucket", "bucket"):
         logger = S3JobLogger()
         logs = logger.get_logs_for_job(uuid4())
 
@@ -31,7 +31,7 @@ def test_s3_job_logger_pushes_logs_to_bucket():
     with mock.patch(
         "cwms_batch_events.core.job_logger.s3.boto3.client",
         return_value=s3_client,
-    ), mock.patch("cwms_batch_events.core.job_logger.s3.S3_BUCKET", "bucket"):
+    ), mock.patch("cwms_batch_events.core.job_logger.s3.settings.s3_bucket", "bucket"):
         logger = S3JobLogger()
         logger.push_logs_for_job(job_id, "hello")
 
@@ -42,6 +42,12 @@ def test_s3_job_logger_pushes_logs_to_bucket():
     )
 
 
+def test_s3_job_logger_requires_s3_bucket():
+    with mock.patch("cwms_batch_events.core.job_logger.s3.settings.s3_bucket", None):
+        with pytest.raises(ValueError, match="S3_BUCKET must be configured for S3 logging"):
+            S3JobLogger()
+
+
 def test_cloudwatch_job_logger_get_job_details_requires_existing_job():
     db = mock.Mock()
     db.get_job_by_id.return_value = None

tests/cwms_batch_events/core/test_settings.py

@@ -0,0 +1,49 @@
+import pytest
+from pydantic import ValidationError
+
+from cwms_batch_events.core.settings import Settings
+
+
+def test_fastapi_root_path_defaults_to_empty_string():
+    settings = Settings(root_path=None, app_key="x", auth_environment="TEST")
+
+    assert settings.fastapi_root_path == ""
+
+
+def test_settings_rejects_blank_strings_as_missing(monkeypatch):
+    monkeypatch.delenv("APP_KEY", raising=False)
+    monkeypatch.delenv("AUTH_ENVIRONMENT", raising=False)
+
+    with pytest.raises(ValidationError, match="app_key"):
+        Settings(app_key="", auth_environment="TEST")
+
+
+def test_settings_require_auth_environment_when_mock_user_false(monkeypatch):
+    monkeypatch.delenv("AUTH_ENVIRONMENT", raising=False)
+
+    with pytest.raises(ValueError, match="AUTH_ENVIRONMENT must be configured"):
+        Settings(app_key="secret", mock_user=False)
+
+
+def test_settings_require_s3_bucket_for_docker_local_runner():
+    with pytest.raises(ValueError, match="S3_BUCKET must be configured"):
+        Settings(app_key="secret", auth_environment="TEST", default_job_runner="docker-local")
+
+
+def test_settings_require_docker_local_endpoints():
+    with pytest.raises(ValueError, match="S3_ENDPOINT_URL must be configured"):
+        Settings(
+            app_key="secret",
+            auth_environment="TEST",
+            default_job_runner="docker-local",
+            s3_bucket="bucket",
+        )
+
+    with pytest.raises(ValueError, match="SQS_ENDPOINT_URL must be configured"):
+        Settings(
+            app_key="secret",
+            auth_environment="TEST",
+            default_job_runner="docker-local",
+            s3_bucket="bucket",
+            s3_endpoint_url="http://s3",
+        )