PDM-892 mypy fixes and setup-java action update

Author: davidhallam4-nhs

.gitallowed

@@ -1,7 +1,9 @@
 
 token: \$\{\{ secrets.GITHUB_TOKEN \}\}
-"token": response.get\("SessionToken"\)
+"token": response\["SessionToken"\]
 token=credentials\["token"\]
+arn:aws:iam::000000000000:root
+account_id=\"000000000000\"
 
 .*(GITHUB|SONAR)_TOKEN: \$\{\{ secrets.(GITHUB|SONAR)_TOKEN \}\}
 .*asttokens = ">=2.1.0"

.github/workflows/merge-develop.yml

@@ -58,7 +58,7 @@ jobs:
 
       - name: setup java
         if: github.actor != 'dependabot[bot]' && (success() || failure())
-        uses: actions/setup-java@v4
+        uses: actions/setup-java@v5
         with:
           distribution: "corretto"
           java-version: "17"

.github/workflows/pull-request.yml

@@ -11,7 +11,7 @@ jobs:
   tox:
     strategy:
       matrix:
-        python-version: ["3.9", "3.10", "3.11"]
+        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
 
     runs-on: ubuntu-latest
     if: github.repository == 'NHSDigital/nhs-aws-helpers'
@@ -145,7 +145,7 @@ jobs:
 
       - name: setup java
         if: success() || failure()
-        uses: actions/setup-java@v4
+        uses: actions/setup-java@v5
         with:
           distribution: "corretto"
           java-version: "17"

nhs_aws_helpers/__init__.py

@@ -75,6 +75,7 @@
 from mypy_boto3_ssm.client import SSMClient
 from mypy_boto3_stepfunctions import SFNClient
 from mypy_boto3_sts.client import STSClient
+from mypy_boto3_sts.type_defs import AssumeRoleRequestTypeDef
 
 from nhs_aws_helpers.common import run_in_executor
 from nhs_aws_helpers.s3_object_writer import S3ObjectWriter
@@ -1056,19 +1057,19 @@ def assumed_credentials(
 
     sts_client = boto3.client("sts", region_name=region, endpoint_url=endpoint_url)
 
-    params = {
-        "RoleArn": f"arn:aws:iam::{account_id}:role/{role}",
-        "RoleSessionName": role_session_name,
-        "DurationSeconds": duration_seconds,
-    }
+    params = AssumeRoleRequestTypeDef(
+        RoleArn=f"arn:aws:iam::{account_id}:role/{role}",
+        RoleSessionName=role_session_name,
+        DurationSeconds=duration_seconds,
+    )
 
-    response = sts_client.assume_role(**params).get("Credentials")
+    response = sts_client.assume_role(**params)["Credentials"]
 
     credentials = {
-        "access_key": response.get("AccessKeyId"),
-        "secret_key": response.get("SecretAccessKey"),
-        "token": response.get("SessionToken"),
-        "expiry_time": response.get("Expiration").isoformat(),
+        "access_key": response["AccessKeyId"],
+        "secret_key": response["SecretAccessKey"],
+        "token": response["SessionToken"],
+        "expiry_time": response["Expiration"].isoformat(),
     }
     return credentials
 

tests/aws_tests.py

@@ -1,6 +1,6 @@
 import logging
 import os
-from typing import Any, cast
+from typing import Any, Literal, TypedDict, cast
 from uuid import uuid4
 
 import pytest
@@ -10,6 +10,7 @@
 from pytest_httpserver import HTTPServer
 
 from nhs_aws_helpers import (
+    assumed_credentials,
     dynamodb_retry_backoff,
     post_create_client,
     register_config_default,
@@ -247,14 +248,20 @@ def test_s3_retries(
     key = f"{expected_folder}/filename.txt"
     httpserver.expect_request(f"/{temp_s3_bucket.name}/{key}").respond_with_json({}, status=503)
 
+    # Type-identical spec to match botocore's internal typing of _RetryDict
+    class BotocoreRetries(TypedDict, total=False):
+        total_max_attempts: int
+        max_attempts: int
+        mode: Literal["legacy", "standard", "adaptive"]
+
     config = Config(
         connect_timeout=float(os.environ.get("BOTO_CONNECT_TIMEOUT", "1")),
         read_timeout=float(os.environ.get("BOTO_READ_TIMEOUT", "1")),
         max_pool_connections=int(os.environ.get("BOTO_MAX_POOL_CONNECTIONS", "10")),
-        retries={
-            "mode": os.environ.get("BOTO_RETRIES_MODE", "standard"),  # type: ignore[typeddict-item]
-            "total_max_attempts": int(os.environ.get("BOTO_RETRIES_TOTAL_MAX_ATTEMPTS", "2")),
-        },
+        retries=BotocoreRetries(
+            mode=os.environ.get("BOTO_RETRIES_MODE", "standard"),  # type: ignore[typeddict-item]
+            total_max_attempts=int(os.environ.get("BOTO_RETRIES_TOTAL_MAX_ATTEMPTS", "2")),
+        ),
     )
 
     def _post_create_aws(boto_module: str, _: str, client):
@@ -359,3 +366,13 @@ def test_s3_upload_multipart_from_copy_missing_part_data(temp_s3_bucket: Bucket)
         target_object.get()
 
     assert client_error.value.response["Error"]["Code"] == "NoSuchKey"
+
+
+def test_assumed_credentials():
+    creds = assumed_credentials(
+        account_id="000000000000", role="MyRole", sts_endpoint_url=os.environ["AWS_ENDPOINT_URL"]
+    )
+    assert creds["access_key"]
+    assert creds["secret_key"]
+    assert creds["token"]
+    assert creds["expiry_time"]

tests/client_tests.py

@@ -0,0 +1,24 @@
+from nhs_aws_helpers import ecs_client, healthlake_client, sts_client
+
+
+def test_sts_client():
+    client = sts_client()
+    assert client.meta.service_model.service_id == "STS"
+
+    identity = client.get_caller_identity()
+    assert identity["Account"] == "000000000000"
+    assert identity["Arn"] == "arn:aws:iam::000000000000:root"
+
+
+def test_healthlake_client():
+    client = healthlake_client()
+    # localstack pro subscription required for HealthLake support
+    # so just check the client is created correctly
+    assert client.meta.service_model.service_id == "HealthLake"
+
+
+def test_ecs_client():
+    client = ecs_client()
+    # localstack pro subscription required for ECS support
+    # so just check the client is created correctly
+    assert client.meta.service_model.service_id == "ECS"