From d6ed40f796ab0a48457736ab5ac705e48955f2ca Mon Sep 17 00:00:00 2001 From: Tom Hammel Date: Mon, 31 Aug 2026 15:47:53 +0200 Subject: [PATCH] server: Implement the read-only `$value` routes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit All `$value` routes previously responded with `501 Not Implemented`. The specification leaves some points open, which are resolved as follows: - Property values are mapped to the JSON type of their `valueType`, so booleans and numbers are not quoted. Values without a JSON literal are returned as their XSD representation. `xs:integer` is unbounded in both Python and JSON and is therefore passed on as it is. - Elements without a value are omitted from Submodels, collections and Entity statements, but kept as `null` within a SubmodelElementList, as omitting them would shift the indices of the remaining elements. - Operations and Capabilities have no ValueOnly representation. They are skipped in containers and result in `400 Bad Request` when requested directly, which is how the `$metadata` route handles them. - `level=core` returns the direct children of the requested resource and their nested containers as empty, matching the examples of the specification. - The entries of `GET /submodels/{id}/submodel-elements/$value` are mapped by idShort, as a bare `SubmodelElementValue` would lose the association between a value and its element. Submodels are returned as-is, since their entries are already named. - The languages of a MultiLanguageProperty are sorted, to keep the order of the response deterministic. The ValueOnly serialization is only defined for JSON, so the routes respond with `406 Not Acceptable` if XML is requested. `extent` remains unimplemented, as on all other routes. `PATCH …/$value`, the `$path` routes and the operation invocation routes are not part of this change. Fixes #446 --- server/README.md | 4 +- server/app/adapter/__init__.py | 1 + server/app/adapter/value_only.py | 223 +++++++++++ server/app/interfaces/repository.py | 85 ++++- .../repository/test_submodel_value.py | 350 ++++++++++++++++++ 5 files changed, 656 insertions(+), 7 deletions(-) create mode 100644 server/app/adapter/value_only.py create mode 100644 server/test/interfaces/repository/test_submodel_value.py diff --git a/server/README.md b/server/README.md index c09d565a1..4bf0a61df 100644 --- a/server/README.md +++ b/server/README.md @@ -130,8 +130,8 @@ Several features and routes are currently not supported: - `/description` These routes are not implemented at this time. -5. Value, Path, and PATCH Routes: - - All `/…/value$`, `/…/path$`, and `PATCH` routes are currently not implemented. +5. Path and PATCH Routes: + - All `/…/$path` and `PATCH` routes are currently not implemented. 6. Operation Invocation Routes: The following routes are not implemented because operation invocation is not yet supported by the `basyx-python-sdk`: diff --git a/server/app/adapter/__init__.py b/server/app/adapter/__init__.py index 10c99aa69..8a3e2af17 100644 --- a/server/app/adapter/__init__.py +++ b/server/app/adapter/__init__.py @@ -1 +1,2 @@ from .jsonization import * +from .value_only import * diff --git a/server/app/adapter/value_only.py b/server/app/adapter/value_only.py new file mode 100644 index 000000000..8eeb61a7a --- /dev/null +++ b/server/app/adapter/value_only.py @@ -0,0 +1,223 @@ +# Copyright (c) 2026 the Eclipse BaSyx Authors +# +# This program and the accompanying materials are made available under the terms of the MIT License, available in +# the LICENSE file of this project. +# +# SPDX-License-Identifier: MIT +""" +ValueOnly serialization of Submodels and SubmodelElements, as defined in "Specification of the Asset Administration +Shell Part 1", section "Value-Only Serialization in JSON". + +The ValueOnly serialization cannot be derived from the JSON schema of the metamodel, since it defines individual rules +per class. Therefore, it is implemented by hand here, on top of the object model of the ``basyx-python-sdk``. + +The returned structures are JSON-serializable with :class:`~basyx.aas.adapter.json.AASToJsonEncoder` (or any subclass, +e.g. the ``ResultToJsonEncoder`` of this server): :class:`~basyx.aas.model.base.Reference` and +:class:`~basyx.aas.model.base.SpecificAssetId` objects are passed through as-is instead of being converted here, to +avoid duplicating the serialization rules of the SDK. +""" + +import base64 +import decimal +import math +from typing import Any, Dict, Iterable, List, Optional + +from basyx.aas import model +from basyx.aas.adapter._generic import ENTITY_TYPES + +__all__ = [ + "has_value_only_representation", + "submodel_element_to_named_value_only", + "submodel_element_to_value_only", + "submodel_to_value_only", +] + +# SubmodelElement types that don't have a ValueOnly representation. Operations only have one in the context of their +# invocation, which is not supported by this server. +_TYPES_WITHOUT_VALUE_ONLY = (model.Operation, model.Capability) + + +def has_value_only_representation(element: model.SubmodelElement) -> bool: + """ + Whether a SubmodelElement has a ValueOnly representation at all. + + :param element: The SubmodelElement to check + :return: ``False`` for Operations and Capabilities, ``True`` otherwise + """ + return not isinstance(element, _TYPES_WITHOUT_VALUE_ONLY) + + +def submodel_to_value_only(submodel: model.Submodel, deep: bool = True) -> Dict[str, Any]: + """ + Serialize a Submodel in its ValueOnly representation. + + :param submodel: The Submodel to serialize + :param deep: If ``False`` (i.e. ``level=core``), only the first level of SubmodelElements is serialized and nested + containers are returned empty + :return: A dictionary, mapping the idShort of each SubmodelElement to its value + """ + return _namespace_to_value_only(submodel.submodel_element, _initial_depth(deep)) + + +def submodel_element_to_value_only(element: model.SubmodelElement, deep: bool = True) -> Any: + """ + Serialize a SubmodelElement in its ValueOnly representation. + + :param element: The SubmodelElement to serialize + :param deep: If ``False`` (i.e. ``level=core``), only the first level of child elements is serialized and nested + containers are returned empty + :raises TypeError: If the given element doesn't have a ValueOnly representation, see + :func:`has_value_only_representation` + :return: The value of the element, ``None`` if it doesn't have one + """ + return _element_to_value_only(element, _initial_depth(deep)) + + +def _initial_depth(deep: bool) -> float: + # The number of container levels that are serialized below the requested resource. level=core includes the direct + # children of the requested resource, but returns their children as empty containers. + return math.inf if deep else 1 + + +def _namespace_to_value_only(elements: Iterable[model.SubmodelElement], depth: float) -> Dict[str, Any]: + data: Dict[str, Any] = {} + if depth <= 0: + return data + for element in elements: + if not has_value_only_representation(element): + continue + value = _element_to_value_only(element, depth - 1) + # Elements without a value are omitted + if value is None: + continue + # The idShort is always set for elements that aren't contained in a SubmodelElementList + assert element.id_short is not None + data[element.id_short] = value + return data + + +def _element_to_value_only(element: model.SubmodelElement, depth: float) -> Any: + if isinstance(element, model.Property): + return _typed_value_to_value_only(element.value, element.value_type) + if isinstance(element, model.MultiLanguageProperty): + return _lang_string_set_to_value_only(element.value) + if isinstance(element, model.Range): + return _range_to_value_only(element) + if isinstance(element, model.Blob): + return { + "contentType": element.content_type, + "value": None if element.value is None else base64.b64encode(element.value).decode(), + } + if isinstance(element, model.File): + return {"contentType": element.content_type, "value": element.value} + if isinstance(element, model.ReferenceElement): + return element.value + # AnnotatedRelationshipElement is a subclass of RelationshipElement and must therefore be checked first + if isinstance(element, model.AnnotatedRelationshipElement): + return { + "first": element.first, + "second": element.second, + "annotations": _annotations_to_value_only(element.annotation, depth), + } + if isinstance(element, model.RelationshipElement): + return {"first": element.first, "second": element.second} + if isinstance(element, model.Entity): + return _entity_to_value_only(element, depth) + if isinstance(element, model.BasicEventElement): + return {"observed": element.observed} + if isinstance(element, model.SubmodelElementList): + return _submodel_element_list_to_value_only(element, depth) + if isinstance(element, model.SubmodelElementCollection): + return _namespace_to_value_only(element.value, depth) + raise TypeError(f"{element!r} doesn't have a ValueOnly representation!") + + +def _typed_value_to_value_only(value: Optional[model.ValueDataType], value_type: model.DataTypeDefXsd) -> Any: + """ + Map a typed value to its JSON representation: booleans become JSON booleans, numeric types become JSON numbers and + everything else becomes a JSON string containing the XSD representation of the value. + """ + if value is None: + return None + if value_type is model.datatypes.Boolean or isinstance(value, bool): + return bool(value) + if isinstance(value, int): + # int subclasses are serialized as JSON numbers as-is. xs:integer is unbounded, so no conversion to float may + # happen here, as that would raise an OverflowError for large values. + return value + if isinstance(value, (float, decimal.Decimal)): + # JSON doesn't support NaN, INF and -INF. The specification acknowledges this gap without resolving it, so the + # XSD representation is returned as a string instead of emitting invalid JSON. This also catches values of + # decimal.Decimal that are finite, but too large to be represented as a float. + if not math.isfinite(float(value)): + return model.datatypes.xsd_repr(value) + # decimal.Decimal cannot be serialized as JSON and may lose precision here + return float(value) if isinstance(value, decimal.Decimal) else value + return model.datatypes.xsd_repr(value) + + +def _lang_string_set_to_value_only(value: Optional[model.MultiLanguageTextType]) -> List[Dict[str, str]]: + if value is None: + return [] + # Sorted by language, to keep the order of the languages deterministic across requests + return [{language: text} for language, text in sorted(value.items())] + + +def _range_to_value_only(element: model.Range) -> Dict[str, Any]: + data: Dict[str, Any] = {} + if element.min is not None: + data["min"] = _typed_value_to_value_only(element.min, element.value_type) + if element.max is not None: + data["max"] = _typed_value_to_value_only(element.max, element.value_type) + return data + + +def _annotations_to_value_only(annotations: Iterable[model.DataElement], depth: float) -> List[Dict[str, Any]]: + if depth <= 0: + return [] + return [ + {annotation.id_short: _element_to_value_only(annotation, depth - 1)} + for annotation in annotations + if annotation.id_short is not None + ] + + +def _entity_to_value_only(element: model.Entity, depth: float) -> Dict[str, Any]: + data: Dict[str, Any] = {"statements": _namespace_to_value_only(element.statement, depth)} + # entityType is mandatory in the specification, but optional in the object model of the SDK + if element.entity_type is not None: + data["entityType"] = ENTITY_TYPES[element.entity_type] + if element.global_asset_id is not None: + data["globalAssetId"] = element.global_asset_id + if len(element.specific_asset_id) > 0: + data["specificAssetIds"] = list(element.specific_asset_id) + return data + + +def _submodel_element_list_to_value_only(element: model.SubmodelElementList, depth: float) -> List[Any]: + if depth <= 0: + return [] + # In contrast to the other containers, elements without a value are not omitted here, to keep the indices of the + # remaining elements intact + return [ + _element_to_value_only(child, depth - 1) if has_value_only_representation(child) else None + for child in element.value + ] + + +def submodel_element_to_named_value_only(element: model.SubmodelElement, deep: bool = True) -> Dict[str, Any]: + """ + Serialize a SubmodelElement as a single-entry dictionary, mapping its idShort to its ValueOnly representation. + + This is the representation used for the entries of ``GET /submodels/{submodelIdentifier}/submodel-elements/$value``. + The specification leaves the shape of these entries open: a bare ``SubmodelElementValue`` would lose the + association between a value and the element it belongs to, hence the elements are named here, just like they are + within the ValueOnly representation of their parent. + + :param element: The SubmodelElement to serialize + :param deep: If ``False`` (i.e. ``level=core``), only the first level of child elements is serialized + :return: A dictionary with a single entry, mapping the idShort of the given element to its value + """ + # The idShort is always set for elements that aren't contained in a SubmodelElementList + assert element.id_short is not None + return {element.id_short: submodel_element_to_value_only(element, deep)} diff --git a/server/app/interfaces/repository.py b/server/app/interfaces/repository.py index fc7b4b0a0..57d2bd7cb 100644 --- a/server/app/interfaces/repository.py +++ b/server/app/interfaces/repository.py @@ -23,11 +23,17 @@ from werkzeug.routing import MapAdapter, Rule, Submount from app._config import API_BASE_PATH +from app.adapter import ( + has_value_only_representation, + submodel_element_to_named_value_only, + submodel_element_to_value_only, + submodel_to_value_only, +) from app.interfaces.base import PagingMetadata from app.model import ServiceDescription, ServiceSpecificationProfileEnum from app.util.converters import IdentifierToBase64URLConverter, IdShortPathConverter, base64url_decode -from .base import APIResponse, HTTPApiDecoder, ObjectStoreWSGIApp, T, is_stripped_request +from .base import APIResponse, HTTPApiDecoder, JsonResponse, ObjectStoreWSGIApp, T, is_stripped_request SUPPORTED_PROFILES: ServiceDescription = ServiceDescription( [ @@ -123,7 +129,7 @@ def __init__( [ Rule("/$metadata", methods=["GET"], endpoint=self.get_submodel_all_metadata), Rule("/$reference", methods=["GET"], endpoint=self.get_submodel_all_reference), - Rule("/$value", methods=["GET"], endpoint=self.not_implemented), + Rule("/$value", methods=["GET"], endpoint=self.get_submodel_all_value), Rule("/$path", methods=["GET"], endpoint=self.not_implemented), Rule("/", methods=["GET"], endpoint=self.get_submodel), Rule("/", methods=["PUT"], endpoint=self.put_submodel), @@ -134,7 +140,7 @@ def __init__( [ Rule("/$metadata", methods=["GET"], endpoint=self.get_submodels_metadata), Rule("/$metadata", methods=["PATCH"], endpoint=self.not_implemented), - Rule("/$value", methods=["GET"], endpoint=self.not_implemented), + Rule("/$value", methods=["GET"], endpoint=self.get_submodels_value), Rule("/$value", methods=["PATCH"], endpoint=self.not_implemented), Rule("/$reference", methods=["GET"], endpoint=self.get_submodels_reference), Rule("/$path", methods=["GET"], endpoint=self.not_implemented), @@ -161,7 +167,11 @@ def __init__( methods=["GET"], endpoint=self.get_submodel_submodel_elements_reference, ), - Rule("/$value", methods=["GET"], endpoint=self.not_implemented), + Rule( + "/$value", + methods=["GET"], + endpoint=self.get_submodel_submodel_elements_value, + ), Rule("/$path", methods=["GET"], endpoint=self.not_implemented), Rule( "/", @@ -206,7 +216,11 @@ def __init__( methods=["GET"], endpoint=self.get_submodel_submodel_elements_id_short_path_reference, ), - Rule("/$value", methods=["GET"], endpoint=self.not_implemented), + Rule( + "/$value", + methods=["GET"], + endpoint=self.get_submodel_submodel_elements_id_short_path_value, + ), Rule( "/$value", methods=["PATCH"], endpoint=self.not_implemented ), @@ -520,6 +534,29 @@ def _get_submodel_submodel_elements_id_short_path(self, url_args: Dict) -> model def _get_concept_description(self, url_args): return self._get_obj_ts(url_args["concept_id"], model.ConceptDescription) + @classmethod + def _assert_json_response(cls, response_t: Type[APIResponse]) -> None: + """ + The ValueOnly serialization is only defined for JSON, thus the ``$value`` routes cannot serve XML. + """ + if not issubclass(response_t, JsonResponse): + raise werkzeug.exceptions.NotAcceptable( + "The ValueOnly serialization is only available as application/json!" + ) + + @classmethod + def _assert_value_only_representation(cls, element: model.SubmodelElement) -> None: + if not has_value_only_representation(element): + raise BadRequest(f"{element.id_short} does not allow the content modifier value!") + + @classmethod + def _value_only_response(cls, response_t: Type[APIResponse], value: object) -> Response: + if value is None: + # An element without a value has `null` as its ValueOnly representation. This cannot be passed to the + # response classes, as they interpret `None` as an empty body and respond with `204 No Content`. + return Response(json.dumps(None), content_type="application/json") + return response_t(value) + # ------ all not implemented ROUTES ------- def not_implemented(self, request: Request, url_args: Dict, **_kwargs) -> Response: raise werkzeug.exceptions.NotImplemented("This route is not implemented!") @@ -688,6 +725,15 @@ def get_submodel_all_metadata( submodels, paging_metadata = self._get_submodels(request) return response_t(list(submodels), paging_metadata=paging_metadata, stripped=True) + def get_submodel_all_value( + self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs + ) -> Response: + self._assert_json_response(response_t) + deep = not is_stripped_request(request) + submodels, paging_metadata = self._get_submodels(request) + values = [submodel_to_value_only(submodel, deep) for submodel in submodels] + return response_t(values, paging_metadata=paging_metadata) + def get_submodel_all_reference( self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs ) -> Response: @@ -715,6 +761,13 @@ def get_submodels_metadata( submodel = self._get_submodel(url_args) return response_t(submodel, stripped=True) + def get_submodels_value( + self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs + ) -> Response: + self._assert_json_response(response_t) + submodel = self._get_submodel(url_args) + return response_t(submodel_to_value_only(submodel, not is_stripped_request(request))) + def get_submodels_reference( self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs ) -> Response: @@ -744,6 +797,19 @@ def get_submodel_submodel_elements_metadata( submodel_elements, paging_metadata = self._get_submodel_submodel_elements(request, url_args) return response_t(list(submodel_elements), paging_metadata=paging_metadata, stripped=True) + def get_submodel_submodel_elements_value( + self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs + ) -> Response: + self._assert_json_response(response_t) + deep = not is_stripped_request(request) + submodel_elements, paging_metadata = self._get_submodel_submodel_elements(request, url_args) + values = [ + submodel_element_to_named_value_only(submodel_element, deep) + for submodel_element in submodel_elements + if has_value_only_representation(submodel_element) + ] + return response_t(values, paging_metadata=paging_metadata) + def get_submodel_submodel_elements_reference( self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs ) -> Response: @@ -769,6 +835,15 @@ def get_submodel_submodel_elements_id_short_path_metadata( raise BadRequest(f"{submodel_element.id_short} does not allow the content modifier metadata!") return response_t(submodel_element, stripped=True) + def get_submodel_submodel_elements_id_short_path_value( + self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs + ) -> Response: + self._assert_json_response(response_t) + submodel_element = self._get_submodel_submodel_elements_id_short_path(url_args) + self._assert_value_only_representation(submodel_element) + value = submodel_element_to_value_only(submodel_element, not is_stripped_request(request)) + return self._value_only_response(response_t, value) + def get_submodel_submodel_elements_id_short_path_reference( self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs ) -> Response: diff --git a/server/test/interfaces/repository/test_submodel_value.py b/server/test/interfaces/repository/test_submodel_value.py new file mode 100644 index 000000000..2e37b52a9 --- /dev/null +++ b/server/test/interfaces/repository/test_submodel_value.py @@ -0,0 +1,350 @@ +# Copyright (c) 2026 the Eclipse BaSyx Authors +# +# This program and the accompanying materials are made available under the terms of the MIT License, available in +# the LICENSE file of this project. +# +# SPDX-License-Identifier: MIT + +import base64 +import math + +from app.util.converters import base64url_encode +from basyx.aas import model +from basyx.aas.examples.data.example_aas import create_full_example + +from ..format_utils import ( + FormatClient, + inject_format_clients, + with_json_client, + with_xml_client, +) +from .helpers import RepositoryEndpointTestBase + +TEST_SUBMODEL_ID = "https://example.org/Test_Submodel" +IDENTIFICATION_SUBMODEL_ID = "http://example.org/Submodels/Assets/TestAsset/Identification" +BILL_OF_MATERIAL_SUBMODEL_ID = "http://example.org/Submodels/Assets/TestAsset/BillOfMaterial" +TYPED_SUBMODEL_ID = "http://example.org/Typed_Submodel" + + +def _create_typed_submodel() -> model.Submodel: + """ + A Submodel containing the value types and edge cases that aren't part of the example data. + """ + return model.Submodel( + id_=TYPED_SUBMODEL_ID, + submodel_element=( + model.Property(id_short="IntProperty", value_type=model.datatypes.Int, value=42), + model.Property(id_short="BoolProperty", value_type=model.datatypes.Boolean, value=False), + model.Property(id_short="DoubleProperty", value_type=model.datatypes.Double, value=1.5), + model.Property(id_short="NanProperty", value_type=model.datatypes.Double, value=math.nan), + # xs:integer is unbounded, this value cannot be represented as a float + model.Property(id_short="HugeIntProperty", value_type=model.datatypes.Integer, value=10**400), + model.Property( + id_short="DateProperty", value_type=model.datatypes.Date, value=model.datatypes.Date(2026, 8, 31) + ), + model.Property(id_short="EmptyProperty", value_type=model.datatypes.String, value=None), + model.SubmodelElementList( + id_short="ListWithEmptyElement", + type_value_list_element=model.Property, + value_type_list_element=model.datatypes.String, + value=( + model.Property(id_short=None, value_type=model.datatypes.String, value="first"), + model.Property(id_short=None, value_type=model.datatypes.String, value=None), + model.Property(id_short=None, value_type=model.datatypes.String, value="third"), + ), + ), + ), + ) + + +@inject_format_clients +class SubmodelValueEndpointsTest(RepositoryEndpointTestBase): + """ + Endpoint tests for the ``$value`` routes of the ``/submodels`` subtree of + :class:`~app.interfaces.repository.WSGIApp`. + + The ValueOnly serialization is only defined for JSON. Therefore, the tests are generated for the + :class:`~..format_utils.JsonFormatClient` only, except for the ones checking that XML is rejected. + """ + + def setUp(self) -> None: + super().setUp() + self.object_store.update(create_full_example()) + self.object_store.add(_create_typed_submodel()) + + @staticmethod + def submodel_path(submodel_id: str) -> str: + return f"/submodels/{base64url_encode(submodel_id)}" + + @classmethod + def elements_path(cls, submodel_id: str, id_short_path: str = "") -> str: + base = f"{cls.submodel_path(submodel_id)}/submodel-elements" + return f"{base}/{id_short_path}" if id_short_path else base + + def get_value(self, format_client: FormatClient, path: str): + response = format_client.get(path) + self.assert_ok(response) + return format_client.parse_object(response) + + # ------------------------------------------------------------------ GET /submodels/$value + + @with_json_client + def test_submodel_all_value_get(self, format_client: FormatClient): + response = format_client.get("/submodels/$value") + + self.assert_ok(response) + self.assertIn( + {"ManufacturerName": "ACPLT", "InstanceId": "978-8234-234-342"}, + format_client.parse_collection(response), + ) + + @with_json_client + def test_submodel_all_value_get_pagination_limit(self, format_client: FormatClient): + response = format_client.get("/submodels/$value?limit=1") + + self.assert_ok(response) + self.assertEqual(1, len(format_client.parse_collection(response))) + self.assertEqual("2", format_client.next_cursor(response)) + + # ------------------------------------------------------------------ GET /submodels//$value + + @with_json_client + def test_submodel_value_get(self, format_client: FormatClient): + value = self.get_value(format_client, f"{self.submodel_path(IDENTIFICATION_SUBMODEL_ID)}/$value") + + self.assertEqual({"ManufacturerName": "ACPLT", "InstanceId": "978-8234-234-342"}, value) + + @with_json_client + def test_submodel_value_get_omits_operations_and_capabilities(self, format_client: FormatClient): + value = self.get_value(format_client, f"{self.submodel_path(TEST_SUBMODEL_ID)}/$value") + + self.assertNotIn("ExampleOperation", value) + self.assertNotIn("ExampleCapability", value) + + @with_json_client + def test_submodel_value_get_level_core_empties_nested_containers(self, format_client: FormatClient): + value = self.get_value(format_client, f"{self.submodel_path(TEST_SUBMODEL_ID)}/$value?level=core") + + # the direct children are present, their children are not + self.assertIn("ExampleSubmodelCollection", value) + self.assertEqual({}, value["ExampleSubmodelCollection"]) + self.assertEqual([], value["ExampleAnnotatedRelationshipElement"]["annotations"]) + + @with_json_client + def test_submodel_value_get_huge_integer(self, format_client: FormatClient): + # integers that exceed the range of a float are also serialized when the whole Submodel is requested + value = self.get_value(format_client, f"{self.submodel_path(TYPED_SUBMODEL_ID)}/$value") + + self.assertEqual(10**400, value["HugeIntProperty"]) + self.assertEqual(42, value["IntProperty"]) + + @with_json_client + def test_submodel_value_get_omits_property_without_value(self, format_client: FormatClient): + value = self.get_value(format_client, f"{self.submodel_path(TYPED_SUBMODEL_ID)}/$value") + + self.assertNotIn("EmptyProperty", value) + + @with_json_client + def test_submodel_value_get_empty_submodel(self, format_client: FormatClient): + empty_submodel_id = "http://example.org/Empty_Submodel" + self.object_store.add(model.Submodel(id_=empty_submodel_id)) + + value = self.get_value(format_client, f"{self.submodel_path(empty_submodel_id)}/$value") + + self.assertEqual({}, value) + + @with_json_client + def test_submodel_value_get_not_found(self, format_client: FormatClient): + response = format_client.get(f"{self.submodel_path('https://example.org/unknown')}/$value") + + self.assert_error(response, 404) + + @with_json_client + def test_submodel_value_get_invalid_level_returns_400(self, format_client: FormatClient): + response = format_client.get(f"{self.submodel_path(TEST_SUBMODEL_ID)}/$value?level=bogus") + + self.assert_error(response, 400) + + @with_json_client + def test_submodel_value_get_extent_not_implemented(self, format_client: FormatClient): + response = format_client.get(f"{self.submodel_path(TEST_SUBMODEL_ID)}/$value?extent=withBlobValue") + + self.assert_error(response, 501) + + # ------------------------------------------------------------------ GET .../submodel-elements/$value + + @with_json_client + def test_submodel_elements_value_get(self, format_client: FormatClient): + response = format_client.get(f"{self.elements_path(IDENTIFICATION_SUBMODEL_ID)}/$value") + + self.assert_ok(response) + self.assertEqual( + [{"ManufacturerName": "ACPLT"}, {"InstanceId": "978-8234-234-342"}], + format_client.parse_collection(response), + ) + + @with_json_client + def test_submodel_elements_value_get_omits_operations_and_capabilities(self, format_client: FormatClient): + response = format_client.get(f"{self.elements_path(TEST_SUBMODEL_ID)}/$value") + + self.assert_ok(response) + id_shorts = [next(iter(entry)) for entry in format_client.parse_collection(response)] + self.assertNotIn("ExampleOperation", id_shorts) + self.assertNotIn("ExampleCapability", id_shorts) + + @with_json_client + def test_submodel_elements_value_get_pagination_limit(self, format_client: FormatClient): + response = format_client.get(f"{self.elements_path(IDENTIFICATION_SUBMODEL_ID)}/$value?limit=1") + + self.assert_ok(response) + self.assertEqual([{"ManufacturerName": "ACPLT"}], format_client.parse_collection(response)) + self.assertEqual("2", format_client.next_cursor(response)) + + # ------------------------------------------------------------------ GET ...//$value + + @with_json_client + def test_property_value_get_value_types(self, format_client: FormatClient): + def get_property_value(id_short: str): + return self.get_value(format_client, f"{self.elements_path(TYPED_SUBMODEL_ID, id_short)}/$value") + + self.assertEqual(42, get_property_value("IntProperty")) + self.assertEqual(False, get_property_value("BoolProperty")) + self.assertEqual(1.5, get_property_value("DoubleProperty")) + # JSON has no representation for NaN, INF and -INF, so the XSD representation is returned instead + self.assertEqual("NaN", get_property_value("NanProperty")) + # integers are not converted to float, as that would raise an OverflowError for large values + self.assertEqual(10**400, get_property_value("HugeIntProperty")) + self.assertEqual("2026-08-31", get_property_value("DateProperty")) + self.assertIsNone(get_property_value("EmptyProperty")) + + @with_json_client + def test_submodel_element_list_value_get_keeps_indices(self, format_client: FormatClient): + value = self.get_value( + format_client, f"{self.elements_path(TYPED_SUBMODEL_ID, 'ListWithEmptyElement')}/$value" + ) + + self.assertEqual(["first", None, "third"], value) + + @with_json_client + def test_multi_language_property_value_get(self, format_client: FormatClient): + value = self.get_value( + format_client, + f"{self.elements_path(TEST_SUBMODEL_ID, 'ExampleSubmodelCollection.ExampleMultiLanguageProperty')}" + "/$value", + ) + + self.assertEqual(2, len(value)) + self.assertEqual([1, 1], [len(entry) for entry in value]) + self.assertEqual(["de", "en-US"], [next(iter(entry)) for entry in value]) + + @with_json_client + def test_range_value_get(self, format_client: FormatClient): + value = self.get_value( + format_client, f"{self.elements_path(TEST_SUBMODEL_ID, 'ExampleSubmodelCollection.ExampleRange')}/$value" + ) + + self.assertEqual({"min": 0, "max": 100}, value) + + @with_json_client + def test_blob_value_get(self, format_client: FormatClient): + value = self.get_value( + format_client, f"{self.elements_path(TEST_SUBMODEL_ID, 'ExampleSubmodelCollection.ExampleBlob')}/$value" + ) + + self.assertEqual( + {"contentType": "application/pdf", "value": base64.b64encode(b"\x01\x02\x03\x04\x05").decode()}, value + ) + + @with_json_client + def test_file_value_get(self, format_client: FormatClient): + value = self.get_value( + format_client, f"{self.elements_path(TEST_SUBMODEL_ID, 'ExampleSubmodelCollection.ExampleFile')}/$value" + ) + + self.assertEqual({"contentType": "application/pdf", "value": "/TestFile.pdf"}, value) + + @with_json_client + def test_reference_element_value_get(self, format_client: FormatClient): + value = self.get_value( + format_client, + f"{self.elements_path(TEST_SUBMODEL_ID, 'ExampleSubmodelCollection.ExampleReferenceElement')}/$value", + ) + + self.assertEqual("ModelReference", value["type"]) + self.assertEqual("Submodel", value["keys"][0]["type"]) + + @with_json_client + def test_relationship_element_value_get(self, format_client: FormatClient): + value = self.get_value( + format_client, f"{self.elements_path(TEST_SUBMODEL_ID, 'ExampleRelationshipElement')}/$value" + ) + + self.assertEqual({"first", "second"}, set(value)) + self.assertEqual("ModelReference", value["first"]["type"]) + + @with_json_client + def test_annotated_relationship_element_value_get(self, format_client: FormatClient): + value = self.get_value( + format_client, f"{self.elements_path(TEST_SUBMODEL_ID, 'ExampleAnnotatedRelationshipElement')}/$value" + ) + + self.assertEqual({"first", "second", "annotations"}, set(value)) + self.assertIn({"ExampleAnnotatedProperty": "exampleValue"}, value["annotations"]) + + @with_json_client + def test_basic_event_element_value_get(self, format_client: FormatClient): + value = self.get_value( + format_client, f"{self.elements_path(TEST_SUBMODEL_ID, 'ExampleBasicEventElement')}/$value" + ) + + self.assertEqual(["observed"], list(value)) + self.assertEqual("ModelReference", value["observed"]["type"]) + + @with_json_client + def test_entity_value_get(self, format_client: FormatClient): + value = self.get_value( + format_client, f"{self.elements_path(BILL_OF_MATERIAL_SUBMODEL_ID, 'ExampleEntity')}/$value" + ) + + self.assertEqual("SelfManagedEntity", value["entityType"]) + self.assertEqual("http://example.org/TestAsset/", value["globalAssetId"]) + self.assertEqual("exampleValue", value["statements"]["ExampleProperty"]) + self.assertEqual("TestKey", value["specificAssetIds"][0]["name"]) + + @with_json_client + def test_submodel_element_collection_value_get(self, format_client: FormatClient): + value = self.get_value( + format_client, f"{self.elements_path(TEST_SUBMODEL_ID, 'ExampleSubmodelCollection')}/$value" + ) + + self.assertIn("ExampleBlob", value) + self.assertIn("ExampleSubmodelList", value) + + @with_json_client + def test_operation_and_capability_value_get_returns_400(self, format_client: FormatClient): + for id_short in ("ExampleOperation", "ExampleCapability"): + with self.subTest(id_short=id_short): + response = format_client.get(f"{self.elements_path(TEST_SUBMODEL_ID, id_short)}/$value") + + self.assert_error(response, 400) + + @with_json_client + def test_submodel_element_value_get_not_found(self, format_client: FormatClient): + response = format_client.get(f"{self.elements_path(TEST_SUBMODEL_ID, 'DoesNotExist')}/$value") + + self.assert_error(response, 404) + + # ------------------------------------------------------------------ content negotiation + + @with_xml_client + def test_value_get_xml_not_acceptable(self, format_client: FormatClient): + for path in ( + "/submodels/$value", + f"{self.submodel_path(TEST_SUBMODEL_ID)}/$value", + f"{self.elements_path(TEST_SUBMODEL_ID)}/$value", + f"{self.elements_path(TEST_SUBMODEL_ID, 'ExampleRelationshipElement')}/$value", + ): + with self.subTest(path=path): + response = format_client.get(path) + + self.assert_error(response, 406)