From 82296d25e630b6b675db9563c1baffbd2c32f397 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Tue, 8 Sep 2026 20:55:46 +0100 Subject: [PATCH] Validate JSON response shapes --- src/vws/_json_utils.py | 69 +++++++++++++++++++++++++ src/vws/async_query.py | 15 ++++-- src/vws/async_vumark_service.py | 8 ++- src/vws/async_vws.py | 31 ++++++++---- src/vws/exceptions/vws_exceptions.py | 6 +-- src/vws/query.py | 15 ++++-- src/vws/vumark_service.py | 8 ++- src/vws/vws.py | 31 ++++++++---- tests/test_query.py | 52 +++++++++++++++++++ tests/test_vws.py | 75 ++++++++++++++++++++++++++++ 10 files changed, 276 insertions(+), 34 deletions(-) create mode 100644 src/vws/_json_utils.py diff --git a/src/vws/_json_utils.py b/src/vws/_json_utils.py new file mode 100644 index 000000000..a5784d7d1 --- /dev/null +++ b/src/vws/_json_utils.py @@ -0,0 +1,69 @@ +"""Validation helpers for JSON received from remote services.""" + +import json +from typing import TypeGuard + + +def _is_json_object(value: object, /) -> TypeGuard[dict[str, object]]: + """Return whether a decoded JSON value is an object.""" + return isinstance(value, dict) + + +def _is_object_list(value: object, /) -> TypeGuard[list[object]]: + """Return whether a decoded JSON value is an array.""" + return isinstance(value, list) + + +def _validated_object(*, value: object) -> dict[str, object]: + """Return a decoded JSON object.""" + if not _is_json_object(value): + msg = "Expected a JSON object." + raise TypeError(msg) + return value + + +def json_object(*, value: str | bytes | bytearray) -> dict[str, object]: + """Decode and validate a JSON object.""" + loaded: object = json.loads(s=value) + return _validated_object(value=loaded) + + +def string_value(*, value: object, name: str) -> str: + """Return a JSON value after validating that it is a string.""" + if not isinstance(value, str): + msg = f"{name} must be a string." + raise TypeError(msg) + return value + + +def string_field(*, value: dict[str, object], name: str) -> str: + """Return a required string field from a JSON object.""" + return string_value(value=value[name], name=name) + + +def string_list_field( + *, + value: dict[str, object], + name: str, +) -> list[str]: + """Return a required list of strings from a JSON object.""" + items = value[name] + if not _is_object_list(items) or not all( + isinstance(item, str) for item in items + ): + msg = f"{name} must be a list of strings." + raise TypeError(msg) + return [item for item in items if isinstance(item, str)] + + +def object_list_field( + *, + value: dict[str, object], + name: str, +) -> list[dict[str, object]]: + """Return a required list of JSON objects.""" + items = value[name] + if not _is_object_list(items): + msg = f"{name} must be a list of JSON objects." + raise TypeError(msg) + return [_validated_object(value=item) for item in items] diff --git a/src/vws/async_query.py b/src/vws/async_query.py index 7cc4d1d76..1e4ec9183 100644 --- a/src/vws/async_query.py +++ b/src/vws/async_query.py @@ -12,6 +12,7 @@ from vws._image_utils import ImageType as _ImageType from vws._image_utils import get_image_data as _get_image_data +from vws._json_utils import json_object, object_list_field, string_field from vws.exceptions.base_exceptions import CloudRecoError from vws.exceptions.cloud_reco_exceptions import ( AuthenticationFailureError, @@ -201,13 +202,16 @@ async def query( raise CloudRecoError(response=response) try: - response_body = json.loads(s=response.text) + response_body = json_object(value=response.text) except json.JSONDecodeError as exc: if response.status_code >= HTTPStatus.BAD_REQUEST: raise CloudRecoError(response=response) from exc raise - result_code = response_body["result_code"] # pyrefly: ignore [unknown-variable-type] + result_code = string_field( + value=response_body, + name="result_code", + ) if result_code != "Success": exception = { "AuthenticationFailure": (AuthenticationFailureError), @@ -217,8 +221,11 @@ async def query( }[result_code] raise exception(response=response) - result_list = list(response_body["results"]) # pyrefly: ignore [unknown-argument-type] + result_list = object_list_field( + value=response_body, + name="results", + ) return [ - QueryResult.from_response_dict(response_dict=item) # pyrefly: ignore [unknown-argument-type] + QueryResult.from_response_dict(response_dict=item) for item in result_list ] diff --git a/src/vws/async_vumark_service.py b/src/vws/async_vumark_service.py index 2de1db7ba..466622f51 100644 --- a/src/vws/async_vumark_service.py +++ b/src/vws/async_vumark_service.py @@ -7,6 +7,7 @@ from beartype import BeartypeConf, beartype from vws._async_vws_request import async_target_api_request +from vws._json_utils import json_object, string_field from vws.exceptions.base_exceptions import VWSError from vws.exceptions.custom_exceptions import ServerError from vws.exceptions.vws_exceptions import TooManyRequestsError @@ -147,8 +148,11 @@ async def generate_vumark_instance( if response.status_code == HTTPStatus.OK: return response.content - result_code = json.loads(s=response.text)["result_code"] # pyrefly: ignore [unknown-variable-type] + result_code = string_field( + value=json_object(value=response.text), + name="result_code", + ) raise VWSError.from_result_code( - result_code=result_code, # pyrefly: ignore [unknown-argument-type] + result_code=result_code, response=response, ) diff --git a/src/vws/async_vws.py b/src/vws/async_vws.py index c35e55207..87a3cfc11 100644 --- a/src/vws/async_vws.py +++ b/src/vws/async_vws.py @@ -13,6 +13,7 @@ from vws._async_vws_request import async_target_api_request from vws._image_utils import ImageType as _ImageType from vws._image_utils import get_image_data as _get_image_data +from vws._json_utils import json_object, string_field, string_list_field from vws._reco_counts import ( reco_counts_report_body, reco_counts_report_path, @@ -152,12 +153,15 @@ async def make_request( ): # pragma: no cover raise ServerError(response=response) - result_code = json.loads(s=response.text)["result_code"] # pyrefly: ignore [unknown-variable-type] + result_code = string_field( + value=json_object(value=response.text), + name="result_code", + ) if result_code == expected_result_code: return response raise VWSError.from_result_code( - result_code=result_code, # pyrefly: ignore [unknown-argument-type] + result_code=result_code, response=response, ) @@ -238,7 +242,10 @@ async def add_target( content_type="application/json", ) - return str(object=json.loads(s=response.text)["target_id"]) # pyrefly: ignore [unknown-argument-type] + return string_field( + value=json_object(value=response.text), + name="target_id", + ) async def get_target_record(self, target_id: str) -> TargetStatusAndRecord: """Get a given target's target record from the Target @@ -276,7 +283,7 @@ async def get_target_record(self, target_id: str) -> TargetStatusAndRecord: content_type="application/json", ) - result_data = json.loads(s=response.text) + result_data = json_object(value=response.text) return TargetStatusAndRecord.from_response_dict( response_dict=result_data, ) @@ -373,7 +380,10 @@ async def list_targets(self) -> list[str]: content_type="application/json", ) - return list(json.loads(s=response.text)["results"]) # pyrefly: ignore [unknown-argument-type] + return string_list_field( + value=json_object(value=response.text), + name="results", + ) async def get_target_summary_report( self, target_id: str @@ -413,7 +423,7 @@ async def get_target_summary_report( content_type="application/json", ) - result_data = dict(json.loads(s=response.text)) + result_data = json_object(value=response.text) return TargetSummaryReport.from_response_dict( response_dict=result_data, ) @@ -450,7 +460,7 @@ async def get_database_summary_report( content_type="application/json", ) - response_data = dict(json.loads(s=response.text)) + response_data = json_object(value=response.text) return DatabaseSummaryReport.from_response_dict( response_dict=response_data, ) @@ -504,7 +514,7 @@ async def request_database_reco_counts_report( content_type="application/json", ) - response_data = dict(json.loads(s=response.text)) + response_data = json_object(value=response.text) return RecoCountsReportRequest.from_response_dict( response_dict=response_data, ) @@ -657,8 +667,9 @@ async def get_duplicate_targets(self, target_id: str) -> list[str]: content_type="application/json", ) - return list( - json.loads(s=response.text)["similar_targets"], # pyrefly: ignore [unknown-argument-type] + return string_list_field( + value=json_object(value=response.text), + name="similar_targets", ) async def update_target( diff --git a/src/vws/exceptions/vws_exceptions.py b/src/vws/exceptions/vws_exceptions.py index 9916dfe7f..4dc01407d 100644 --- a/src/vws/exceptions/vws_exceptions.py +++ b/src/vws/exceptions/vws_exceptions.py @@ -6,11 +6,11 @@ api#result-codes. """ -import json from urllib.parse import urlparse from beartype import beartype +from vws._json_utils import json_object, string_field from vws.exceptions.base_exceptions import VWSError @@ -147,8 +147,8 @@ def target_name(self) -> str: if not isinstance(response_body, str | bytes): # pragma: no cover msg = "A target-name error response must have a request body." raise TypeError(msg) - request_json = json.loads(s=response_body) - return str(object=request_json["name"]) # pyrefly: ignore [unknown-argument-type] + request_json = json_object(value=response_body) + return string_field(value=request_json, name="name") @beartype diff --git a/src/vws/query.py b/src/vws/query.py index f5c173355..f93690312 100644 --- a/src/vws/query.py +++ b/src/vws/query.py @@ -10,6 +10,7 @@ from vws._image_utils import ImageType as _ImageType from vws._image_utils import get_image_data as _get_image_data +from vws._json_utils import json_object, object_list_field, string_field from vws.exceptions.base_exceptions import CloudRecoError from vws.exceptions.cloud_reco_exceptions import ( AuthenticationFailureError, @@ -171,13 +172,16 @@ def query( raise CloudRecoError(response=response) try: - response_body = json.loads(s=response.text) + response_body = json_object(value=response.text) except json.JSONDecodeError as exc: if response.status_code >= HTTPStatus.BAD_REQUEST: raise CloudRecoError(response=response) from exc raise - result_code = response_body["result_code"] # pyrefly: ignore [unknown-variable-type] + result_code = string_field( + value=response_body, + name="result_code", + ) if result_code != "Success": exception = { "AuthenticationFailure": AuthenticationFailureError, @@ -187,8 +191,11 @@ def query( }[result_code] raise exception(response=response) - result_list = list(response_body["results"]) # pyrefly: ignore [unknown-argument-type] + result_list = object_list_field( + value=response_body, + name="results", + ) return [ - QueryResult.from_response_dict(response_dict=item) # pyrefly: ignore [unknown-argument-type] + QueryResult.from_response_dict(response_dict=item) for item in result_list ] diff --git a/src/vws/vumark_service.py b/src/vws/vumark_service.py index 7764c1624..8d61dee18 100644 --- a/src/vws/vumark_service.py +++ b/src/vws/vumark_service.py @@ -5,6 +5,7 @@ from beartype import BeartypeConf, beartype +from vws._json_utils import json_object, string_field from vws._vws_request import target_api_request from vws.exceptions.base_exceptions import VWSError from vws.exceptions.custom_exceptions import ServerError @@ -130,8 +131,11 @@ def generate_vumark_instance( if response.status_code == HTTPStatus.OK: return response.content - result_code = json.loads(s=response.text)["result_code"] # pyrefly: ignore [unknown-variable-type] + result_code = string_field( + value=json_object(value=response.text), + name="result_code", + ) raise VWSError.from_result_code( - result_code=result_code, # pyrefly: ignore [unknown-argument-type] + result_code=result_code, response=response, ) diff --git a/src/vws/vws.py b/src/vws/vws.py index 5e977af4a..b02268f92 100644 --- a/src/vws/vws.py +++ b/src/vws/vws.py @@ -10,6 +10,7 @@ from vws._image_utils import ImageType as _ImageType from vws._image_utils import get_image_data as _get_image_data +from vws._json_utils import json_object, string_field, string_list_field from vws._reco_counts import ( reco_counts_report_body, reco_counts_report_path, @@ -138,12 +139,15 @@ def make_request( ): # pragma: no cover raise ServerError(response=response) - result_code = json.loads(s=response.text)["result_code"] # pyrefly: ignore [unknown-variable-type] + result_code = string_field( + value=json_object(value=response.text), + name="result_code", + ) if result_code == expected_result_code: return response raise VWSError.from_result_code( - result_code=result_code, # pyrefly: ignore [unknown-argument-type] + result_code=result_code, response=response, ) @@ -224,7 +228,10 @@ def add_target( content_type="application/json", ) - return str(object=json.loads(s=response.text)["target_id"]) # pyrefly: ignore [unknown-argument-type] + return string_field( + value=json_object(value=response.text), + name="target_id", + ) def get_target_record(self, target_id: str) -> TargetStatusAndRecord: """Get a given target's target record from the Target Management @@ -262,7 +269,7 @@ def get_target_record(self, target_id: str) -> TargetStatusAndRecord: content_type="application/json", ) - result_data = json.loads(s=response.text) + result_data = json_object(value=response.text) return TargetStatusAndRecord.from_response_dict( response_dict=result_data, ) @@ -351,7 +358,10 @@ def list_targets(self) -> list[str]: content_type="application/json", ) - return list(json.loads(s=response.text)["results"]) # pyrefly: ignore [unknown-argument-type] + return string_list_field( + value=json_object(value=response.text), + name="results", + ) def get_target_summary_report(self, target_id: str) -> TargetSummaryReport: """Get a summary report for a target. @@ -388,7 +398,7 @@ def get_target_summary_report(self, target_id: str) -> TargetSummaryReport: content_type="application/json", ) - result_data = dict(json.loads(s=response.text)) + result_data = json_object(value=response.text) return TargetSummaryReport.from_response_dict( response_dict=result_data, ) @@ -423,7 +433,7 @@ def get_database_summary_report(self) -> DatabaseSummaryReport: content_type="application/json", ) - response_data = dict(json.loads(s=response.text)) + response_data = json_object(value=response.text) return DatabaseSummaryReport.from_response_dict( response_dict=response_data, ) @@ -477,7 +487,7 @@ def request_database_reco_counts_report( content_type="application/json", ) - response_data = dict(json.loads(s=response.text)) + response_data = json_object(value=response.text) return RecoCountsReportRequest.from_response_dict( response_dict=response_data, ) @@ -630,7 +640,10 @@ def get_duplicate_targets(self, target_id: str) -> list[str]: content_type="application/json", ) - return list(json.loads(s=response.text)["similar_targets"]) # pyrefly: ignore [unknown-argument-type] + return string_list_field( + value=json_object(value=response.text), + name="similar_targets", + ) def update_target( self, diff --git a/tests/test_query.py b/tests/test_query.py index b77d1385d..cb30318cc 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -2,7 +2,10 @@ import datetime import io # noqa: TC003 +import json +import secrets import uuid +from http import HTTPStatus from typing import BinaryIO import pytest @@ -13,6 +16,40 @@ from vws import VWS, CloudRecoService from vws.include_target_data import CloudRecoIncludeTargetData +from vws.response import Response + + +class _JSONResponseTransport: + """A transport which returns one JSON response body.""" + + def __init__(self, *, body: object) -> None: + """Create a transport for the given JSON body.""" + self._text = json.dumps(obj=body) + + def close(self) -> None: + """Close the transport.""" + + def __call__( + self, + *, + method: str, + url: str, + headers: dict[str, str], + data: bytes, + request_timeout: float | tuple[float, float], + ) -> Response: + """Return the configured response body.""" + del method, headers, data, request_timeout + content = self._text.encode() + return Response( + text=self._text, + url=url, + status_code=HTTPStatus.OK, + headers={"Content-Type": "application/json"}, + request_body=None, + tell_position=len(content), + content=content, + ) class TestQuery: @@ -47,6 +84,21 @@ def test_match( [matching_target] = cloud_reco_client.query(image=image) assert matching_target.target_id == target_id + @staticmethod + def test_invalid_results(*, image: io.BytesIO | BinaryIO) -> None: + """Query results in responses must be a list of objects.""" + transport = _JSONResponseTransport( + body={"result_code": "Success", "results": 1} + ) + client = CloudRecoService( + client_access_key="access-key", + client_secret_key=secrets.token_hex(), + transport=transport, + ) + + with pytest.raises(expected_exception=TypeError): + _ = client.query(image=image) + class TestDefaultRequestTimeout: """Tests for the default request timeout.""" diff --git a/tests/test_vws.py b/tests/test_vws.py index 3a4de94e9..9313455a0 100644 --- a/tests/test_vws.py +++ b/tests/test_vws.py @@ -4,6 +4,7 @@ import calendar import datetime import io # noqa: TC003 +import json import secrets import time import uuid @@ -40,6 +41,39 @@ from vws.vumark_accept import VuMarkAccept +class _JSONResponseTransport: + """A transport which returns one JSON response body.""" + + def __init__(self, *, body: object) -> None: + """Create a transport for the given JSON body.""" + self._text = json.dumps(obj=body) + + def close(self) -> None: + """Close the transport.""" + + def __call__( + self, + *, + method: str, + url: str, + headers: dict[str, str], + data: bytes, + request_timeout: float | tuple[float, float], + ) -> Response: + """Return the configured response body.""" + del method, headers, data, request_timeout + content = self._text.encode() + return Response( + text=self._text, + url=url, + status_code=HTTPStatus.OK, + headers={"Content-Type": "application/json"}, + request_body=None, + tell_position=len(content), + content=content, + ) + + class TestAddTarget: """Tests for adding a target.""" @@ -694,6 +728,47 @@ def test_get_duplicate_targets( duplicates = vws_client.get_duplicate_targets(target_id=target_id) assert duplicates == [similar_target_id] + @staticmethod + @pytest.mark.parametrize( + argnames="similar_targets", + argvalues=[1, ["target-id", 1]], + ) + def test_invalid_duplicate_target_response( + *, similar_targets: object + ) -> None: + """Duplicate target IDs in responses must be a list of strings.""" + transport = _JSONResponseTransport( + body={ + "result_code": "Success", + "similar_targets": similar_targets, + } + ) + client = VWS( + server_access_key="access-key", + server_secret_key=secrets.token_hex(), + transport=transport, + ) + + with pytest.raises(expected_exception=TypeError): + _ = client.get_duplicate_targets(target_id="target-id") + + +@pytest.mark.parametrize( + argnames="body", + argvalues=[[], {"result_code": 1}], +) +def test_invalid_vws_response_envelope(*, body: object) -> None: + """VWS responses must be objects with a string result code.""" + transport = _JSONResponseTransport(body=body) + client = VWS( + server_access_key="access-key", + server_secret_key=secrets.token_hex(), + transport=transport, + ) + + with pytest.raises(expected_exception=TypeError): + client.delete_target(target_id="target-id") + class TestUpdateTarget: """Tests for updating a target."""