Validate dynamic values instead of ignoring Ty

src/vws/_model_targets.py

@@ -83,8 +83,15 @@ def access_token_from_response(*, response: Response) -> tuple[str, float]:
     if response.status_code != HTTPStatus.OK:
         raise ModelTargetOAuth2Error(response=response)
 
-    response_data = dict(json.loads(s=response.text))
-    return response_data["access_token"], float(response_data["expires_in"])  # ty: ignore[unsound-return-statement]
+    response_data = dict[str, object](json.loads(s=response.text))
+    access_token = response_data.get("access_token")
+    expires_in = response_data.get("expires_in")
+    if not isinstance(access_token, str) or not isinstance(
+        expires_in,
+        str | int | float,
+    ):
+        raise ModelTargetOAuth2Error(response=response)
+    return access_token, float(expires_in)
 
 
 @beartype(conf=BeartypeConf(is_pep484_tower=True))

tests/conftest.py

@@ -35,6 +35,11 @@
 _MODEL_TARGET_CLIENT_SECRET = "client-secret"  # noqa: S105
 
 
+def _image_file_mode(*, value: Literal["r+b", "rb"]) -> Literal["r+b", "rb"]:
+    """Type a file mode supplied by the pytest parameter API."""
+    return value
+
+
 @pytest.fixture(name="_mock_database")
 def fixture_mock_database() -> Generator[CloudDatabase]:
     """Yield a mock ``CloudDatabase``."""
@@ -225,7 +230,7 @@ def fixture_image_file(
     file = tmp_path / "image.jpg"
     buffer = high_quality_image.getvalue()
     _ = file.write_bytes(data=buffer)
-    mode: Literal["r+b", "rb"] = request.param  # ty: ignore[unsound-assignment]
+    mode = _image_file_mode(value=request.param)
     with file.open(mode=mode) as file_obj:
         yield file_obj
 

tests/test_model_targets.py

@@ -17,6 +17,7 @@
 )
 
 from vws import ModelTargetService
+from vws._model_targets import access_token_from_response
 from vws.exceptions.custom_exceptions import ServerError
 from vws.exceptions.model_target_exceptions import (
     ModelTargetAuthenticationError,
@@ -52,11 +53,16 @@
 
 
 @beartype
-def _response(*, text: str) -> Response:
+def _response_with_status(
+    *,
+    text: str,
+    status_code: HTTPStatus,
+) -> Response:
     """Get a response with a given body.
 
     Args:
         text: The body of the response.
+        status_code: The response status code.
 
     Returns:
         A response with the given body.
@@ -65,14 +71,23 @@ def _response(*, text: str) -> Response:
     return Response(
         text=text,
         url="https://vws.vuforia.com/modeltargets/datasets",
-        status_code=HTTPStatus.BAD_REQUEST,
+        status_code=status_code,
         headers={},
         request_body=None,
         tell_position=len(content),
         content=content,
     )
 
 
+@beartype
+def _response(*, text: str) -> Response:
+    """Get a bad-request response with a given body."""
+    return _response_with_status(
+        text=text,
+        status_code=HTTPStatus.BAD_REQUEST,
+    )
+
+
 @beartype
 class _CountingTransport:
     """A transport which counts the requests made to each path."""
@@ -759,6 +774,26 @@ def test_oauth2_error_description() -> None:
         assert error.error_description == description
 
 
+@pytest.mark.parametrize(
+    argnames="payload",
+    argvalues=[
+        {"access_token": 1, "expires_in": 3600},
+        {"access_token": "token", "expires_in": []},
+    ],
+)
+def test_invalid_oauth2_token_response_values(
+    *, payload: dict[str, object]
+) -> None:
+    """OAuth token responses must contain correctly typed values."""
+    response = _response_with_status(
+        text=json.dumps(obj=payload),
+        status_code=HTTPStatus.OK,
+    )
+
+    with pytest.raises(expected_exception=ModelTargetOAuth2Error):
+        _ = access_token_from_response(response=response)
+
+
 class TestBaseVWSURL:
     """Tests for using a custom base URL."""