Narrow reco counts request JSON

Author: adamtheturtle

src/mock_vws/_reco_counts_web_api.py

@@ -3,19 +3,19 @@
 import base64
 import datetime
 import email.utils
-import json
 import logging
 import re
 import secrets
 import string
 import uuid
 from collections.abc import Mapping
 from http import HTTPStatus
-from typing import Any, Protocol, runtime_checkable
+from typing import Protocol, TypedDict, runtime_checkable
 from urllib.parse import parse_qs, urlencode, urlsplit
 from zoneinfo import ZoneInfo
 
 from beartype import beartype
+from pydantic import TypeAdapter, ValidationError
 
 from mock_vws._constants import ResultCodes
 from mock_vws._mock_common import json_dump
@@ -38,6 +38,15 @@
 _SIGNING_REGION = "us-west-1"
 
 
+class _RecoCountsRequest(TypedDict):
+    """JSON body for a recognition-count report request."""
+
+    month: str
+
+
+_RECO_COUNTS_REQUEST_ADAPTER = TypeAdapter(type=_RecoCountsRequest)
+
+
 @runtime_checkable
 class RecoCountsReportStore(Protocol):
     """Storage for generated reco counts reports."""
@@ -285,13 +294,13 @@ def create_reco_counts_report(
         FailError: The given month is not a month in the ``YYYY-mm`` form
             which the report can be requested for.
     """
-    request_json: dict[str, Any] = json.loads(s=request_body)  # pyrefly: ignore [explicit-any]
+    try:
+        request_json = _RECO_COUNTS_REQUEST_ADAPTER.validate_json(request_body)
+    except ValidationError as exc:
+        _LOGGER.warning(msg='The given "month" is not in the YYYY-mm form.')
+        raise FailError(status_code=HTTPStatus.BAD_REQUEST) from exc
     month = request_json["month"]
-    if not isinstance(month, str) or not bool(
-        _MONTH_PATTERN.fullmatch(
-            string=month,
-        )
-    ):
+    if not bool(_MONTH_PATTERN.fullmatch(string=month)):
         _LOGGER.warning(msg='The given "month" is not in the YYYY-mm form.')
         raise FailError(status_code=HTTPStatus.BAD_REQUEST)
 

tests/mock_vws/test_reco_counts_report.py

@@ -83,11 +83,11 @@ def _month_offset_from_now(*, months: int) -> str:
 
 
 @beartype
-def _request_reco_counts_report(
+def _request_reco_counts_report_json(
     *,
     vuforia_database: CloudDatabase,
     database_id: str,
-    month: str | int,
+    request_json: object,
 ) -> requests.Response:
     """Request a reco counts report and return the response.
 
@@ -96,7 +96,7 @@ def _request_reco_counts_report(
     """
     request_path = f"/imagetargets/databases/{database_id}/reports/recoCounts"
     content_type = "application/json"
-    content = json.dumps(obj={"month": month}).encode(encoding="utf-8")
+    content = json.dumps(obj=request_json).encode(encoding="utf-8")
     date = rfc_1123_date()
     authorization_string = authorization_header(
         access_key=vuforia_database.server_access_key,
@@ -121,6 +121,21 @@ def _request_reco_counts_report(
     )
 
 
+@beartype
+def _request_reco_counts_report(
+    *,
+    vuforia_database: CloudDatabase,
+    database_id: str,
+    month: str | int,
+) -> requests.Response:
+    """Request a reco counts report for one month."""
+    return _request_reco_counts_report_json(
+        vuforia_database=vuforia_database,
+        database_id=database_id,
+        request_json={"month": month},
+    )
+
+
 @beartype
 def _presigned_url(*, vuforia_database: CloudDatabase, month: str) -> str:
     """Request a report for the given month and return its download
@@ -386,6 +401,22 @@ def test_malformed_month(
         response_json = json.loads(s=response.text)
         assert response_json["result_code"] == ResultCodes.FAIL.value
 
+    @staticmethod
+    def test_body_is_not_an_object(
+        *,
+        vuforia_database: CloudDatabase,
+    ) -> None:
+        """The request body must be a JSON object."""
+        response = _request_reco_counts_report_json(
+            vuforia_database=vuforia_database,
+            database_id=vuforia_database.database_id,
+            request_json=[],
+        )
+
+        assert response.status_code == HTTPStatus.BAD_REQUEST
+        response_json = json.loads(s=response.text)
+        assert response_json["result_code"] == ResultCodes.FAIL.value
+
     @staticmethod
     def test_unknown_database_id(*, vuforia_database: CloudDatabase) -> None:
         """The path must name the database which the request's server