diff --git a/src/vws/_model_targets.py b/src/vws/_model_targets.py index 58799b8ec..9ed51c437 100644 --- a/src/vws/_model_targets.py +++ b/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)) diff --git a/tests/conftest.py b/tests/conftest.py index 8b95f5dbf..515034e35 100644 --- a/tests/conftest.py +++ b/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 diff --git a/tests/test_model_targets.py b/tests/test_model_targets.py index fc8f9d4d4..19fe753af 100644 --- a/tests/test_model_targets.py +++ b/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,7 +71,7 @@ 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), @@ -73,6 +79,15 @@ def _response(*, text: str) -> Response: ) +@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."""