Merge pull request #3218 from VWS-Python/fix/report-json-boundaries

Validate report response values

Author: adamtheturtle

newsfragments/3218.change.rst

@@ -0,0 +1 @@
+Validate query and Model Target report response values at construction time.

src/vws/reports.py

@@ -3,12 +3,42 @@
 import csv
 import datetime
 import io
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from enum import Enum, unique
-from typing import Any, Self
+from typing import Any, Self, TypeIs
 
 from beartype import BeartypeConf, beartype
+from beartype.door import TypeHint
+
+
+def _checked[T](value: object, hint: type[T], /) -> T:
+    """Return a value after checking its runtime type."""
+    if not _is_type(value, hint):
+        msg = f"Expected {hint!r}, got {value!r}."
+        raise TypeError(msg)
+    return value
+
+
+def _is_type[T](value: object, hint: type[T], /) -> TypeIs[T]:
+    """Return whether a value satisfies a runtime type."""
+    return TypeHint(hint=hint).is_bearable(obj=value)
+
+
+def _number(value: object, /) -> int | float:
+    """Return a runtime-validated JSON number."""
+    if isinstance(value, bool) or not isinstance(value, int | float):
+        msg = f"Expected a number, got {value!r}."
+        raise TypeError(msg)
+    return value
+
+
+def _optional_string(value: object, /) -> str | None:
+    """Return a runtime-validated optional string."""
+    if value is not None and not isinstance(value, str):
+        msg = f"Expected an optional string, got {value!r}."
+        raise TypeError(msg)
+    return value
 
 
 @beartype
@@ -143,23 +173,29 @@ class QueryResult:
     target_data: TargetData | None
 
     @classmethod
-    def from_response_dict(cls, response_dict: dict[str, Any]) -> Self:  # pyrefly: ignore [explicit-any]
+    def from_response_dict(
+        cls,
+        response_dict: Mapping[str, object],
+    ) -> Self:
         """Construct from a VWS API query result item dict."""
         target_data: TargetData | None = None
         if "target_data" in response_dict:
-            target_data_dict = response_dict["target_data"]
+            target_data_dict = _checked(
+                response_dict["target_data"], dict[str, object]
+            )
             target_timestamp = datetime.datetime.fromtimestamp(
-                timestamp=target_data_dict["target_timestamp"],  # pyrefly: ignore [unknown-argument-type]
+                timestamp=_number(target_data_dict["target_timestamp"]),
                 tz=datetime.UTC,
             )
             target_data = TargetData(
-                name=target_data_dict["name"],  # pyrefly: ignore [unknown-argument-type]
-                # pyrefly: ignore [unknown-argument-type]
-                application_metadata=target_data_dict["application_metadata"],
+                name=_checked(target_data_dict["name"], str),
+                application_metadata=_optional_string(
+                    target_data_dict["application_metadata"]
+                ),
                 target_timestamp=target_timestamp,
             )
         return cls(
-            target_id=response_dict["target_id"],
+            target_id=_checked(response_dict["target_id"], str),
             target_data=target_data,
         )
 
@@ -301,48 +337,59 @@ class ModelTargetDatasetStatusReport:
     """
 
     @classmethod
-    def from_response_dict(cls, response_dict: dict[str, Any]) -> Self:  # pyrefly: ignore [explicit-any]
+    def from_response_dict(
+        cls,
+        response_dict: Mapping[str, object],
+    ) -> Self:
         """Construct from a Model Target Web API response dict."""
         error: ModelTargetGenerationError | None = None
         if "error" in response_dict:
-            error_dict = dict(response_dict["error"])
+            error_dict = _checked(response_dict["error"], dict[str, object])
             error = ModelTargetGenerationError(
-                code=error_dict["code"],
-                message=error_dict["message"],
+                code=_checked(error_dict["code"], str),
+                message=_checked(error_dict["message"], str),
             )
 
         warning: ModelTargetGenerationWarning | None = None
         if "warning" in response_dict:
-            warning_dict = dict(response_dict["warning"])
+            warning_dict = _checked(
+                response_dict["warning"], dict[str, object]
+            )
+            details = _checked(
+                warning_dict["details"], list[dict[str, object]]
+            )
             warning = ModelTargetGenerationWarning(
-                code=warning_dict["code"],
-                message=warning_dict["message"],
-                target=warning_dict["target"],
+                code=_checked(warning_dict["code"], str),
+                message=_checked(warning_dict["message"], str),
+                target=_checked(warning_dict["target"], str),
                 details=[
                     ModelTargetGenerationDetail(
-                        code=detail["code"],  # pyrefly: ignore [unknown-argument-type]
-                        # pyrefly: ignore [unknown-argument-type]
-                        message=detail["message"],
+                        code=_checked(detail["code"], str),
+                        message=_checked(detail["message"], str),
                     )
-                    for detail in warning_dict["details"]
+                    for detail in details
                 ],
             )
 
         eta: datetime.datetime | None = None
         if "eta" in response_dict:
-            eta = datetime.datetime.fromisoformat(response_dict["eta"])
+            eta = datetime.datetime.fromisoformat(
+                _checked(response_dict["eta"], str),
+            )
 
         completed_at: datetime.datetime | None = None
         if "completedAt" in response_dict:
             completed_at = datetime.datetime.fromisoformat(
-                response_dict["completedAt"],
+                _checked(response_dict["completedAt"], str),
             )
 
         return cls(
-            status=ModelTargetDatasetStatuses(value=response_dict["status"]),
-            dataset_uuid=response_dict["uuid"],
+            status=ModelTargetDatasetStatuses(
+                value=_checked(response_dict["status"], str),
+            ),
+            dataset_uuid=_checked(response_dict["uuid"], str),
             created_at=datetime.datetime.fromisoformat(
-                response_dict["createdAt"],
+                _checked(response_dict["createdAt"], str),
             ),
             eta=eta,
             completed_at=completed_at,

tests/test_reports.py

@@ -0,0 +1,35 @@
+"""Tests for public report constructors."""
+
+import pytest
+
+from vws.reports import QueryResult
+
+
+@pytest.mark.parametrize(
+    argnames="response",
+    argvalues=[
+        {"target_id": 1},
+        {
+            "target_id": "target-id",
+            "target_data": {
+                "name": "target-name",
+                "application_metadata": None,
+                "target_timestamp": True,
+            },
+        },
+        {
+            "target_id": "target-id",
+            "target_data": {
+                "name": "target-name",
+                "application_metadata": 1,
+                "target_timestamp": 0,
+            },
+        },
+    ],
+)
+def test_query_result_rejects_invalid_response_values(
+    *, response: dict[str, object]
+) -> None:
+    """Query reports reject values of the wrong type."""
+    with pytest.raises(expected_exception=TypeError):
+        _ = QueryResult.from_response_dict(response_dict=response)