From 4413c62c8800c7253a74d190bf58e5126b6f699e 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 ++++++- server/test/interfaces/test_value_only.py | 264 ++++++++++++++++++++++ 5 files changed, 570 insertions(+), 7 deletions(-) create mode 100644 server/app/adapter/value_only.py create mode 100644 server/test/interfaces/test_value_only.py diff --git a/server/README.md b/server/README.md index 1a3e4217..d3d52c34 100644 --- a/server/README.md +++ b/server/README.md @@ -121,8 +121,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 10c99aa6..8a3e2af1 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 00000000..8eeb61a7 --- /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 8f931c78..e5f32779 100644 --- a/server/app/interfaces/repository.py +++ b/server/app/interfaces/repository.py @@ -22,11 +22,17 @@ from werkzeug.exceptions import BadRequest, Conflict, NotFound from werkzeug.routing import MapAdapter, Rule, Submount +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( [ @@ -122,7 +128,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), @@ -133,7 +139,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), @@ -160,7 +166,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( "/", @@ -205,7 +215,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 ), @@ -519,6 +533,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!") @@ -687,6 +724,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: @@ -714,6 +760,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: @@ -743,6 +796,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: @@ -768,6 +834,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/test_value_only.py b/server/test/interfaces/test_value_only.py new file mode 100644 index 00000000..391df86b --- /dev/null +++ b/server/test/interfaces/test_value_only.py @@ -0,0 +1,264 @@ +# 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 json +import math +import unittest +from typing import Any + +from app.interfaces.repository import WSGIApp +from basyx.aas import model +from basyx.aas.adapter.aasx import DictSupplementaryFileContainer +from basyx.aas.examples.data.example_aas import create_full_example +from basyx.aas.model import DictIdentifiableStore +from werkzeug.test import Client + +BASE_PATH = "/api/v3.1" +TEST_SUBMODEL_ID = "https://example.org/Test_Submodel" +IDENTIFICATION_SUBMODEL_ID = "http://example.org/Submodels/Assets/TestAsset/Identification" +TYPED_SUBMODEL_ID = "http://example.org/Typed_Submodel" + + +def _encode(identifier: str) -> str: + return base64.urlsafe_b64encode(identifier.encode()).decode() + + +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"), + ), + ), + ), + ) + + +class ValueOnlyTest(unittest.TestCase): + def setUp(self) -> None: + self.example_data = create_full_example() + self.example_data.add(_create_typed_submodel()) + self.client = Client(WSGIApp(self.example_data, DictSupplementaryFileContainer())) + + def _get_json(self, url: str, status_code: int = 200, **kwargs: Any) -> Any: + response = self.client.get(f"{BASE_PATH}{url}", **kwargs) + self.assertEqual(status_code, response.status_code, response.data) + return json.loads(response.data) + + # --------- GET /submodels/{submodelIdentifier}/$value --------- + + def test_submodel_value(self) -> None: + value = self._get_json(f"/submodels/{_encode(IDENTIFICATION_SUBMODEL_ID)}/$value") + self.assertEqual({"ManufacturerName": "ACPLT", "InstanceId": "978-8234-234-342"}, value) + + def test_submodel_value_omits_operations_and_capabilities(self) -> None: + value = self._get_json(f"/submodels/{_encode(TEST_SUBMODEL_ID)}/$value") + self.assertNotIn("ExampleOperation", value) + self.assertNotIn("ExampleCapability", value) + + def test_submodel_value_level_core_empties_nested_containers(self) -> None: + value = self._get_json(f"/submodels/{_encode(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"]) + + def test_submodel_value_unknown_submodel(self) -> None: + self._get_json(f"/submodels/{_encode('http://example.org/does_not_exist')}/$value", status_code=404) + + # --------- GET /submodels/{submodelIdentifier}/submodel-elements/{idShortPath}/$value --------- + + def test_property_value_types(self) -> None: + base_url = f"/submodels/{_encode(TYPED_SUBMODEL_ID)}/submodel-elements" + self.assertEqual(42, self._get_json(f"{base_url}/IntProperty/$value")) + self.assertEqual(False, self._get_json(f"{base_url}/BoolProperty/$value")) + self.assertEqual(1.5, self._get_json(f"{base_url}/DoubleProperty/$value")) + # JSON has no representation for NaN, INF and -INF, so the XSD representation is returned instead + self.assertEqual("NaN", self._get_json(f"{base_url}/NanProperty/$value")) + # integers are not converted to float, as that would raise an OverflowError for large values + self.assertEqual(10**400, self._get_json(f"{base_url}/HugeIntProperty/$value")) + self.assertEqual("2026-08-31", self._get_json(f"{base_url}/DateProperty/$value")) + self.assertIsNone(self._get_json(f"{base_url}/EmptyProperty/$value")) + + def test_submodel_value_with_huge_integer(self) -> None: + # integers that exceed the range of a float are also serialized when the whole Submodel is requested + value = self._get_json(f"/submodels/{_encode(TYPED_SUBMODEL_ID)}/$value") + self.assertEqual(10**400, value["HugeIntProperty"]) + self.assertEqual(42, value["IntProperty"]) + + def test_property_without_value_is_omitted_in_containers(self) -> None: + value = self._get_json(f"/submodels/{_encode(TYPED_SUBMODEL_ID)}/$value") + self.assertNotIn("EmptyProperty", value) + + def test_submodel_element_list_keeps_indices(self) -> None: + value = self._get_json(f"/submodels/{_encode(TYPED_SUBMODEL_ID)}/submodel-elements/ListWithEmptyElement/$value") + self.assertEqual(["first", None, "third"], value) + + def test_multi_language_property_value(self) -> None: + value = self._get_json( + f"/submodels/{_encode(TEST_SUBMODEL_ID)}" + "/submodel-elements/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]) + + def test_range_value(self) -> None: + value = self._get_json( + f"/submodels/{_encode(TEST_SUBMODEL_ID)}/submodel-elements/ExampleSubmodelCollection.ExampleRange/$value" + ) + self.assertEqual({"min": 0, "max": 100}, value) + + def test_blob_value(self) -> None: + value = self._get_json( + f"/submodels/{_encode(TEST_SUBMODEL_ID)}/submodel-elements/ExampleSubmodelCollection.ExampleBlob/$value" + ) + self.assertEqual( + {"contentType": "application/pdf", "value": base64.b64encode(b"\x01\x02\x03\x04\x05").decode()}, value + ) + + def test_file_value(self) -> None: + value = self._get_json( + f"/submodels/{_encode(TEST_SUBMODEL_ID)}/submodel-elements/ExampleSubmodelCollection.ExampleFile/$value" + ) + self.assertEqual({"contentType": "application/pdf", "value": "/TestFile.pdf"}, value) + + def test_reference_element_value(self) -> None: + value = self._get_json( + f"/submodels/{_encode(TEST_SUBMODEL_ID)}" + "/submodel-elements/ExampleSubmodelCollection.ExampleReferenceElement/$value" + ) + self.assertEqual("ModelReference", value["type"]) + self.assertEqual("Submodel", value["keys"][0]["type"]) + + def test_relationship_element_value(self) -> None: + value = self._get_json( + f"/submodels/{_encode(TEST_SUBMODEL_ID)}/submodel-elements/ExampleRelationshipElement/$value" + ) + self.assertEqual({"first", "second"}, set(value)) + self.assertEqual("ModelReference", value["first"]["type"]) + + def test_annotated_relationship_element_value(self) -> None: + value = self._get_json( + f"/submodels/{_encode(TEST_SUBMODEL_ID)}/submodel-elements/ExampleAnnotatedRelationshipElement/$value" + ) + self.assertEqual({"first", "second", "annotations"}, set(value)) + self.assertIn({"ExampleAnnotatedProperty": "exampleValue"}, value["annotations"]) + + def test_basic_event_element_value(self) -> None: + value = self._get_json( + f"/submodels/{_encode(TEST_SUBMODEL_ID)}/submodel-elements/ExampleBasicEventElement/$value" + ) + self.assertEqual(["observed"], list(value)) + self.assertEqual("ModelReference", value["observed"]["type"]) + + def test_entity_value(self) -> None: + value = self._get_json( + f"/submodels/{_encode('http://example.org/Submodels/Assets/TestAsset/BillOfMaterial')}" + "/submodel-elements/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"]) + + def test_submodel_element_collection_value(self) -> None: + value = self._get_json( + f"/submodels/{_encode(TEST_SUBMODEL_ID)}/submodel-elements/ExampleSubmodelCollection/$value" + ) + self.assertIn("ExampleBlob", value) + self.assertIn("ExampleSubmodelList", value) + + def test_operation_and_capability_are_rejected(self) -> None: + base_url = f"/submodels/{_encode(TEST_SUBMODEL_ID)}/submodel-elements" + self._get_json(f"{base_url}/ExampleOperation/$value", status_code=400) + self._get_json(f"{base_url}/ExampleCapability/$value", status_code=400) + + def test_unknown_submodel_element(self) -> None: + self._get_json(f"/submodels/{_encode(TEST_SUBMODEL_ID)}/submodel-elements/DoesNotExist/$value", status_code=404) + + # --------- GET /submodels/{submodelIdentifier}/submodel-elements/$value --------- + + def test_submodel_elements_value(self) -> None: + result = self._get_json(f"/submodels/{_encode(IDENTIFICATION_SUBMODEL_ID)}/submodel-elements/$value") + self.assertEqual( + [{"ManufacturerName": "ACPLT"}, {"InstanceId": "978-8234-234-342"}], + result["result"], + ) + + def test_submodel_elements_value_omits_operations_and_capabilities(self) -> None: + result = self._get_json(f"/submodels/{_encode(TEST_SUBMODEL_ID)}/submodel-elements/$value") + id_shorts = [next(iter(entry)) for entry in result["result"]] + self.assertNotIn("ExampleOperation", id_shorts) + self.assertNotIn("ExampleCapability", id_shorts) + + def test_submodel_elements_value_pagination(self) -> None: + result = self._get_json(f"/submodels/{_encode(IDENTIFICATION_SUBMODEL_ID)}/submodel-elements/$value?limit=1") + self.assertEqual([{"ManufacturerName": "ACPLT"}], result["result"]) + self.assertEqual("2", result["paging_metadata"]["cursor"]) + + # --------- GET /submodels/$value --------- + + def test_submodel_all_value(self) -> None: + result = self._get_json("/submodels/$value") + self.assertIn({"ManufacturerName": "ACPLT", "InstanceId": "978-8234-234-342"}, result["result"]) + + def test_submodel_all_value_pagination(self) -> None: + result = self._get_json("/submodels/$value?limit=1") + self.assertEqual(1, len(result["result"])) + self.assertEqual("2", result["paging_metadata"]["cursor"]) + + # --------- serialization modifiers and content negotiation --------- + + def test_xml_is_not_acceptable(self) -> None: + response = self.client.get( + f"{BASE_PATH}/submodels/{_encode(TEST_SUBMODEL_ID)}/$value", headers={"Accept": "application/xml"} + ) + self.assertEqual(406, response.status_code) + + def test_extent_is_not_implemented(self) -> None: + response = self.client.get(f"{BASE_PATH}/submodels/{_encode(TEST_SUBMODEL_ID)}/$value?extent=withBlobValue") + self.assertEqual(501, response.status_code) + + def test_invalid_level(self) -> None: + response = self.client.get(f"{BASE_PATH}/submodels/{_encode(TEST_SUBMODEL_ID)}/$value?level=invalid") + self.assertEqual(400, response.status_code) + + +class ValueOnlyEmptyStoreTest(unittest.TestCase): + def test_empty_submodel(self) -> None: + object_store: DictIdentifiableStore = DictIdentifiableStore() + object_store.add(model.Submodel(id_=TYPED_SUBMODEL_ID)) + client = Client(WSGIApp(object_store, DictSupplementaryFileContainer())) + response = client.get(f"{BASE_PATH}/submodels/{_encode(TYPED_SUBMODEL_ID)}/$value") + self.assertEqual(200, response.status_code) + self.assertEqual({}, json.loads(response.data)) + + +if __name__ == "__main__": + unittest.main()