Migrate dev to test

.github/workflows/cwbi-test-push-api.yml

@@ -0,0 +1,72 @@
+name: Python Tests on PR
+
+on:
+  pull_request:
+    branches:
+      - cwbi-dev
+      - cwbi-test
+      - cwbi-prod
+    paths:
+      - "cwms_batch_events/**"
+      - "tests/**"
+      - "requirements.txt"
+      - "requirements-dev.txt"
+      - "pytest.ini"
+      - ".github/workflows/cwbi-test-push-api.yml"
+
+permissions:
+  contents: read
+
+jobs:
+  python-tests:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.13"
+          cache: "pip"
+          cache-dependency-path: |
+            requirements.txt
+            requirements-dev.txt
+
+      - name: Install dependencies
+        run: python -m pip install -r requirements-dev.txt
+
+      - name: Run pytest
+        run: pytest --cov-report=html --cov-report=json
+
+      - name: Upload coverage HTML report
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: python-htmlcov
+          path: htmlcov/
+
+      - name: Add coverage summary
+        if: always()
+        run: |
+          python - <<'PY'
+          import json
+          import os
+          from pathlib import Path
+
+          coverage_file = Path("coverage.json")
+          if not coverage_file.exists():
+              raise SystemExit("coverage.json was not generated")
+
+          data = json.loads(coverage_file.read_text())
+          total = data["totals"]["percent_covered_display"]
+
+          summary = Path(os.environ["GITHUB_STEP_SUMMARY"])
+          summary.write_text(
+              "## Python Coverage\n\n"
+              f"- Total coverage: `{total}%`\n"
+              "- HTML report: download the `python-htmlcov` artifact from this workflow run.\n",
+              encoding="utf-8",
+          )
+          PY

.gitignore

@@ -42,6 +42,7 @@ htmlcov/
 .nox/
 .coverage
 .coverage.*
+coverage.json
 .cache
 nosetests.xml
 coverage.xml

CONTRIBUTING.md

@@ -25,4 +25,7 @@
 ## Notes
 * Type Standardization
   * TypeScript types are generated from the API for use in the frontend using [OpenAPI TypeScript](https://openapi-ts.dev/). 
-  * If API types are updated or modified, run `npm run generate:types` to update the type definitions.
+  * If API types are updated or modified, run `npm run generate:types` to update the type definitions.
+* Python Testing
+  * Run `pytest` from the project root to execute the Python unit test suite.
+  * The suite is configured to report coverage for `cwms_batch_events` and should stay fast and free of external dependencies.

cwms_batch_events/api/routers/scripts.py

@@ -63,7 +63,7 @@ def post_script(
         return job_db.store_script(payload)
     except ValueError as e:
         raise HTTPException(
-            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(e)
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(e)
         )
     except SlugError as e:
         raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e))
@@ -87,7 +87,7 @@ def put_script(
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(e))
     except ValueError as e:
         raise HTTPException(
-            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(e)
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(e)
         )
 
 

cwms_batch_events/core/auth/user/dependencies.py

@@ -1,6 +1,7 @@
 from fastapi import Depends, HTTPException, Request, status
 from fastapi.security import APIKeyHeader, HTTPAuthorizationCredentials, HTTPBearer
 
+from cachetools import TTLCache
 from cwms_batch_events.core.auth.user.jwt import verify_jwt
 from cwms_batch_events.core.auth.user.models import User
 from cwms_batch_events.core.auth.user.roles import (
@@ -24,6 +25,8 @@
     description="Use format: apikey <your-api-key>",
 )
 
+user_cache: TTLCache[str, User] = TTLCache(maxsize=1024, ttl=300)
+
 
 async def get_auth_header_for_docs(
     bearer: str = Depends(bearer_scheme),
@@ -47,6 +50,10 @@ async def get_auth_credentials(
 async def get_current_user_cwms(
     credentials: HTTPAuthorizationCredentials = Depends(get_auth_credentials),
 ) -> User:
+    cache_key = f"{credentials.scheme}:{credentials.credentials}"
+    if cache_key in user_cache:
+        return user_cache[cache_key]
+
     if credentials.scheme.lower() == "bearer":
         token = credentials.credentials
         try:
@@ -71,13 +78,16 @@ async def get_current_user_cwms(
 
     allowed_offices = get_user_allowed_offices(cda_user)
     admin_offices = get_user_admin_offices(cda_user)
-    return User(
+    user = User(
         username=cda_user.user_name,
         offices=allowed_offices,
         admin_offices=admin_offices,
         roles=cda_user.roles,
     )
 
+    user_cache[cache_key] = user
+    return user
+
 
 async def get_current_user_mock() -> User:
     return User(

cwms_batch_events/core/auth/user/jwt.py

@@ -16,7 +16,7 @@
 
 ISSUER = {
     "PROD": "https://identity.sec.usace.army.mil/auth/realms/cwbi",
-    "TEST": "https://identity-test.cwbi.us/auth/realms/cwbi",
+    "TEST": "https://identity-test.cwbi.mil/auth/realms/cwbi",
 }
 
 

cwms_batch_events/lambdas/dispatch_job/dispatcher.py

@@ -100,6 +100,7 @@ def lambda_handler(event, context):
             external_job_id = dispatch_job(message)
         except ClientError:
             logger.exception("Failed to submit batch job for message: %s", message)
+            raise
 
         try:
             bind_request = BindExternalJobIdRequest(external_job_id=external_job_id)

pytest.ini

@@ -0,0 +1,6 @@
+[pytest]
+testpaths = tests
+addopts = -ra --strict-markers --cov=cwms_batch_events --cov-branch --cov-report=term-missing
+markers =
+    unit: fast isolated tests
+    integration: tests that exercise component boundaries

requirements-dev.txt

@@ -1,3 +1,5 @@
 -r requirements.txt
 pytest
-docker
+pytest-cov
+pytest-trio
+docker

requirements.txt

@@ -1,4 +1,5 @@
 boto3
+cachetools
 fastapi
 psycopg2-binary
 pydantic

tests/conftest.py

@@ -0,0 +1,13 @@
+import os
+import sys
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+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")

tests/cwms_batch_events/api/conftest.py

@@ -0,0 +1,49 @@
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi.testclient import TestClient
+
+from cwms_batch_events.api.dependencies import (
+    get_current_user,
+    get_job_database,
+    get_job_logger,
+    get_job_queue,
+)
+from cwms_batch_events.api.main import app
+from cwms_batch_events.core.auth.service.dependencies import require_internal_auth
+from cwms_batch_events.core.auth.user.models import User
+from tests.factories import make_user
+
+
+@pytest.fixture
+def user() -> User:
+    return make_user()
+
+
+@pytest.fixture
+def job_db():
+    return MagicMock()
+
+
+@pytest.fixture
+def job_logger():
+    return MagicMock()
+
+
+@pytest.fixture
+def job_queue():
+    return MagicMock()
+
+
+@pytest.fixture
+def client(user: User, job_db: MagicMock, job_logger: MagicMock, job_queue: MagicMock):
+    app.dependency_overrides[get_current_user] = lambda: user
+    app.dependency_overrides[get_job_database] = lambda: job_db
+    app.dependency_overrides[get_job_logger] = lambda: job_logger
+    app.dependency_overrides[get_job_queue] = lambda: job_queue
+    app.dependency_overrides[require_internal_auth] = lambda: True
+
+    with TestClient(app) as test_client:
+        yield test_client
+
+    app.dependency_overrides.clear()

tests/cwms_batch_events/api/test_dependencies.py

@@ -0,0 +1,75 @@
+from unittest import mock
+
+from cwms_batch_events.api import dependencies
+
+
+def test_get_db_session_closes_session():
+    session = mock.Mock()
+
+    with mock.patch(
+        "cwms_batch_events.api.dependencies.create_session",
+        return_value=session,
+    ):
+        generator = dependencies.get_db_session()
+        yielded = next(generator)
+
+        assert yielded is session
+
+        try:
+            next(generator)
+        except StopIteration:
+            pass
+
+    session.close.assert_called_once()
+
+
+def test_get_job_database_wraps_session_in_postgres_job_database():
+    session = object()
+
+    with mock.patch(
+        "cwms_batch_events.api.dependencies.PostgresJobDatabase",
+        return_value="db",
+    ) as postgres_db:
+        db = dependencies.get_job_database(session)
+
+    assert db == "db"
+    postgres_db.assert_called_once_with(db=session)
+
+
+def test_get_job_logger_uses_s3_for_local_runner():
+    with mock.patch(
+        "cwms_batch_events.api.dependencies.settings.default_job_runner", "docker-local"
+    ), mock.patch(
+        "cwms_batch_events.api.dependencies.S3JobLogger",
+        return_value="logger",
+    ) as s3_logger:
+        logger = dependencies.get_job_logger(mock.Mock())
+
+    assert logger == "logger"
+    s3_logger.assert_called_once_with()
+
+
+def test_get_job_logger_uses_cloudwatch_for_batch_runner():
+    db = mock.Mock()
+
+    with mock.patch(
+        "cwms_batch_events.api.dependencies.settings.default_job_runner", "batch"
+    ), mock.patch(
+        "cwms_batch_events.api.dependencies.CloudWatchJobLogger",
+        return_value="logger",
+    ) as cloudwatch_logger:
+        logger = dependencies.get_job_logger(db)
+
+    assert logger == "logger"
+    cloudwatch_logger.assert_called_once_with(db)
+
+
+def test_get_job_queue_constructs_job_queue():
+    with mock.patch(
+        "cwms_batch_events.api.dependencies.JobQueue",
+        return_value="queue",
+    ) as job_queue_cls:
+        queue = dependencies.get_job_queue()
+
+    assert queue == "queue"
+    job_queue_cls.assert_called_once_with()

tests/cwms_batch_events/api/test_health.py

@@ -0,0 +1,5 @@
+def test_health_returns_ok(client):
+    response = client.get("/health")
+
+    assert response.status_code == 200
+    assert response.json() == {"status": "ok"}