Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
158 changes: 100 additions & 58 deletions src/mock_vws/_model_target_web_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
import uuid
import zipfile
from http import HTTPStatus
from typing import Any, Protocol, TypeGuard, runtime_checkable
from typing import Protocol, TypeGuard, runtime_checkable
from urllib.parse import parse_qs

from beartype import beartype
Expand Down Expand Up @@ -148,7 +148,7 @@ def remove_oauth2_client_credential(self, client_id: str) -> None:
def _json_response(
*,
status_code: HTTPStatus,
body: dict[str, Any], # pyrefly: ignore [explicit-any]
body: dict[str, JSONValue],
) -> _ResponseType:
"""Return a JSON response."""
body_json = json_dump(body=body)
Expand All @@ -172,11 +172,11 @@ def _error_response(
details: list[dict[str, str]] | None,
) -> _ResponseType:
"""Return an error response shaped like the Model Target Web API."""
error: dict[str, Any] = {"code": code, "message": message} # pyrefly: ignore [explicit-any]
error: dict[str, JSONValue] = {"code": code, "message": message}
if target is not None:
error["target"] = target
if details is not None:
error["details"] = details
error["details"] = [dict[str, JSONValue](detail) for detail in details]
return _json_response(status_code=status_code, body={"error": error})


Expand Down Expand Up @@ -208,7 +208,10 @@ def _oauth2_error_response(
body: dict[str, str],
) -> _ResponseType:
"""Return an OAuth2 error response."""
return _json_response(status_code=status_code, body=body)
return _json_response(
status_code=status_code,
body=dict[str, JSONValue](body),
)


@beartype
Expand Down Expand Up @@ -451,7 +454,7 @@ def _require_state_based_scope(
def _fake_jwt(*, token_source: bytes, scopes: frozenset[str]) -> str:
"""Return a deterministic bearer token for the mock."""

def encode_part(value: dict[str, Any]) -> str: # pyrefly: ignore [explicit-any]
def encode_part(value: dict[str, JSONValue]) -> str:
"""Return a base64url-encoded token part."""
raw_part = json.dumps(
obj=value,
Expand Down Expand Up @@ -779,7 +782,10 @@ def update_oauth2_client_credential_scopes(
)
return _json_response(
status_code=HTTPStatus.OK,
body={"clientId": client_id, "scopes": scopes},
body={
"clientId": client_id,
"scopes": list[JSONValue](scopes),
},
)


Expand Down Expand Up @@ -852,7 +858,10 @@ def _load_request_json(


@beartype
def _cad_data_source_details(*, models: list[Any]) -> list[dict[str, str]]: # pyrefly: ignore [explicit-any]
def _cad_data_source_details(
*,
models: list[dict[str, JSONValue]],
) -> list[dict[str, str]]:
"""Return validation details for each model's CAD data source.

One and only one of ``cadDataUrl``, ``cadDataBlob`` and ``cadDataUuid``
Expand Down Expand Up @@ -887,7 +896,7 @@ def _cad_data_source_details(*, models: list[Any]) -> list[dict[str, str]]: # p
@beartype
def _model_field_details(
*,
models: list[Any], # pyrefly: ignore [explicit-any]
models: list[dict[str, JSONValue]],
dataset_type: ModelTargetDatasetType,
) -> list[dict[str, str]]:
"""Return validation details for the fields of each model."""
Expand Down Expand Up @@ -938,18 +947,17 @@ def _model_field_details(
for field, allowed_values in sorted(enum_field_values.items()):
if field in {"motionHint", "trackingMode"}:
continue
if field not in model or model[field] in allowed_values: # pyrefly: ignore [unknown-argument-type]
value = model.get(field)
if not isinstance(value, str) or value in allowed_values:
continue
value = model[field] # pyrefly: ignore [unknown-variable-type]
messages = {
"automaticColoring": (
"invalid automaticColoring. Should be one of 'never', "
f"'always', 'auto'. You provided '{value}'"
),
"cadDataFormat": (
# pyrefly: ignore [unknown-argument-type]
"Unrecognized cadDataFormat '"
f"{str(object=value).upper()}'. "
f"{value.upper()}'. "
"Allowed values are: ZIP, GLB, DRC_GLB, DRC_GLTF, DAE, "
"FBX, IGES, OBJ, PVS, PVZ, STL, VRML, or specify no "
"cadDataFormat to auto-detect GLB and zipped glTFs."
Expand Down Expand Up @@ -997,27 +1005,24 @@ def _model_field_details(


@beartype
def _view_details(*, models: list[Any]) -> list[dict[str, str]]: # pyrefly: ignore [explicit-any]
def _view_details(
*,
models: list[dict[str, JSONValue]],
) -> list[dict[str, str]]:
"""Return validation details for the guide views of each model."""
views = [
views: list[tuple[int, int, JSONValue]] = [
(model_index, view_index, view)
for model_index, model in enumerate(iterable=models)
for view_index, view in enumerate(iterable=model.get("views", []))
for views_value in (model.get("views"),)
if isinstance(views_value, list)
for view_index, view in enumerate(iterable=views_value)
]

object_details = [
{
"code": "VALIDATION_ERROR",
"message": (
f"/models({model_index})/views({view_index}): "
"error.expected.jsobject"
),
}
object_views: list[tuple[int, int, dict[str, JSONValue]]] = [
(model_index, view_index, view)
for model_index, view_index, view in views
if not isinstance(view, dict)
if isinstance(view, dict)
]
if bool(object_details):
return object_details

missing_details = [
{
Expand All @@ -1027,7 +1032,7 @@ def _view_details(*, models: list[Any]) -> list[dict[str, str]]: # pyrefly: ign
"element is required"
),
}
for model_index, view_index, view in views
for model_index, view_index, view in object_views
for field in ("name",)
if field not in view
]
Expand All @@ -1042,7 +1047,7 @@ def _view_details(*, models: list[Any]) -> list[dict[str, str]]: # pyrefly: ign
"error.expected.jsstring"
),
}
for model_index, view_index, view in views
for model_index, view_index, view in object_views
if not isinstance(view["name"], str)
]
position_details = [
Expand All @@ -1053,7 +1058,7 @@ def _view_details(*, models: list[Any]) -> list[dict[str, str]]: # pyrefly: ign
"/guideViewPosition: error.expected.jsobject"
),
}
for model_index, view_index, view in views
for model_index, view_index, view in object_views
if "guideViewPosition" in view
and not isinstance(view["guideViewPosition"], dict)
]
Expand All @@ -1073,14 +1078,18 @@ def _is_json_number(*, value: object) -> bool:
@beartype
def _guide_view_position_details(
*,
models: list[Any], # pyrefly: ignore [explicit-any]
models: list[dict[str, JSONValue]],
) -> list[dict[str, str]]:
"""Return validation details for the guide view positions."""
positions = [
(model_index, view_index, view["guideViewPosition"])
positions: list[tuple[int, int, dict[str, JSONValue]]] = [
(model_index, view_index, position)
for model_index, model in enumerate(iterable=models)
for view_index, view in enumerate(iterable=model.get("views", []))
if "guideViewPosition" in view
for views_value in (model.get("views"),)
if isinstance(views_value, list)
for view_index, view in enumerate(iterable=views_value)
if isinstance(view, dict)
for position in (view.get("guideViewPosition"),)
if isinstance(position, dict)
]

missing_details = [
Expand Down Expand Up @@ -1124,8 +1133,10 @@ def _guide_view_position_details(
}
for model_index, view_index, position in positions
for field in ("rotation", "translation")
for element_index, element in enumerate(iterable=position[field])
if not _is_json_number(value=element) # pyrefly: ignore [unknown-argument-type]
for elements in (position[field],)
if isinstance(elements, list)
for element_index, element in enumerate(iterable=elements)
if not _is_json_number(value=element)
]


Expand All @@ -1137,7 +1148,7 @@ def _configuration_states(
) -> tuple[frozenset[str] | None, dict[str, str] | None]:
"""Load the state names from a State-Based Model Target config."""
try:
configuration: Any = json.loads(s=configuration_string) # pyrefly: ignore [explicit-any]
configuration: object = json.loads(s=configuration_string)
except json.JSONDecodeError:
return None, {
"code": "VALIDATION_ERROR",
Expand Down Expand Up @@ -1168,12 +1179,18 @@ def _configuration_states(


@beartype
def _state_based_details(*, models: list[Any]) -> list[dict[str, str]]: # pyrefly: ignore [explicit-any]
def _state_based_details(
*,
models: list[dict[str, JSONValue]],
) -> list[dict[str, str]]:
"""Return validation details for State-Based Model Targets."""
state_fields = [
(model_index, view_index, view["states"])
state_fields: list[tuple[int, int, dict[str, JSONValue], JSONValue]] = [
(model_index, view_index, view, view["states"])
for model_index, model in enumerate(iterable=models)
for view_index, view in enumerate(iterable=model.get("views", []))
for views_value in (model.get("views"),)
if isinstance(views_value, list)
for view_index, view in enumerate(iterable=views_value)
if isinstance(view, dict)
if "states" in view
]
array_details = [
Expand All @@ -1184,12 +1201,20 @@ def _state_based_details(*, models: list[Any]) -> list[dict[str, str]]: # pyref
"error.expected.jsarray"
),
}
for model_index, view_index, states in state_fields
for model_index, view_index, _view, states in state_fields
if not isinstance(states, list)
]
if bool(array_details):
return array_details

state_lists: list[
tuple[int, int, dict[str, JSONValue], list[JSONValue]]
] = [
(model_index, view_index, view, states)
for model_index, view_index, view, states in state_fields
if isinstance(states, list)
]

element_details = [
{
"code": "VALIDATION_ERROR",
Expand All @@ -1198,17 +1223,28 @@ def _state_based_details(*, models: list[Any]) -> list[dict[str, str]]: # pyref
f"({state_index}): error.expected.jsstring"
),
}
for model_index, view_index, states in state_fields
for model_index, view_index, _view, states in state_lists
for state_index, state in enumerate(iterable=states)
if not isinstance(state, str)
]
if bool(element_details):
return element_details

string_state_lists: list[tuple[int, str, list[str]]] = [
(
model_index,
name,
[state for state in states if isinstance(state, str)],
)
for model_index, _view_index, view, states in state_lists
for name in (view.get("name"),)
if isinstance(name, str)
]

details: list[dict[str, str]] = []
configured_states: dict[int, frozenset[str]] = {}
for model_index, model in enumerate(iterable=models):
configuration_string = model.get("stateBasedConfigurationJsonString") # pyrefly: ignore [unknown-variable-type]
configuration_string = model.get("stateBasedConfigurationJsonString")
if not isinstance(configuration_string, str):
continue
state_names, detail = _configuration_states(
Expand All @@ -1223,7 +1259,7 @@ def _state_based_details(*, models: list[Any]) -> list[dict[str, str]]: # pyref
if bool(details):
return details

for model_index, view_index, states in state_fields:
for model_index, name, states in string_state_lists:
if model_index not in configured_states:
details.append(
{
Expand All @@ -1241,12 +1277,12 @@ def _state_based_details(*, models: list[Any]) -> list[dict[str, str]]: # pyref
"code": "VALIDATION_ERROR",
"message": (
"states in entrypoint "
f"{models[model_index]['views'][view_index]['name']}' "
f"{name}' "
"must be a subset of all states"
),
}
for state in states
if state not in configured_states[model_index] # pyrefly: ignore [unknown-argument-type]
if state not in configured_states[model_index]
)

return details
Expand All @@ -1255,7 +1291,7 @@ def _state_based_details(*, models: list[Any]) -> list[dict[str, str]]: # pyref
@beartype
def _model_count_details(
*,
models: list[Any], # pyrefly: ignore [explicit-any]
models: list[dict[str, JSONValue]],
dataset_type: ModelTargetDatasetType,
) -> list[dict[str, str]]:
"""Return validation details for the number of models."""
Expand Down Expand Up @@ -1306,7 +1342,7 @@ def _model_count_details(
@beartype
def _top_level_details(
*,
request_json: dict[str, Any], # pyrefly: ignore [explicit-any]
request_json: dict[str, JSONValue],
) -> list[dict[str, str]]:
"""Return validation details for the top-level dataset fields."""
missing_details = [
Expand Down Expand Up @@ -1345,23 +1381,29 @@ def _top_level_details(
@beartype
def _validate_dataset_request(
*,
request_json: dict[str, Any], # pyrefly: ignore [explicit-any]
request_json: dict[str, JSONValue],
dataset_type: ModelTargetDatasetType,
) -> _ResponseType | None:
"""Validate the dataset request enough for useful mock feedback."""
details = _top_level_details(request_json=request_json)
if not bool(details):
# Vuforia's schema validator reads fields from non-object model and
# view values as though they were empty objects.
models: list[Any] = [ # pyrefly: ignore [explicit-any]
model if isinstance(model, dict) else {}
for model in request_json["models"]
]
models_value = request_json["models"]
models: list[dict[str, JSONValue]] = (
[
model if isinstance(model, dict) else {}
for model in models_value
]
if isinstance(models_value, list)
else []
)
for model in models:
if isinstance(model.get("views"), list):
views_value = model.get("views")
if isinstance(views_value, list):
model["views"] = [
view if isinstance(view, dict) else dict[str, Any]()
for view in model["views"]
view if isinstance(view, dict) else dict[str, JSONValue]()
for view in views_value
]
details = _model_field_details(
models=models, dataset_type=dataset_type
Expand Down
16 changes: 0 additions & 16 deletions tests/mock_vws/test_model_target_web_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -2633,22 +2633,6 @@ def test_dataset_scope_and_shape_errors() -> None:
status_codes=HTTPStatus.BAD_REQUEST,
)

@staticmethod
def test_view_helper_rejects_non_objects() -> None:
"""The view validator reports view values which are not
objects.
"""
# pylint: disable=protected-access
details = _model_target_web_api._view_details( # noqa: SLF001
models=[{"views": [1]}],
)
assert details == [
{
"code": "VALIDATION_ERROR",
"message": "/models(0)/views(0): error.expected.jsobject",
},
]

@staticmethod
def test_target_manager_missing_credential_delete() -> None:
"""The internal target manager returns 404 for an unknown
Expand Down
Loading