diff --git a/.github/workflows/pr.yml b/.github/workflows/pr.yml index ce3008e4d..31da5a9aa 100644 --- a/.github/workflows/pr.yml +++ b/.github/workflows/pr.yml @@ -343,6 +343,41 @@ jobs: run: | python -m build + server-test: + # This job runs the unittests on the python versions specified down at the matrix + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.10", "3.12"] + defaults: + run: + working-directory: ./server + + steps: + - uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 #v5.1.0 + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v7 + with: + python-version: ${{ matrix.python-version }} + cache: "pip" + cache-dependency-path: "**/pyproject.toml" + - name: Install Python dependencies + run: | + python -m pip install --upgrade pip + python -m pip install ../sdk + python -m pip install .[dev] + - name: Test with coverage + unittest + env: + # No server/container is started in this job, so the Docker integration tests should just skip here, + # not be required to pass (see test_helpers.py). + REQUIRE_SERVER_INTEGRATION_TESTS: "0" + run: | + python -m coverage run --source=app -m unittest + - name: Report test coverage + if: ${{ always() }} + run: | + python -m coverage report -m + server-static-analysis: # This job runs static code analysis, namely ruff and mypy runs-on: ubuntu-latest @@ -390,8 +425,10 @@ jobs: publish: false platform: linux/amd64 - name: Run container + # Enable STORAGE_PERSISTENCY as registry fails with in-memory store (#626) + # TODO(#626): revisit when `DictDescriptorStore` throws now exception on `commit()` run: | - docker run -d --name basyx-python-${{ matrix.profile }} -p 9080:80 --pull=never ${{ steps.build.outputs.image-ref }} + docker run -d --name basyx-python-${{ matrix.profile }} -p 8080:80 -eSTORAGE_PERSISTENCY=True --pull=never ${{ steps.build.outputs.image-ref }} - name: Wait for container and server initialization run: | timeout 30s bash -c ' @@ -401,7 +438,27 @@ jobs: ' - name: Check if service is alive run: | - curl -f http://localhost:9080/api/${{ env.X_API_VERSION }}/description + curl -f http://localhost:8080/api/${{ env.X_API_VERSION }}/description + - name: Set up Python ${{ env.X_PYTHON_MIN_VERSION }} + uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v7 + with: + python-version: ${{ env.X_PYTHON_MIN_VERSION }} + cache: "pip" + cache-dependency-path: "**/pyproject.toml" + - name: Install Python dependencies + working-directory: ./server + run: | + python -m pip install --upgrade pip + python -m pip install ../sdk + python -m pip install . + - name: Run Docker integration tests + # Each profile has its own test module + working-directory: ./server + env: + # Fail instead of silently skipping if the container isn't actually reachable (see test_helpers.py). + REQUIRE_SERVER_INTEGRATION_TESTS: "1" + run: | + python -m unittest test.docker_integration.test_docker_integration_${{ matrix.profile }} -v - name: Stop and remove the container run: | docker stop basyx-python-${{ matrix.profile }} && docker rm basyx-python-${{ matrix.profile }} diff --git a/server/pyproject.toml b/server/pyproject.toml index e87990397..38ab0a9f9 100644 --- a/server/pyproject.toml +++ b/server/pyproject.toml @@ -42,6 +42,7 @@ dependencies = [ [project.optional-dependencies] dev = [ + "coverage", "mypy", "pycodestyle", "ruff==0.16.0", diff --git a/server/test/_helper/__init__.py b/server/test/_helper/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/server/test/_helper/test_helpers.py b/server/test/_helper/test_helpers.py new file mode 100644 index 000000000..89b2444e9 --- /dev/null +++ b/server/test/_helper/test_helpers.py @@ -0,0 +1,29 @@ +import configparser +import os +import os.path +import urllib.error +import urllib.request + +TEST_CONFIG = configparser.ConfigParser() +TEST_CONFIG.read( + ( + os.path.join(os.path.dirname(__file__), "..", "test_config.default.ini"), + os.path.join(os.path.dirname(__file__), "..", "test_config.ini"), + ) +) + + +# By default, the Docker integration tests are skipped whenever no server is reachable, so that a plain local +# `python -m unittest` run doesn't require a running Docker container. Set this environment variable to "1"/"true" +# (e.g. in CI) to instead make those tests fail loudly if no server is reachable, so a broken Docker container +# can't silently cause the tests to be skipped without anyone noticing. +REQUIRE_SERVER = os.environ.get("REQUIRE_SERVER_INTEGRATION_TESTS", "false").lower() in {"1", "true", "yes"} + +# Check if the server is available. Otherwise, skip tests (unless REQUIRE_SERVER is set, see above). +try: + urllib.request.urlopen(TEST_CONFIG["server"]["url"] + "/description", timeout=2) + SERVER_OKAY = True + SERVER_ERROR = None +except urllib.error.URLError as e: + SERVER_OKAY = False + SERVER_ERROR = e diff --git a/server/test/docker_integration/__init__.py b/server/test/docker_integration/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/server/test/docker_integration/test_docker_integration_discovery.py b/server/test/docker_integration/test_docker_integration_discovery.py new file mode 100644 index 000000000..49aeb212f --- /dev/null +++ b/server/test/docker_integration/test_docker_integration_discovery.py @@ -0,0 +1,77 @@ +import json +import unittest +import urllib.error +import urllib.request + +from app.interfaces.discovery import SUPPORTED_PROFILES +from app.util.converters import base64url_encode + +from test._helper.test_helpers import REQUIRE_SERVER, SERVER_ERROR, SERVER_OKAY, TEST_CONFIG + +SERVER_BASE_URL = TEST_CONFIG["server"]["url"] + + +@unittest.skipUnless( + SERVER_OKAY or REQUIRE_SERVER, f"No server reachable at {SERVER_BASE_URL}: {SERVER_ERROR}" +) +class DiscoveryDockerIntegrationTest(unittest.TestCase): + """ + Smoke tests against a real, already-running discovery server instance (e.g. started via + ``docker run -p 8080:80 basyx-python-discovery``), analogous to ``test_docker_integration_repository.py`` for the + repository profile: skipped entirely if no server is reachable at ``SERVER_BASE_URL``. + + Set the ``REQUIRE_SERVER_INTEGRATION_TESTS`` environment variable to make this test class fail instead of + being skipped when no server is reachable (see ``test._helper.test_helpers``). + """ + + AAS_ID = "https://example.org/Test_AssetAdministrationShell_Discovery" + ASSET_LINK = {"name": "MySerialNumber", "value": "SN-12345"} + + @classmethod + def setUpClass(cls) -> None: + if not SERVER_OKAY: + raise RuntimeError( + f"REQUIRE_SERVER_INTEGRATION_TESTS is set, but no server is reachable at " + f"{SERVER_BASE_URL}: {SERVER_ERROR}" + ) + + def tearDown(self) -> None: + delete_request = urllib.request.Request( + f"{SERVER_BASE_URL}/lookup/shells/{base64url_encode(self.AAS_ID)}", method="DELETE" + ) + urllib.request.urlopen(delete_request).close() + + # ------------------------------------------------------------------ GET /description + + def test_description_profiles(self): + with urllib.request.urlopen(SERVER_BASE_URL + "/description") as response: + self.assertEqual(200, response.status) + data = json.loads(response.read()) + + expected_profiles = {profile.value for profile in SUPPORTED_PROFILES.profiles} + self.assertEqual(expected_profiles, set(data["profiles"])) + + # ------------------------------------------------------------------ POST/GET/DELETE /lookup/shells/ + + def test_asset_link_roundtrip(self): + aas_asset_links_path = f"{SERVER_BASE_URL}/lookup/shells/{base64url_encode(self.AAS_ID)}" + body = json.dumps([self.ASSET_LINK]).encode("utf-8") + + post_request = urllib.request.Request( + aas_asset_links_path, data=body, headers={"Content-Type": "application/json"}, method="POST" + ) + with urllib.request.urlopen(post_request) as response: + self.assertEqual(200, response.status) + + with urllib.request.urlopen(aas_asset_links_path) as response: + self.assertEqual(200, response.status) + retrieved = json.loads(response.read()) + self.assertIn(self.ASSET_LINK, retrieved) + + delete_request = urllib.request.Request(aas_asset_links_path, method="DELETE") + with urllib.request.urlopen(delete_request) as response: + self.assertEqual(204, response.status) + + with urllib.request.urlopen(aas_asset_links_path) as response: + self.assertEqual(200, response.status) + self.assertEqual([], json.loads(response.read())) diff --git a/server/test/docker_integration/test_docker_integration_registry.py b/server/test/docker_integration/test_docker_integration_registry.py new file mode 100644 index 000000000..a36b7e61a --- /dev/null +++ b/server/test/docker_integration/test_docker_integration_registry.py @@ -0,0 +1,92 @@ +import json +import unittest +import urllib.error +import urllib.request + +from app.adapter.jsonization import ServerAASToJsonEncoder +from app.interfaces.registry import SUPPORTED_PROFILES +from app.model import AssetAdministrationShellDescriptor +from app.util.converters import base64url_encode + +from test._helper.test_helpers import REQUIRE_SERVER, SERVER_ERROR, SERVER_OKAY, TEST_CONFIG + +SERVER_BASE_URL = TEST_CONFIG["server"]["url"] + + +@unittest.skipUnless( + SERVER_OKAY or REQUIRE_SERVER, f"No server reachable at {SERVER_BASE_URL}: {SERVER_ERROR}" +) +class RegistryDockerIntegrationTest(unittest.TestCase): + """ + Smoke tests against a real, already-running registry server instance (e.g. started via + ``docker run -p 8080:80 basyx-python-registry``), analogous to ``test_docker_integration_repository.py`` for the + repository profile: skipped entirely if no server is reachable at ``SERVER_BASE_URL``. + + Set the ``REQUIRE_SERVER_INTEGRATION_TESTS`` environment variable to make this test class fail instead of + being skipped when no server is reachable (see ``test._helper.test_helpers``). + """ + + DESCRIPTOR_ID = "https://example.org/Test_AssetAdministrationShellDescriptor" + + @classmethod + def setUpClass(cls) -> None: + if not SERVER_OKAY: + raise RuntimeError( + f"REQUIRE_SERVER_INTEGRATION_TESTS is set, but no server is reachable at " + f"{SERVER_BASE_URL}: {SERVER_ERROR}" + ) + + def tearDown(self) -> None: + self._delete_descriptor(self.DESCRIPTOR_ID, ignore_missing=True) + + @staticmethod + def _delete_descriptor(descriptor_id: str, ignore_missing: bool = False) -> None: + request = urllib.request.Request( + f"{SERVER_BASE_URL}/shell-descriptors/{base64url_encode(descriptor_id)}", method="DELETE" + ) + try: + urllib.request.urlopen(request) + except urllib.error.HTTPError as e: + e.close() + if not (ignore_missing and e.code == 404): + raise + + # ------------------------------------------------------------------ GET /description + + def test_description_profiles(self): + with urllib.request.urlopen(SERVER_BASE_URL + "/description") as response: + self.assertEqual(200, response.status) + data = json.loads(response.read()) + + expected_profiles = {profile.value for profile in SUPPORTED_PROFILES.profiles} + self.assertEqual(expected_profiles, set(data["profiles"])) + + # ------------------------------------------------------------------ POST/GET/DELETE /shell-descriptors + + def test_shell_descriptor_roundtrip(self): + descriptor = AssetAdministrationShellDescriptor(id_=self.DESCRIPTOR_ID, id_short="TestDescriptor") + body = json.dumps(descriptor, cls=ServerAASToJsonEncoder).encode("utf-8") + descriptor_path = f"{SERVER_BASE_URL}/shell-descriptors/{base64url_encode(descriptor.id)}" + + post_request = urllib.request.Request( + SERVER_BASE_URL + "/shell-descriptors", + data=body, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(post_request) as response: + self.assertEqual(201, response.status) + + with urllib.request.urlopen(descriptor_path) as response: + self.assertEqual(200, response.status) + retrieved = json.loads(response.read()) + self.assertEqual(descriptor.id, retrieved["id"]) + self.assertEqual("TestDescriptor", retrieved["idShort"]) + + delete_request = urllib.request.Request(descriptor_path, method="DELETE") + with urllib.request.urlopen(delete_request) as response: + self.assertEqual(204, response.status) + + with self.assertRaises(urllib.error.HTTPError) as cm: + urllib.request.urlopen(descriptor_path) + self.assertEqual(404, cm.exception.code) diff --git a/server/test/docker_integration/test_docker_integration_repository.py b/server/test/docker_integration/test_docker_integration_repository.py new file mode 100644 index 000000000..b571bcadb --- /dev/null +++ b/server/test/docker_integration/test_docker_integration_repository.py @@ -0,0 +1,83 @@ +import json +import unittest +import urllib.error +import urllib.request + +from app.interfaces.repository import SUPPORTED_PROFILES +from app.util.converters import base64url_encode +from basyx.aas.adapter.json import AASFromJsonDecoder, AASToJsonEncoder +from basyx.aas.examples.data.example_aas import ( + AASDataChecker, + check_example_asset_administration_shell, + create_example_asset_administration_shell, +) + +from test._helper.test_helpers import REQUIRE_SERVER, SERVER_ERROR, SERVER_OKAY, TEST_CONFIG + +SERVER_BASE_URL = TEST_CONFIG["server"]["url"] + + +@unittest.skipUnless( + SERVER_OKAY or REQUIRE_SERVER, f"No server reachable at {SERVER_BASE_URL}: {SERVER_ERROR}" +) +class ServerDockerIntegrationTest(unittest.TestCase): + """ + Smoke tests against a real, already-running server instance (e.g. started via + ``docker run -p 8080:80 basyx-python-server``), analogous to how ``test_couchdb.py`` tests + against a real CouchDB instance: skipped entirely if no server is reachable at ``SERVER_BASE_URL``. + + Set the ``REQUIRE_SERVER_INTEGRATION_TESTS`` environment variable to make this test class fail instead of + being skipped when no server is reachable (see ``test._helper.test_helpers``). CI uses this to ensure a + broken Docker container is reported as a failure rather than silently skipping the tests. + """ + + @classmethod + def setUpClass(cls) -> None: + if not SERVER_OKAY: + raise RuntimeError( + f"REQUIRE_SERVER_INTEGRATION_TESTS is set, but no server is reachable at " + f"{SERVER_BASE_URL}: {SERVER_ERROR}" + ) + + def tearDown(self) -> None: + self._delete_shell(create_example_asset_administration_shell().id, ignore_missing=True) + + @staticmethod + def _delete_shell(shell_id: str, ignore_missing: bool = False) -> None: + request = urllib.request.Request(f"{SERVER_BASE_URL}/shells/{base64url_encode(shell_id)}", method="DELETE") + try: + urllib.request.urlopen(request) + except urllib.error.HTTPError as e: + e.close() + if not (ignore_missing and e.code == 404): + raise + + # ------------------------------------------------------------------ GET /description + + def test_description_profiles(self): + with urllib.request.urlopen(SERVER_BASE_URL + "/description") as response: + self.assertEqual(200, response.status) + data = json.loads(response.read()) + + expected_profiles = {profile.value for profile in SUPPORTED_PROFILES.profiles} + self.assertEqual(expected_profiles, set(data["profiles"])) + + # ------------------------------------------------------------------ POST/GET/DELETE /shells + + def test_shell_roundtrip(self): + shell = create_example_asset_administration_shell() + body = json.dumps(shell, cls=AASToJsonEncoder).encode("utf-8") + shell_path = f"{SERVER_BASE_URL}/shells/{base64url_encode(shell.id)}" + + post_request = urllib.request.Request( + SERVER_BASE_URL + "/shells", data=body, headers={"Content-Type": "application/json"}, method="POST" + ) + with urllib.request.urlopen(post_request) as response: + self.assertEqual(201, response.status) + + with urllib.request.urlopen(shell_path) as response: + self.assertEqual(200, response.status) + retrieved = json.loads(response.read(), cls=AASFromJsonDecoder) + + checker = AASDataChecker(raise_immediately=True) + check_example_asset_administration_shell(checker, retrieved) diff --git a/server/test/interfaces/format_utils.py b/server/test/interfaces/format_utils.py new file mode 100644 index 000000000..2f169613c --- /dev/null +++ b/server/test/interfaces/format_utils.py @@ -0,0 +1,259 @@ +import abc +import json +from typing import Any, Callable, Optional + +import app.adapter +from basyx.aas import adapter +from basyx.aas.adapter._generic import XML_NS_MAP +from lxml import etree +from werkzeug.test import Client, TestResponse + + +class FormatClient(abc.ABC): + """ + Wraps a :class:`werkzeug.test.Client` and hides the request/response *format* (JSON or XML) behind a small, + format-agnostic API, so an endpoint test can be written once and run against every format. + + * request helpers (:meth:`get` / :meth:`post` / ...) inject the ``Accept`` / ``Content-Type`` headers and + serialize model objects passed as ``obj`` with :meth:`serialize`, + * parsing helpers (:meth:`parse_object` / :meth:`parse_collection` / :meth:`identifier` / ...) turn a response + body into plain Python values that are identical for both formats. + """ + + content_type: str + + def __init__(self, client: Client): + self.client = client + + def request(self, method: str, path: str, obj: Optional[object] = None, data: Any = None, **kwargs) -> TestResponse: + """ + Issue a request, setting the ``Accept`` header to the class' :attr:`content_type`. + + :param method: HTTP method to perform the request with + :param path: path to perform the request to + :param obj: If given, the object is parsed to :attr:`content_type` using :meth:`serialize` and sent as body. + :param data: If given, this is directly sent as body. Caution: gets overridden by :param:`obj` + :param kwargs: Additional arguments passed directly to :meth:`werkzeug.test.Client.open`. Can be used to set + different ``Content-Type`` for :param:`data`. + :return: The :class:`~werkzeug.test.TestResponse` object + """ + headers = dict(kwargs.get("headers", {})) + headers["Accept"] = self.content_type + + if obj is not None: + data = self.serialize(obj) + kwargs["content_type"] = self.content_type + + kwargs.update({"data": data, "headers": headers}) + + return self.client.open( + path, method=method, **kwargs + ) + + def get(self, path: str, **kwargs) -> TestResponse: + return self.request("GET", path, **kwargs) + + def get_paginated(self, path: str, limit: int, max_pages:int, **kwargs) -> list[list[Any]]: + """ + Iteratively query the paginated endpoint :param:`path` with the given :param:`limit` until + the server indicates the complete collection was read. + + :param path: path to perform the request to. + :param limit: the limit for the paginated request, controls maximum size of each page. + :param max_pages: request fails if more than these pages are returned by endpoint. + :param kwargs: additional arguments to pass to the query. + :return: list of returned pages, each page is a list as returned by :meth:`parse_collection`. + """ + pages: list[list[str]] = [] + cursor: str | None = None + while True: + start_or_and = "?" if "?" not in path else "&" + query = f"{path}{start_or_and}limit={limit}" + if cursor is not None: + query += f"&cursor={cursor}" + response = self.get(query, **kwargs) + assert 200 == response.status_code + page_content = self.parse_collection(response) + assert limit >= len(page_content), "paginated result contains more than limit items" + pages.append(page_content) + cursor = self.next_cursor(response) + if cursor is None: + break + assert len(pages) <= max_pages, "cursor never signalled the last page" + + return pages + + def post(self, path: str, obj: Optional[object] = None, **kwargs: Any) -> TestResponse: + return self.request("POST", path, obj=obj, **kwargs) + + def put(self, path: str, obj: Optional[object] = None, **kwargs: Any) -> TestResponse: + return self.request("PUT", path, obj=obj, **kwargs) + + def patch(self, path: str, obj: Optional[object] = None, **kwargs: Any) -> TestResponse: + return self.request("PATCH", path, obj=obj, **kwargs) + + def delete(self, path: str, **kwargs: Any) -> TestResponse: + return self.request("DELETE", path, **kwargs) + + # ------------------------------------------------------------------ format-specific hooks + + @abc.abstractmethod + def serialize(self, obj: object) -> bytes: + """Serialize a model object to a request body in the format under test.""" + + @abc.abstractmethod + def parse_object(self, response: TestResponse) -> Any: + """Return the single-object node of an object response (accepted by :meth:`identifier`, :meth:`field`, ...).""" + + @abc.abstractmethod + def parse_collection(self, response: TestResponse) -> list[Any]: + """Return the list of item nodes of a collection response, in document order.""" + + @abc.abstractmethod + def identifier(self, node: Any) -> str: + """The ``id`` of an Identifiable from a :meth:`parse_object` / :meth:`parse_collection` node.""" + + @abc.abstractmethod + def reference_target(self, node: Any) -> str: + """The value of the last key of a Reference node (a single-reference response or a collection item).""" + + @abc.abstractmethod + def field(self, node: Any, name: str) -> Optional[str]: + """The text of a direct scalar child ``name`` of ``node`` (JSON member / ``aas:``-prefixed XML element).""" + + @abc.abstractmethod + def result_success(self, response: TestResponse) -> bool: + """The value of the ``success`` flag in a ``Result`` body.""" + + @abc.abstractmethod + def next_cursor(self, response: TestResponse) -> Optional[str]: + """The paging cursor pointing at the next page, or ``None`` once the last page has been returned.""" + + +class JsonFormatClient(FormatClient): + content_type = "application/json" + + def serialize(self, obj: object) -> bytes: + return json.dumps(obj, cls=app.adapter.jsonization.ServerAASToJsonEncoder).encode("utf-8") + + def _payload(self, response: TestResponse) -> Any: + return json.loads(response.get_data(as_text=True)) + + def parse_object(self, response: TestResponse) -> Any: + return self._payload(response) + + def parse_collection(self, response: TestResponse) -> list[Any]: + payload = self._payload(response) + if isinstance(payload, dict) and "result" in payload: + return list(payload["result"]) + return list(payload) + + def identifier(self, node: Any) -> str: + return node["id"] + + def reference_target(self, node: Any) -> str: + return node["keys"][-1]["value"] + + def field(self, node: Any, name: str) -> Optional[str]: + return node.get(name) + + def result_success(self, response: TestResponse) -> bool: + body = self._payload(response) + return "success" not in body or bool(body["success"]) + + def next_cursor(self, response: TestResponse) -> Optional[str]: + payload = self._payload(response) + if isinstance(payload, dict): + return payload.get("paging_metadata", {}).get("cursor") + return None + + +class XmlFormatClient(FormatClient): + content_type = "application/xml" + + def serialize(self, obj: object) -> bytes: + item_elem = adapter.xml.object_to_xml_element(obj) + etree.cleanup_namespaces(item_elem, top_nsmap=XML_NS_MAP) + return etree.tostring(item_elem, xml_declaration=True, encoding="utf-8") + + def _root(self, response: TestResponse) -> etree._Element: + return etree.fromstring(response.data) + + def parse_object(self, response: TestResponse) -> Any: + # An object response is with the object's children hoisted onto it, so the root itself is the + # object node and `identifier` / `field` find e.g. directly beneath it. + return self._root(response) + + def parse_collection(self, response: TestResponse) -> list[Any]: + # A collection response is with one child element per item. + return list(self._root(response)) + + def identifier(self, node: Any) -> str: + found = node.findtext("aas:id", namespaces=XML_NS_MAP) + assert found is not None + return found + + def reference_target(self, node: Any) -> str: + values = node.findall(".//aas:key/aas:value", namespaces=XML_NS_MAP) + assert values, "no keys in reference node" + return values[-1].text + + def field(self, node: Any, name: str) -> Optional[str]: + return node.findtext(f"aas:{name}", namespaces=XML_NS_MAP) + + def result_success(self, response: TestResponse) -> bool: + # true|false... -- not namespaced in Result bodies. + success_elem = self._root(response).find("success") + return success_elem is None or success_elem.text == "true" + + def next_cursor(self, response: TestResponse) -> Optional[str]: + # The cursor is an attribute on the root; it is unconditionally serialized, so a + # missing next page shows up as the literal string "None" rather than an absent attribute. + cursor = self._root(response).get("cursor") + return cursor if cursor not in (None, "None") else None + + +def with_json_client(func): + client_types = getattr(func, "_client_types", []) + client_types.append(("json", JsonFormatClient)) + func._client_types = client_types + return func + +def with_xml_client(func): + client_types = getattr(func, "_client_types", []) + client_types.append(("xml", XmlFormatClient)) + func._client_types = client_types + return func + +def with_custom_client(name: str, client_type: type[FormatClient]): + def wrapper(func): + client_types = getattr(func, "_client_types", []) + client_types.append((name, client_type)) + func._client_types = client_types + return func + return wrapper + +def inject_format_clients(cls): + """Decorator to use on :class:`unittest.TestCase` when decorating functions with + :meth:`with_json_client`, :meth:`with_xml_client` or :meth:`with_custom_client`. For each + :class:`FormatClient` that is defined via these decorators on a function, a new function is + added to the class. The new function gets the specfied :class:`FormatcClient` injected as second + parameter. The name of the new function gets the format name as suffix. + """ + + def build_test(method: Callable[[Any, FormatClient], Any], format_client_type: type[FormatClient]): + def wrapper(self): + formatted_client = format_client_type(getattr(cls, "client")) + return method(self, formatted_client) + return wrapper + + for name, method in list(vars(cls).items()): + method_client_type: Optional[list[str]] = getattr(method, "_client_types", None) + if method_client_type is None: + continue + + # Method was decorated -> remove original method and insert new methods + delattr(cls, name) + for (format_name, format_client_type) in method_client_type: + setattr(cls, f"{name}_{format_name}", build_test(method, format_client_type)) + return cls diff --git a/server/test/interfaces/repository/__init__.py b/server/test/interfaces/repository/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/server/test/interfaces/repository/helpers.py b/server/test/interfaces/repository/helpers.py new file mode 100644 index 000000000..0ebcae2a0 --- /dev/null +++ b/server/test/interfaces/repository/helpers.py @@ -0,0 +1,64 @@ +import unittest +from typing import TypeVar +from unittest import mock + +from app.interfaces import repository +from basyx.aas import model +from basyx.aas.adapter import aasx +from basyx.aas.examples.data.example_aas_missing_attributes import ( + create_example_asset_administration_shell, +) +from basyx.aas.model import Identifiable +from werkzeug.test import Client, TestResponse + +T = TypeVar('T') + +class RepositoryEndpointTestBase(unittest.TestCase): + __test__ = False + + object_store: model.SetIdentifiableStore[Identifiable] + file_store: mock.Mock + repository_server: repository.WSGIApp + client: Client + + @classmethod + def setUpClass(cls) -> None: + super().setUpClass() + + cls.object_store = model.SetIdentifiableStore() # DictIdentifiableStore breaks, when IDs change + cls.file_store = mock.Mock(spec=aasx.AbstractSupplementaryFileContainer) + cls.repository_server = repository.WSGIApp(cls.object_store, cls.file_store, base_path="") + cls.client = Client(cls.repository_server) + + def setUp(self) -> None: + self.object_store.clear() + self.file_store.reset_mock() + + @classmethod + def two_shells_store(cls): + store = model.DictIdentifiableStore() + store.add(create_example_asset_administration_shell()) + second_shell = create_example_asset_administration_shell() + second_shell.id = "https://example.org/Test_AssetAdministrationShell_Second" + store.add(second_shell) + return store + + # ------------------------------------------------------------------ shared assertion helpers + + def assert_ok(self, response: TestResponse) -> None: + self.assertEqual(200, response.status_code, msg=response.get_data(as_text=True)) + + def assert_error(self, response: TestResponse, status_code: int) -> None: + self.assertEqual(status_code, response.status_code, msg=response.get_data(as_text=True)) + self.assertIn("success", response.get_data(as_text=True), msg=response.get_data(as_text=True)) + + +class TestServiceDescription(RepositoryEndpointTestBase): + __test__ = True + + def test_description(self): + response = self.client.get("/description") + self.assertEqual(200, response.status_code) + body = response.get_data(as_text=True) + self.assertIn("AssetAdministrationShellRepositoryServiceSpecification/SSP-001", body) + self.assertIn("SubmodelRepositoryServiceSpecification/SSP-001", body) diff --git a/server/test/interfaces/repository/test_concept_description.py b/server/test/interfaces/repository/test_concept_description.py new file mode 100644 index 000000000..e04234c87 --- /dev/null +++ b/server/test/interfaces/repository/test_concept_description.py @@ -0,0 +1,204 @@ +from app.util.converters import base64url_encode +from basyx.aas import model +from basyx.aas.examples.data.example_aas_missing_attributes import ( + create_example_asset_administration_shell, + create_example_concept_description, +) + +from ..format_utils import ( + FormatClient, + inject_format_clients, + with_json_client, + with_xml_client, +) +from .helpers import RepositoryEndpointTestBase + + +@inject_format_clients +class ConceptDescriptionsEndpointsTest(RepositoryEndpointTestBase): + """ + Endpoint tests for the implemented ``/concept-descriptions`` routes of + :class:`~app.interfaces.repository.WSGIApp`. + + Bodies are written once against the format-agnostic ``format_client`` helper. For each test two + variants are generated where the :class:`~..format_utils.JsonFormatClient` and + :class:`~..format_utils.XmlFormatClient` are injected respectively. + """ + + __test__ = True + + EXAMPLE_ID = "https://example.org/Test_ConceptDescription_Missing" + SECOND_ID = "https://example.org/Test_ConceptDescription_Second" + + def two_concept_descriptions_store(self) -> model.DictIdentifiableStore: + store: model.DictIdentifiableStore = model.DictIdentifiableStore() + store.add(create_example_concept_description()) + second = create_example_concept_description() + second.id = self.SECOND_ID + second.id_short = "SecondConceptDescription" + store.add(second) + return store + + def _get_concept_description_ids(self, format_client: FormatClient, query: str) -> set: + response = format_client.get(f"/concept-descriptions?{query}") + self.assert_ok(response) + return {format_client.identifier(node) for node in format_client.parse_collection(response)} + + # ------------------------------------------------------------------ GET /concept-descriptions + + @with_json_client + @with_xml_client + def test_concept_descriptions_get(self, format_client: FormatClient): + self.object_store.update(self.two_concept_descriptions_store()) + + response = format_client.get("/concept-descriptions") + + self.assert_ok(response) + self.assertEqual(2, len(format_client.parse_collection(response))) + + @with_json_client + @with_xml_client + def test_concept_descriptions_get_empty(self, format_client: FormatClient): + response = format_client.get("/concept-descriptions") + + self.assert_ok(response) + self.assertEqual(0, len(format_client.parse_collection(response))) + + @with_json_client + @with_xml_client + def test_concept_descriptions_get_only_returns_concept_descriptions(self, format_client: FormatClient): + # The object store is shared between the shell/submodel/concept-description repositories, so the + # collection endpoint has to filter by type. + self.object_store.add(create_example_concept_description()) + self.object_store.add(create_example_asset_administration_shell()) + + ids = self._get_concept_description_ids(format_client, "") + + self.assertEqual({self.EXAMPLE_ID}, ids) + + @with_json_client + @with_xml_client + def test_concept_descriptions_get_supports_pagination(self, format_client: FormatClient): + self.object_store.update(self.two_concept_descriptions_store()) + + pages = format_client.get_paginated("/concept-descriptions", limit=1, max_pages=2) + + self.assertEqual([1, 1], [len(page) for page in pages]) + + # ------------------------------------------------------------------ POST /concept-descriptions + + @with_json_client + @with_xml_client + def test_concept_descriptions_post_success(self, format_client: FormatClient): + example_cd = create_example_concept_description() + + response = format_client.post("/concept-descriptions", obj=example_cd) + + self.assertEqual(201, response.status_code) + self.assertIn(base64url_encode(example_cd.id), response.headers["Location"]) + self.assertIsNotNone(self.object_store.get(example_cd.id, None)) + + @with_json_client + @with_xml_client + def test_concept_descriptions_post_bad(self, format_client: FormatClient): + example_cd = create_example_concept_description() + example_cd.id = None # type: ignore + + response = format_client.post("/concept-descriptions", obj=example_cd) + + self.assert_error(response, 400) + + @with_json_client + @with_xml_client + def test_concept_descriptions_post_conflict(self, format_client: FormatClient): + example_cd = create_example_concept_description() + self.object_store.add(example_cd) + + response = format_client.post("/concept-descriptions", obj=example_cd) + + self.assert_error(response, 409) + + # ------------------------------------------------------------------ GET /concept-descriptions/ + + @with_json_client + @with_xml_client + def test_concept_description_get_success(self, format_client: FormatClient): + self.object_store.update(self.two_concept_descriptions_store()) + + response = format_client.get(f"/concept-descriptions/{base64url_encode(self.EXAMPLE_ID)}") + + self.assert_ok(response) + self.assertEqual(self.EXAMPLE_ID, format_client.identifier(format_client.parse_object(response))) + + @with_json_client + @with_xml_client + def test_concept_description_get_not_found(self, format_client: FormatClient): + response = format_client.get( + f"/concept-descriptions/{base64url_encode('https://example.org/unknown')}" + ) + + self.assert_error(response, 404) + + @with_json_client + @with_xml_client + def test_concept_description_get_wrong_type_returns_404(self, format_client: FormatClient): + # An Identifiable with this id exists, but it is a shell, not a ConceptDescription. + shell = create_example_asset_administration_shell() + self.object_store.add(shell) + + response = format_client.get(f"/concept-descriptions/{base64url_encode(shell.id)}") + + self.assert_error(response, 404) + + # ------------------------------------------------------------------ PUT /concept-descriptions/ + + @with_json_client + @with_xml_client + def test_concept_description_put_success(self, format_client: FormatClient): + self.object_store.add(create_example_concept_description()) + updated_cd = create_example_concept_description() + updated_cd.id_short = "UpdatedIdShort" + + response = format_client.put( + f"/concept-descriptions/{base64url_encode(updated_cd.id)}", obj=updated_cd + ) + + self.assertEqual(204, response.status_code) + retrieved_cd = self.object_store.get(updated_cd.id, None) + self.assertIsInstance(retrieved_cd, model.ConceptDescription) + assert isinstance(retrieved_cd, model.ConceptDescription) # make mypy happy + self.assertEqual("UpdatedIdShort", retrieved_cd.id_short) + + @with_json_client + @with_xml_client + def test_concept_description_put_not_found(self, format_client: FormatClient): + updated_cd = create_example_concept_description() + updated_cd.id = "https://example.org/unknown" + + response = format_client.put( + f"/concept-descriptions/{base64url_encode(updated_cd.id)}", obj=updated_cd + ) + + self.assert_error(response, 404) + + # ------------------------------------------------------------------ DELETE /concept-descriptions/ + + @with_json_client + @with_xml_client + def test_concept_description_delete_success(self, format_client: FormatClient): + example_cd = create_example_concept_description() + self.object_store.add(example_cd) + + response = format_client.delete(f"/concept-descriptions/{base64url_encode(example_cd.id)}") + + self.assertEqual(204, response.status_code) + self.assertIsNone(self.object_store.get(example_cd.id, None)) + + @with_json_client + @with_xml_client + def test_concept_description_delete_not_found(self, format_client: FormatClient): + response = format_client.delete( + f"/concept-descriptions/{base64url_encode('https://example.org/unknown')}" + ) + + self.assert_error(response, 404) diff --git a/server/test/interfaces/repository/test_shells.py b/server/test/interfaces/repository/test_shells.py new file mode 100644 index 000000000..081047503 --- /dev/null +++ b/server/test/interfaces/repository/test_shells.py @@ -0,0 +1,630 @@ +import base64 +import json +from typing import Iterable + +from app.util.converters import base64url_encode +from basyx.aas import model +from basyx.aas.adapter.json import AASToJsonEncoder +from basyx.aas.examples.data.example_aas_missing_attributes import ( + create_example_asset_administration_shell, + create_example_submodel, +) + +from ..format_utils import ( + FormatClient, + inject_format_clients, + with_json_client, + with_xml_client, +) +from .helpers import RepositoryEndpointTestBase + + +def _encode_name_value_pair(name: str, value: str) -> str: + payload = json.dumps({"name": name, "value": value}) + return base64.urlsafe_b64encode(payload.encode()).decode() + + +def _encode_global_asset_id(value: str) -> str: + return _encode_name_value_pair("globalAssetId", value) + + +def _encode_specific_asset_id(specific_asset_id: model.SpecificAssetId) -> str: + return _encode_name_value_pair("specificAssetId", json.dumps(specific_asset_id, cls=AASToJsonEncoder)) + + +@inject_format_clients +class ShellsEndpointsTest(RepositoryEndpointTestBase): + """ + Endpoint tests for the implemented ``/shells`` routes of :class:`~app.interfaces.repository.WSGIApp`. + + Bodies are written once against the format-agnostic ``format_client`` helper. + For each test two variants are generated where the :class:`~..format_utils.JsonFormatClient` and + :class:`~..format_utils.XmlFormatClient` are injected respectively. + """ + + __test__ = True + + # ------------------------------------------------------------------ GET /shells + + @with_json_client + @with_xml_client + def test_shells_get(self, format_client: FormatClient): + self.object_store.update(self.two_shells_store()) + + response = format_client.get("/shells") + + self.assert_ok(response) + self.assertEqual(2, len(format_client.parse_collection(response))) + + @with_json_client + @with_xml_client + def test_shells_get_supports_pagination(self, format_client: FormatClient): + self.object_store.update(self.two_shells_store()) + + pages = format_client.get_paginated("/shells", limit=1, max_pages=2) + + self.assertEqual([1, 1], [len(page) for page in pages]) + + # ------------------------------------------------------------------ GET /shells?idShort=...&assetIds=... + + @staticmethod + def _specific_asset_id(name: str, value: str, subject: str) -> model.SpecificAssetId: + return model.SpecificAssetId( + name=name, + value=value, + external_subject_id=model.ExternalReference((model.Key(model.KeyTypes.GLOBAL_REFERENCE, subject),)), + ) + + @staticmethod + def _shell( + id_: str, + id_short: str, + global_asset_id: str, + specific_asset_ids: Iterable[model.SpecificAssetId] = (), + ) -> model.AssetAdministrationShell: + return model.AssetAdministrationShell( + asset_information=model.AssetInformation( + asset_kind=model.AssetKind.INSTANCE, + global_asset_id=global_asset_id, + specific_asset_id=specific_asset_ids, + ), + id_=id_, + id_short=id_short, + ) + + def shells_for_filtering_store(self): + store = model.DictIdentifiableStore() + store.add( + self._shell( + "https://example.org/shell-alpha", + "Alpha", + "https://example.org/asset-alpha", + [self._specific_asset_id("Serial", "111", "https://example.org/subject-alpha")], + ) + ) + store.add( + self._shell( + "https://example.org/shell-beta", + "Beta", + "https://example.org/asset-beta", + [ + self._specific_asset_id("Serial", "222", "https://example.org/subject-beta"), + self._specific_asset_id("Batch", "xyz", "https://example.org/subject-beta"), + ], + ) + ) + store.add(self._shell("https://example.org/shell-gamma", "Alpha", "https://example.org/asset-alpha")) + return store + + def _get_shell_ids(self, format_client: FormatClient, query: str) -> set: + response = format_client.get(f"/shells?{query}") + self.assert_ok(response) + return {format_client.identifier(node) for node in format_client.parse_collection(response)} + + @with_json_client + @with_xml_client + def test_shells_get_filter_by_id_short(self, format_client: FormatClient): + self.object_store.update(self.shells_for_filtering_store()) + + ids = self._get_shell_ids(format_client, "idShort=Alpha") + + self.assertEqual( + {"https://example.org/shell-alpha", "https://example.org/shell-gamma"}, ids + ) + + @with_json_client + @with_xml_client + def test_shells_get_filter_by_id_short_no_match(self, format_client: FormatClient): + self.object_store.update(self.shells_for_filtering_store()) + + ids = self._get_shell_ids(format_client, "idShort=Unknown") + + self.assertEqual(set(), ids) + + @with_json_client + @with_xml_client + def test_shells_get_filter_by_global_asset_id(self, format_client: FormatClient): + self.object_store.update(self.shells_for_filtering_store()) + + query = f"assetIds={_encode_global_asset_id('https://example.org/asset-alpha')}" + ids = self._get_shell_ids(format_client, query) + + self.assertEqual( + {"https://example.org/shell-alpha", "https://example.org/shell-gamma"}, ids + ) + + @with_json_client + @with_xml_client + def test_shells_get_filter_by_multiple_global_asset_ids_is_or(self, format_client: FormatClient): + self.object_store.update(self.shells_for_filtering_store()) + + query = "&".join( + [ + f"assetIds={_encode_global_asset_id('https://example.org/asset-beta')}", + f"assetIds={_encode_global_asset_id('https://example.org/nonexistent-asset')}", + ] + ) + ids = self._get_shell_ids(format_client, query) + + self.assertEqual({"https://example.org/shell-beta"}, ids) + + @with_json_client + @with_xml_client + def test_shells_get_filter_by_specific_asset_id(self, format_client: FormatClient): + self.object_store.update(self.shells_for_filtering_store()) + specific_id = self._specific_asset_id("Serial", "222", "https://example.org/subject-beta") + + query = f"assetIds={_encode_specific_asset_id(specific_id)}" + ids = self._get_shell_ids(format_client, query) + + self.assertEqual({"https://example.org/shell-beta"}, ids) + + @with_json_client + @with_xml_client + def test_shells_get_filter_by_multiple_specific_asset_ids_requires_all(self, format_client: FormatClient): + self.object_store.update(self.shells_for_filtering_store()) + serial = self._specific_asset_id("Serial", "222", "https://example.org/subject-beta") + batch = self._specific_asset_id("Batch", "xyz", "https://example.org/subject-beta") + + query = "&".join( + [f"assetIds={_encode_specific_asset_id(serial)}", f"assetIds={_encode_specific_asset_id(batch)}"] + ) + ids = self._get_shell_ids(format_client, query) + + self.assertEqual({"https://example.org/shell-beta"}, ids) + + @with_json_client + @with_xml_client + def test_shells_get_filter_by_specific_asset_ids_from_different_shells_matches_none( + self, format_client: FormatClient + ): + self.object_store.update(self.shells_for_filtering_store()) + alpha_specific_id = self._specific_asset_id("Serial", "111", "https://example.org/subject-alpha") + beta_specific_id = self._specific_asset_id("Serial", "222", "https://example.org/subject-beta") + + query = "&".join( + [ + f"assetIds={_encode_specific_asset_id(alpha_specific_id)}", + f"assetIds={_encode_specific_asset_id(beta_specific_id)}", + ] + ) + ids = self._get_shell_ids(format_client, query) + + self.assertEqual(set(), ids) + + @with_json_client + @with_xml_client + def test_shells_get_filter_by_specific_and_global_asset_id_is_and(self, format_client: FormatClient): + # shell-beta has this specificAssetId, but not this globalAssetId (that's shell-alpha's) -- combining + # both must AND the two conditions together, so neither shell matches. + self.object_store.update(self.shells_for_filtering_store()) + beta_specific_id = self._specific_asset_id("Serial", "222", "https://example.org/subject-beta") + + query = "&".join( + [ + f"assetIds={_encode_specific_asset_id(beta_specific_id)}", + f"assetIds={_encode_global_asset_id('https://example.org/asset-alpha')}", + ] + ) + ids = self._get_shell_ids(format_client, query) + + self.assertEqual(set(), ids) + + @with_json_client + @with_xml_client + def test_shells_get_filter_by_id_short_and_asset_ids_is_and(self, format_client: FormatClient): + # idShort=Alpha matches shell-alpha and shell-gamma, but only shell-alpha carries this specificAssetId. + self.object_store.update(self.shells_for_filtering_store()) + alpha_specific_id = self._specific_asset_id("Serial", "111", "https://example.org/subject-alpha") + + query = f"idShort=Alpha&assetIds={_encode_specific_asset_id(alpha_specific_id)}" + ids = self._get_shell_ids(format_client, query) + + self.assertEqual({"https://example.org/shell-alpha"}, ids) + + @with_json_client + @with_xml_client + def test_shells_get_filter_by_malformed_asset_id_returns_400(self, format_client: FormatClient): + self.object_store.update(self.shells_for_filtering_store()) + malformed = base64.urlsafe_b64encode(json.dumps({"name": "globalAssetId"}).encode()).decode() + + response = format_client.get(f"/shells?assetIds={malformed}") + + self.assert_error(response, 400) + + # ------------------------------------------------------------------ POST /shells + + @with_json_client + @with_xml_client + def test_shells_post_success(self, format_client: FormatClient): + example_shell = create_example_asset_administration_shell() + + response = format_client.post("/shells", obj=example_shell) + + self.assertEqual(201, response.status_code) + self.assertIsNotNone(self.object_store.get(example_shell.id, None)) + + @with_json_client + @with_xml_client + def test_shells_post_bad(self, format_client: FormatClient): + example_shell = create_example_asset_administration_shell() + example_shell.id = None # type: ignore + + response = format_client.post("/shells", obj=example_shell) + + self.assert_error(response, 400) + + @with_json_client + @with_xml_client + def test_shells_post_conflict(self, format_client: FormatClient): + example_shell = create_example_asset_administration_shell() + self.object_store.add(example_shell) + + response = format_client.post("/shells", obj=example_shell) + + self.assert_error(response, 409) + + # ------------------------------------------------------------------ GET /shells/$reference + + @with_json_client + @with_xml_client + def test_shells_reference_get(self, format_client: FormatClient): + self.object_store.update(self.two_shells_store()) + example_shell = next(iter(self.object_store)) + + response = format_client.get("/shells/$reference") + + self.assert_ok(response) + references = format_client.parse_collection(response) + self.assertEqual(2, len(references)) + self.assertIn(example_shell.id, [format_client.reference_target(ref) for ref in references]) + + @with_json_client + @with_xml_client + def test_shells_reference_get_supports_pagination(self, format_client: FormatClient): + self.object_store.update(self.two_shells_store()) + + pages = format_client.get_paginated("/shells/$reference", limit=1, max_pages=2) + + self.assertEqual([1, 1], [len(page) for page in pages]) + + # ------------------------------------------------------------------ GET /shells/ + + @with_json_client + @with_xml_client + def test_shell_get_success(self, format_client: FormatClient): + self.object_store.update(self.two_shells_store()) + example_shell = next(iter(self.object_store)) + + response = format_client.get(f"/shells/{base64url_encode(example_shell.id)}") + + self.assert_ok(response) + self.assertEqual(example_shell.id, format_client.identifier(format_client.parse_object(response))) + + @with_json_client + @with_xml_client + def test_shell_get_not_found(self, format_client: FormatClient): + response = format_client.get(f"/shells/{base64url_encode('https://example.org/unknown')}") + + self.assert_error(response, 404) + + # ------------------------------------------------------------------ GET /shells//$reference + + @with_json_client + @with_xml_client + def test_shell_reference_get(self, format_client: FormatClient): + self.object_store.update(self.two_shells_store()) + example_shell = next(iter(self.object_store)) + + response = format_client.get(f"/shells/{base64url_encode(example_shell.id)}/$reference") + + self.assert_ok(response) + self.assertEqual(example_shell.id, format_client.reference_target(format_client.parse_object(response))) + + # ------------------------------------------------------------------ PUT /shells/ + + @with_json_client + @with_xml_client + def test_shell_put_success(self, format_client: FormatClient): + self.object_store.add(create_example_asset_administration_shell()) + updated_shell = create_example_asset_administration_shell() + updated_shell.id_short = "UpdatedIdShort" + + response = format_client.put(f"/shells/{base64url_encode(updated_shell.id)}", obj=updated_shell) + + self.assertEqual(204, response.status_code) + retrieved_shell = self.object_store.get(updated_shell.id, None) + self.assertIsInstance(retrieved_shell, model.AssetAdministrationShell) + assert isinstance(retrieved_shell, model.AssetAdministrationShell) # make mypy happy + self.assertEqual("UpdatedIdShort", retrieved_shell.id_short) + + @with_json_client + @with_xml_client + def test_shell_put_not_found(self, format_client: FormatClient): + updated_shell = create_example_asset_administration_shell() + + response = format_client.put( + f"/shells/{base64url_encode('https://example.org/unknown')}", obj=updated_shell + ) + + self.assert_error(response, 404) + + # ------------------------------------------------------------------ DELETE /shells/ + + @with_json_client + @with_xml_client + def test_shell_delete_success(self, format_client: FormatClient): + example_shell = create_example_asset_administration_shell() + self.object_store.add(example_shell) + + response = format_client.delete(f"/shells/{base64url_encode(example_shell.id)}") + + self.assertEqual(204, response.status_code) + self.assertIsNone(self.object_store.get(example_shell.id, None)) + + @with_json_client + @with_xml_client + def test_shell_delete_not_found(self, format_client: FormatClient): + response = format_client.delete(f"/shells/{base64url_encode('https://example.org/unknown')}") + + self.assert_error(response, 404) + + # ------------------------------------------------------------------ GET /shells//asset-information + + @with_json_client + @with_xml_client + def test_shell_asset_information_get(self, format_client: FormatClient): + example_shell = create_example_asset_administration_shell() + self.object_store.add(example_shell) + + response = format_client.get(f"/shells/{base64url_encode(example_shell.id)}/asset-information") + + self.assert_ok(response) + self.assertEqual( + example_shell.asset_information.global_asset_id, + format_client.field(format_client.parse_object(response), "globalAssetId"), + ) + + # ------------------------------------------------------------------ PUT /shells//asset-information + + @with_json_client + @with_xml_client + def test_shell_asset_information_put(self, format_client: FormatClient): + example_shell = create_example_asset_administration_shell() + self.object_store.add(example_shell) + new_asset_information = model.AssetInformation( + asset_kind=model.AssetKind.INSTANCE, + global_asset_id="http://example.org/changed_asset", + ) + + response = format_client.put( + f"/shells/{base64url_encode(example_shell.id)}/asset-information", + obj=new_asset_information, + ) + + self.assertEqual(204, response.status_code) + retrieved_shell = self.object_store.get(example_shell.id) + self.assertIsInstance(retrieved_shell, model.AssetAdministrationShell) + assert isinstance(retrieved_shell, model.AssetAdministrationShell) # make mypy happy + self.assertEqual( + "http://example.org/changed_asset", + retrieved_shell.asset_information.global_asset_id, + ) + + # ------------------------------------------------------------------ GET /shells//submodel-refs + + @with_json_client + @with_xml_client + def test_shell_submodel_refs_get(self, format_client: FormatClient): + example_shell = create_example_asset_administration_shell() + self.object_store.add(example_shell) + + response = format_client.get(f"/shells/{base64url_encode(example_shell.id)}/submodel-refs") + + self.assert_ok(response) + references = format_client.parse_collection(response) + self.assertEqual(1, len(references)) + self.assertEqual( + "https://example.org/Test_Submodel_Missing", format_client.reference_target(references[0]) + ) + + @with_json_client + @with_xml_client + def test_shell_submodel_refs_get_supports_pagination(self, format_client: FormatClient): + example_shell = create_example_asset_administration_shell() + example_shell.submodel.add( + model.ModelReference( + (model.Key(model.KeyTypes.SUBMODEL, "https://example.org/Second_Submodel_Ref"),), model.Submodel + ) + ) + self.object_store.add(example_shell) + + pages = format_client.get_paginated( + f"/shells/{base64url_encode(example_shell.id)}/submodel-refs", limit=1, max_pages=2 + ) + + self.assertEqual([1, 1], [len(page) for page in pages]) + + # ------------------------------------------------------------------ POST /shells//submodel-refs + + @with_json_client + @with_xml_client + def test_shell_submodel_refs_post_success(self, format_client: FormatClient): + example_shell = create_example_asset_administration_shell() + self.object_store.add(example_shell) + new_ref = model.ModelReference( + (model.Key(model.KeyTypes.SUBMODEL, "https://example.org/NewSubmodel"),), model.Submodel + ) + + response = format_client.post(f"/shells/{base64url_encode(example_shell.id)}/submodel-refs", obj=new_ref) + + self.assertEqual(201, response.status_code) + retrieved_shell = self.object_store.get(example_shell.id) + self.assertIsInstance(retrieved_shell, model.AssetAdministrationShell) + assert isinstance(retrieved_shell, model.AssetAdministrationShell) # make mypy happy + identifiers = {ref.get_identifier() for ref in retrieved_shell.submodel} + self.assertIn("https://example.org/NewSubmodel", identifiers) + + @with_json_client + @with_xml_client + def test_shell_submodel_refs_post_conflict(self, format_client: FormatClient): + example_shell = create_example_asset_administration_shell() + self.object_store.add(example_shell) + existing_ref = model.ModelReference( + (model.Key(model.KeyTypes.SUBMODEL, "https://example.org/Test_Submodel_Missing"),), model.Submodel + ) + + response = format_client.post( + f"/shells/{base64url_encode(example_shell.id)}/submodel-refs", obj=existing_ref + ) + + self.assert_error(response, 409) + + # ------------------------------------------------------------------ DELETE /shells//submodel-refs/ + + @with_json_client + @with_xml_client + def test_shell_submodel_refs_delete_success(self, format_client: FormatClient): + example_shell = create_example_asset_administration_shell() + self.object_store.add(example_shell) + submodel_id = "https://example.org/Test_Submodel_Missing" + + response = format_client.delete( + f"/shells/{base64url_encode(example_shell.id)}/submodel-refs/{base64url_encode(submodel_id)}" + ) + + self.assertEqual(204, response.status_code) + retrieved_shell = self.object_store.get(example_shell.id) + self.assertIsInstance(retrieved_shell, model.AssetAdministrationShell) + assert isinstance(retrieved_shell, model.AssetAdministrationShell) # make mypy happy + self.assertEqual(0, len(list(retrieved_shell.submodel))) + + @with_json_client + @with_xml_client + def test_shell_submodel_refs_delete_not_found(self, format_client: FormatClient): + example_shell = create_example_asset_administration_shell() + self.object_store.add(example_shell) + + response = format_client.delete( + f"/shells/{base64url_encode(example_shell.id)}/submodel-refs/" + f"{base64url_encode('https://example.org/unknown')}" + ) + + self.assert_error(response, 404) + + # ------------------------------------------------------------------ PUT /shells//submodels/ + + @with_json_client + @with_xml_client + def test_shell_submodel_refs_submodel_put(self, format_client: FormatClient): + example_shell = create_example_asset_administration_shell() + self.object_store.add(example_shell) + self.object_store.add(create_example_submodel()) + updated_submodel = create_example_submodel() + updated_submodel.id_short = "UpdatedSubmodel" + + response = format_client.put( + f"/shells/{base64url_encode(example_shell.id)}/submodels/{base64url_encode(updated_submodel.id)}", + obj=updated_submodel, + ) + + self.assertEqual(204, response.status_code) + retrieved_sm = self.object_store.get(updated_submodel.id) + self.assertIsInstance(retrieved_sm, model.Submodel) + assert isinstance(retrieved_sm, model.Submodel) # make mypy happy + self.assertEqual("UpdatedSubmodel", retrieved_sm.id_short) + + @with_json_client + @with_xml_client + def test_shell_submodel_refs_submodel_put_changed_id_success(self, format_client: FormatClient): + example_shell = create_example_asset_administration_shell() + self.object_store.add(example_shell) + self.object_store.add(create_example_submodel()) + updated_submodel = create_example_submodel() + old_sm_id = updated_submodel.id + updated_submodel.id = "https://example.org/Test_Submodel_Updated" + + response = format_client.put( + f"/shells/{base64url_encode(example_shell.id)}/submodels/{base64url_encode(old_sm_id)}", + obj=updated_submodel, + ) + + self.assertEqual(204, response.status_code) + retrieved_shell = self.object_store.get(example_shell.id) + self.assertIsInstance(retrieved_shell, model.AssetAdministrationShell) + assert isinstance(retrieved_shell, model.AssetAdministrationShell) # make mypy happy + + self.assertIn(model.ModelReference.from_referable(updated_submodel), retrieved_shell.submodel) + self.assertNotIn(model.ModelReference.from_referable(create_example_submodel()), retrieved_shell.submodel) + + # ------------------------------------------------------------------ DELETE /shells//submodels/ + + @with_json_client + @with_xml_client + def test_shell_submodel_refs_submodel_delete(self, format_client: FormatClient): + example_shell = create_example_asset_administration_shell() + self.object_store.add(example_shell) + example_submodel = create_example_submodel() + self.object_store.add(example_submodel) + + response = format_client.delete( + f"/shells/{base64url_encode(example_shell.id)}/submodels/{base64url_encode(example_submodel.id)}" + ) + + self.assertEqual(204, response.status_code) + self.assertIsNone(self.object_store.get(example_submodel.id, None)) + retrieved_shell = self.object_store.get(example_shell.id) + self.assertIsInstance(retrieved_shell, model.AssetAdministrationShell) + assert isinstance(retrieved_shell, model.AssetAdministrationShell) # make mypy happy + self.assertEqual(0, len(list(retrieved_shell.submodel))) + + # ------------------------------------------------------------------ /shells//submodels/ redirect + + @with_json_client + @with_xml_client + def test_shell_submodel_refs_submodel_redirect(self, format_client: FormatClient): + example_shell = create_example_asset_administration_shell() + self.object_store.add(example_shell) + submodel_id = "https://example.org/Test_Submodel_Missing" + + response = format_client.get( + f"/shells/{base64url_encode(example_shell.id)}/submodels/{base64url_encode(submodel_id)}" + ) + + self.assertEqual(307, response.status_code) + self.assertIn(f"/submodels/{base64url_encode(submodel_id)}", response.headers["Location"]) + + @with_json_client + @with_xml_client + def test_shell_submodel_refs_submodel_redirect_with_path(self, format_client: FormatClient): + example_shell = create_example_asset_administration_shell() + self.object_store.add(example_shell) + submodel_id = "https://example.org/Test_Submodel_Missing" + + response = format_client.get( + f"/shells/{base64url_encode(example_shell.id)}/submodels/{base64url_encode(submodel_id)}" + f"/submodel-elements" + ) + + self.assertEqual(307, response.status_code) + self.assertTrue(response.headers["Location"].endswith("/submodel-elements")) diff --git a/server/test/interfaces/repository/test_submodels.py b/server/test/interfaces/repository/test_submodels.py new file mode 100644 index 000000000..0340019f0 --- /dev/null +++ b/server/test/interfaces/repository/test_submodels.py @@ -0,0 +1,1132 @@ +import io +import json +from unittest import mock + +from app.util.converters import base64url_encode +from basyx.aas import model +from basyx.aas.adapter.json import AASToJsonEncoder +from basyx.aas.examples.data.example_aas_missing_attributes import create_example_submodel + +from ..format_utils import ( + FormatClient, + inject_format_clients, + with_json_client, + with_xml_client, +) +from .helpers import RepositoryEndpointTestBase + + +def _encode_reference(reference: model.Reference) -> str: + return base64url_encode(json.dumps(reference, cls=AASToJsonEncoder)) + + +# semanticId carried by ``create_example_submodel()``. +EXAMPLE_SEMANTIC_ID = model.ExternalReference( + (model.Key(model.KeyTypes.GLOBAL_REFERENCE, "http://example.org/SubmodelTemplates/ExampleSubmodel"),) +) + + +@inject_format_clients +class SubmodelsEndpointsTest(RepositoryEndpointTestBase): + """ + Endpoint tests for the implemented ``/submodels`` routes of :class:`~app.interfaces.repository.WSGIApp` + that operate on the Submodel itself (the ``/submodel-elements`` subtree is covered separately). + + Bodies are written once against the format-agnostic ``format_client`` helper. + For each test two variants are generated where the :class:`~..format_utils.JsonFormatClient` and + :class:`~..format_utils.XmlFormatClient` are injected respectively. + """ + + __test__ = True + + SECOND_ID = "https://example.org/Test_Submodel_Second" + + def two_submodels_store(self) -> model.DictIdentifiableStore: + store: model.DictIdentifiableStore = model.DictIdentifiableStore() + store.add(create_example_submodel()) + second = create_example_submodel() + second.id = self.SECOND_ID + second.id_short = "SecondSubmodel" + store.add(second) + return store + + def _get_submodel_ids(self, format_client: FormatClient, query: str) -> set: + response = format_client.get(f"/submodels?{query}") + self.assert_ok(response) + return {format_client.identifier(node) for node in format_client.parse_collection(response)} + + # ------------------------------------------------------------------ GET /submodels + + @with_json_client + @with_xml_client + def test_submodels_get(self, format_client: FormatClient): + self.object_store.update(self.two_submodels_store()) + + response = format_client.get("/submodels") + + self.assert_ok(response) + self.assertEqual(2, len(format_client.parse_collection(response))) + + @with_json_client + @with_xml_client + def test_submodels_get_empty(self, format_client: FormatClient): + response = format_client.get("/submodels") + + self.assert_ok(response) + self.assertEqual(0, len(format_client.parse_collection(response))) + + @with_json_client + @with_xml_client + def test_submodels_get_supports_pagination(self, format_client: FormatClient): + self.object_store.update(self.two_submodels_store()) + + pages = format_client.get_paginated("/submodels", limit=1, max_pages=2) + + self.assertEqual([1, 1], [len(page) for page in pages]) + + # ------------------------------------------------------------------ GET /submodels?idShort=...&semanticId=... + + @with_json_client + @with_xml_client + def test_submodels_get_filter_by_id_short(self, format_client: FormatClient): + self.object_store.update(self.two_submodels_store()) + + ids = self._get_submodel_ids(format_client, "idShort=SecondSubmodel") + + self.assertEqual({self.SECOND_ID}, ids) + + @with_json_client + @with_xml_client + def test_submodels_get_filter_by_id_short_no_match(self, format_client: FormatClient): + self.object_store.update(self.two_submodels_store()) + + ids = self._get_submodel_ids(format_client, "idShort=Unknown") + + self.assertEqual(set(), ids) + + @with_json_client + @with_xml_client + def test_submodels_get_filter_by_semantic_id(self, format_client: FormatClient): + self.object_store.update(self.two_submodels_store()) + + ids = self._get_submodel_ids(format_client, f"semanticId={_encode_reference(EXAMPLE_SEMANTIC_ID)}") + + self.assertEqual({"https://example.org/Test_Submodel_Missing", self.SECOND_ID}, ids) + + @with_json_client + @with_xml_client + def test_submodels_get_filter_by_semantic_id_no_match(self, format_client: FormatClient): + self.object_store.update(self.two_submodels_store()) + other = model.ExternalReference((model.Key(model.KeyTypes.GLOBAL_REFERENCE, "https://example.org/other"),)) + + ids = self._get_submodel_ids(format_client, f"semanticId={_encode_reference(other)}") + + self.assertEqual(set(), ids) + + # ------------------------------------------------------------------ POST /submodels + + @with_json_client + @with_xml_client + def test_submodels_post_success(self, format_client: FormatClient): + example_submodel = create_example_submodel() + + response = format_client.post("/submodels", obj=example_submodel) + + self.assertEqual(201, response.status_code) + self.assertIsNotNone(self.object_store.get(example_submodel.id, None)) + + @with_json_client + @with_xml_client + def test_submodels_post_bad(self, format_client: FormatClient): + example_submodel = create_example_submodel() + example_submodel.id = None # type: ignore + + response = format_client.post("/submodels", obj=example_submodel) + + self.assert_error(response, 400) + + @with_json_client + @with_xml_client + def test_submodels_post_conflict(self, format_client: FormatClient): + example_submodel = create_example_submodel() + self.object_store.add(example_submodel) + + response = format_client.post("/submodels", obj=example_submodel) + + self.assert_error(response, 409) + + # ------------------------------------------------------------------ GET /submodels/$metadata + + @with_json_client + @with_xml_client + def test_submodels_metadata_get(self, format_client: FormatClient): + self.object_store.update(self.two_submodels_store()) + + response = format_client.get("/submodels/$metadata") + + self.assert_ok(response) + self.assertEqual(2, len(format_client.parse_collection(response))) + + @with_json_client + @with_xml_client + def test_submodels_metadata_get_supports_pagination(self, format_client: FormatClient): + self.object_store.update(self.two_submodels_store()) + + pages = format_client.get_paginated("/submodels/$metadata", limit=1, max_pages=2) + + self.assertEqual([1, 1], [len(page) for page in pages]) + + def test_submodels_metadata_get_rejects_level(self): + self.object_store.add(create_example_submodel()) + + response = self.client.get("/submodels/$metadata?level=deep") + + self.assert_error(response, 400) + + # ------------------------------------------------------------------ GET /submodels/$reference + + @with_json_client + @with_xml_client + def test_submodels_reference_get(self, format_client: FormatClient): + self.object_store.update(self.two_submodels_store()) + + response = format_client.get("/submodels/$reference") + + self.assert_ok(response) + references = format_client.parse_collection(response) + self.assertEqual(2, len(references)) + self.assertEqual( + {"https://example.org/Test_Submodel_Missing", self.SECOND_ID}, + {format_client.reference_target(ref) for ref in references}, + ) + + @with_json_client + @with_xml_client + def test_submodels_reference_get_supports_pagination(self, format_client: FormatClient): + self.object_store.update(self.two_submodels_store()) + + pages = format_client.get_paginated("/submodels/$reference", limit=1, max_pages=2) + + self.assertEqual([1, 1], [len(page) for page in pages]) + + # ------------------------------------------------------------------ GET /submodels/ + + @with_json_client + @with_xml_client + def test_submodel_get_success(self, format_client: FormatClient): + example_submodel = create_example_submodel() + self.object_store.add(example_submodel) + + response = format_client.get(f"/submodels/{base64url_encode(example_submodel.id)}") + + self.assert_ok(response) + self.assertEqual(example_submodel.id, format_client.identifier(format_client.parse_object(response))) + + @with_json_client + @with_xml_client + def test_submodel_get_not_found(self, format_client: FormatClient): + response = format_client.get(f"/submodels/{base64url_encode('https://example.org/unknown')}") + + self.assert_error(response, 404) + + def test_submodel_get_stripped_omits_submodel_elements(self): + example_submodel = create_example_submodel() + self.object_store.add(example_submodel) + path = f"/submodels/{base64url_encode(example_submodel.id)}" + + full = self.client.get(path) + stripped = self.client.get(f"{path}?level=core") + + self.assert_ok(full) + self.assert_ok(stripped) + self.assertIn("submodelElements", full.get_data(as_text=True)) + self.assertNotIn("submodelElements", stripped.get_data(as_text=True)) + + def test_submodel_get_invalid_level_returns_400(self): + example_submodel = create_example_submodel() + self.object_store.add(example_submodel) + + response = self.client.get(f"/submodels/{base64url_encode(example_submodel.id)}?level=bogus") + + self.assert_error(response, 400) + + def test_submodel_get_extent_not_implemented(self): + example_submodel = create_example_submodel() + self.object_store.add(example_submodel) + + response = self.client.get( + f"/submodels/{base64url_encode(example_submodel.id)}?extent=withBlobValue" + ) + + self.assert_error(response, 501) + + # ------------------------------------------------------------------ PUT /submodels/ + + @with_json_client + @with_xml_client + def test_submodel_put_success(self, format_client: FormatClient): + self.object_store.add(create_example_submodel()) + updated_submodel = create_example_submodel() + updated_submodel.id_short = "UpdatedIdShort" + + response = format_client.put( + f"/submodels/{base64url_encode(updated_submodel.id)}", obj=updated_submodel + ) + + self.assertEqual(204, response.status_code) + retrieved_submodel = self.object_store.get(updated_submodel.id, None) + self.assertIsInstance(retrieved_submodel, model.Submodel) + assert isinstance(retrieved_submodel, model.Submodel) # make mypy happy + self.assertEqual("UpdatedIdShort", retrieved_submodel.id_short) + + @with_json_client + @with_xml_client + def test_submodel_put_not_found(self, format_client: FormatClient): + updated_submodel = create_example_submodel() + + response = format_client.put( + f"/submodels/{base64url_encode('https://example.org/unknown')}", obj=updated_submodel + ) + + self.assert_error(response, 404) + + # ------------------------------------------------------------------ DELETE /submodels/ + + @with_json_client + @with_xml_client + def test_submodel_delete_success(self, format_client: FormatClient): + example_submodel = create_example_submodel() + self.object_store.add(example_submodel) + + response = format_client.delete(f"/submodels/{base64url_encode(example_submodel.id)}") + + self.assertEqual(204, response.status_code) + self.assertIsNone(self.object_store.get(example_submodel.id, None)) + + @with_json_client + @with_xml_client + def test_submodel_delete_not_found(self, format_client: FormatClient): + response = format_client.delete(f"/submodels/{base64url_encode('https://example.org/unknown')}") + + self.assert_error(response, 404) + + # ------------------------------------------------------------------ GET /submodels//$metadata + + @with_json_client + @with_xml_client + def test_submodel_metadata_get(self, format_client: FormatClient): + example_submodel = create_example_submodel() + self.object_store.add(example_submodel) + + response = format_client.get(f"/submodels/{base64url_encode(example_submodel.id)}/$metadata") + + self.assert_ok(response) + self.assertEqual( + example_submodel.id, format_client.identifier(format_client.parse_object(response)) + ) + + def test_submodel_metadata_get_omits_submodel_elements(self): + example_submodel = create_example_submodel() + self.object_store.add(example_submodel) + + response = self.client.get(f"/submodels/{base64url_encode(example_submodel.id)}/$metadata") + + self.assert_ok(response) + self.assertNotIn("submodelElements", response.get_data(as_text=True)) + + def test_submodel_metadata_get_rejects_level(self): + example_submodel = create_example_submodel() + self.object_store.add(example_submodel) + + response = self.client.get( + f"/submodels/{base64url_encode(example_submodel.id)}/$metadata?level=core" + ) + + self.assert_error(response, 400) + + def test_submodel_metadata_get_not_found(self): + response = self.client.get( + f"/submodels/{base64url_encode('https://example.org/unknown')}/$metadata" + ) + + self.assert_error(response, 404) + + # ------------------------------------------------------------------ GET /submodels//$reference + + @with_json_client + @with_xml_client + def test_submodel_reference_get(self, format_client: FormatClient): + example_submodel = create_example_submodel() + self.object_store.add(example_submodel) + + response = format_client.get(f"/submodels/{base64url_encode(example_submodel.id)}/$reference") + + self.assert_ok(response) + self.assertEqual( + example_submodel.id, format_client.reference_target(format_client.parse_object(response)) + ) + + @with_json_client + @with_xml_client + def test_submodel_reference_get_not_found(self, format_client: FormatClient): + response = format_client.get( + f"/submodels/{base64url_encode('https://example.org/unknown')}/$reference" + ) + + self.assert_error(response, 404) + + +# ExampleSubmodelCollection (a nested namespace) and one of its children, carried by +# ``create_example_submodel()``; used as ready-made id_short paths in the tests below. +NESTED_COLLECTION = "ExampleSubmodelCollection" +NESTED_PROPERTY = "ExampleSubmodelCollection.ExampleProperty" +NESTED_BLOB = "ExampleSubmodelCollection.ExampleBlob" +NESTED_FILE = "ExampleSubmodelCollection.ExampleFile" +EXAMPLE_QUALIFIER_TYPE = "http://example.org/Qualifier/ExampleQualifier" + + +@inject_format_clients +class SubmodelElementsEndpointsTest(RepositoryEndpointTestBase): + """ + Endpoint tests for the ``/submodels//submodel-elements`` subtree of + :class:`~app.interfaces.repository.WSGIApp`, including the ``$metadata`` / ``$reference`` modifiers, + the ``attachment`` file routes and the ``qualifiers`` routes. + + Bodies are written once against the format-agnostic ``format_client`` helper. + For each test two variants are generated where the :class:`~..format_utils.JsonFormatClient` and + :class:`~..format_utils.XmlFormatClient` are injected respectively. + """ + + __test__ = True + + #: Number of top-level submodel elements in ``create_example_submodel()``. + TOP_LEVEL_COUNT = 6 + + def add_example_submodel(self) -> model.Submodel: + submodel = create_example_submodel() + self.object_store.add(submodel) + return submodel + + def _stored_submodel(self, submodel_id: str) -> model.Submodel: + submodel = self.object_store.get(submodel_id) + assert isinstance(submodel, model.Submodel) + return submodel + + @staticmethod + def elements_path(submodel_id: str, id_short_path: str = "") -> str: + base = f"/submodels/{base64url_encode(submodel_id)}/submodel-elements" + return f"{base}/{id_short_path}" if id_short_path else base + + def _nested_element(self, submodel: model.Submodel, id_short_path: str) -> model.SubmodelElement: + referable: model.Referable = submodel + for id_short in id_short_path.split("."): + assert isinstance(referable, model.UniqueIdShortNamespace) + referable = referable.get_referable(id_short) + assert isinstance(referable, model.SubmodelElement) + return referable + + def _nested_property(self, submodel: model.Submodel, id_short_path: str) -> model.Property: + element = self._nested_element(submodel, id_short_path) + assert isinstance(element, model.Property) + return element + + def _nested_file(self, submodel: model.Submodel, id_short_path: str) -> model.File: + element = self._nested_element(submodel, id_short_path) + assert isinstance(element, model.File) + return element + + def _nested_blob(self, submodel: model.Submodel, id_short_path: str) -> model.Blob: + element = self._nested_element(submodel, id_short_path) + assert isinstance(element, model.Blob) + return element + + # ------------------------------------------------------------------ GET .../submodel-elements + + @with_json_client + @with_xml_client + def test_submodel_elements_get(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.get(self.elements_path(submodel.id)) + + self.assert_ok(response) + self.assertEqual(self.TOP_LEVEL_COUNT, len(format_client.parse_collection(response))) + + @with_json_client + @with_xml_client + def test_submodel_elements_get_pagination_limit(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.get(f"{self.elements_path(submodel.id)}?limit=2") + + self.assert_ok(response) + self.assertEqual(2, len(format_client.parse_collection(response))) + + @with_json_client + @with_xml_client + def test_submodel_elements_get_supports_pagination(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + pages = format_client.get_paginated(self.elements_path(submodel.id), limit=3, max_pages=2) + + self.assertEqual([3, 3], [len(page) for page in pages]) + + @with_json_client + @with_xml_client + def test_submodel_elements_get_submodel_not_found(self, format_client: FormatClient): + response = format_client.get(self.elements_path("https://example.org/unknown")) + + self.assert_error(response, 404) + + # ------------------------------------------------------------------ POST .../submodel-elements + + @with_json_client + @with_xml_client + def test_submodel_elements_post_success(self, format_client: FormatClient): + submodel = self.add_example_submodel() + new_element = model.Property("NewProperty", model.datatypes.String, "some-value") + + response = format_client.post(self.elements_path(submodel.id), obj=new_element) + + self.assertEqual(201, response.status_code) + self.assertIn("Location", response.headers) + retrieved = self._stored_submodel(submodel.id) + self.assertIsInstance(retrieved.get_referable("NewProperty"), model.Property) + + @with_json_client + @with_xml_client + def test_submodel_elements_post_conflict(self, format_client: FormatClient): + submodel = self.add_example_submodel() + duplicate = model.Property("ExampleCapability", model.datatypes.String, "v") + + response = format_client.post(self.elements_path(submodel.id), obj=duplicate) + + self.assert_error(response, 409) + + @with_json_client + @with_xml_client + def test_submodel_elements_post_submodel_not_found(self, format_client: FormatClient): + new_element = model.Property("NewProperty", model.datatypes.String, "v") + + response = format_client.post( + self.elements_path("https://example.org/unknown"), obj=new_element + ) + + self.assert_error(response, 404) + + # ------------------------------------------------------------------ GET .../submodel-elements/$metadata + + @with_json_client + @with_xml_client + def test_submodel_elements_metadata_get(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.get(f"{self.elements_path(submodel.id)}/$metadata") + + self.assert_ok(response) + self.assertEqual(self.TOP_LEVEL_COUNT, len(format_client.parse_collection(response))) + + @with_json_client + @with_xml_client + def test_submodel_elements_metadata_get_supports_pagination(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + pages = format_client.get_paginated(f"{self.elements_path(submodel.id)}/$metadata", limit=3, max_pages=2) + + self.assertEqual([3, 3], [len(page) for page in pages]) + + def test_submodel_elements_metadata_get_rejects_level(self): + submodel = self.add_example_submodel() + + response = self.client.get(f"{self.elements_path(submodel.id)}/$metadata?level=deep") + + self.assert_error(response, 400) + + # ------------------------------------------------------------------ GET .../submodel-elements/$reference + + @with_json_client + @with_xml_client + def test_submodel_elements_reference_get(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.get(f"{self.elements_path(submodel.id)}/$reference") + + self.assert_ok(response) + references = format_client.parse_collection(response) + self.assertEqual(self.TOP_LEVEL_COUNT, len(references)) + self.assertIn( + "ExampleCapability", {format_client.reference_target(ref) for ref in references} + ) + + @with_json_client + @with_xml_client + def test_submodel_elements_reference_get_supports_pagination(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + pages = format_client.get_paginated(f"{self.elements_path(submodel.id)}/$reference", limit=3, max_pages=2) + + self.assertEqual([3, 3], [len(page) for page in pages]) + + # ------------------------------------------------------------------ GET .../submodel-elements/ + + @with_json_client + @with_xml_client + def test_submodel_element_get_top_level(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.get(self.elements_path(submodel.id, "ExampleCapability")) + + self.assert_ok(response) + self.assertEqual( + "ExampleCapability", format_client.field(format_client.parse_object(response), "idShort") + ) + + @with_json_client + @with_xml_client + def test_submodel_element_get_nested(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.get(self.elements_path(submodel.id, NESTED_PROPERTY)) + + self.assert_ok(response) + self.assertEqual( + "ExampleProperty", format_client.field(format_client.parse_object(response), "idShort") + ) + + @with_json_client + @with_xml_client + def test_submodel_element_get_not_found(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.get(self.elements_path(submodel.id, "DoesNotExist")) + + self.assert_error(response, 404) + + @with_json_client + @with_xml_client + def test_submodel_element_get_nested_not_found(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.get( + self.elements_path(submodel.id, f"{NESTED_COLLECTION}.DoesNotExist") + ) + + self.assert_error(response, 404) + + def test_submodel_element_get_path_through_non_namespace_returns_400(self): + submodel = self.add_example_submodel() + + response = self.client.get(self.elements_path(submodel.id, f"{NESTED_PROPERTY}.Child")) + + self.assert_error(response, 400) + + def test_submodel_element_get_malformed_id_short_path_returns_400(self): + submodel = self.add_example_submodel() + + response = self.client.get(self.elements_path(submodel.id, "a..b")) + + self.assert_error(response, 400) + + # ------------------------------------------------------------------ POST .../submodel-elements/ + + @with_json_client + @with_xml_client + def test_submodel_element_post_child_success(self, format_client: FormatClient): + submodel = self.add_example_submodel() + new_element = model.Property("AddedChild", model.datatypes.String, "v") + + response = format_client.post( + self.elements_path(submodel.id, NESTED_COLLECTION), obj=new_element + ) + + self.assertEqual(201, response.status_code) + retrieved = self._stored_submodel(submodel.id) + self.assertIsInstance( + self._nested_element(retrieved, f"{NESTED_COLLECTION}.AddedChild"), model.Property + ) + + @with_json_client + @with_xml_client + def test_submodel_element_post_child_conflict(self, format_client: FormatClient): + submodel = self.add_example_submodel() + duplicate = model.Property("ExampleProperty", model.datatypes.String, "v") + + response = format_client.post( + self.elements_path(submodel.id, NESTED_COLLECTION), obj=duplicate + ) + + self.assert_error(response, 409) + + def test_submodel_element_post_into_non_namespace_returns_400(self): + submodel = self.add_example_submodel() + payload = json.dumps( + model.Property("Child", model.datatypes.String, "v"), cls=AASToJsonEncoder + ) + + response = self.client.post( + self.elements_path(submodel.id, NESTED_PROPERTY), + data=payload, + content_type="application/json", + ) + + self.assert_error(response, 400) + + # ------------------------------------------------------------------ PUT .../submodel-elements/ + + @with_json_client + @with_xml_client + def test_submodel_element_put_success(self, format_client: FormatClient): + submodel = self.add_example_submodel() + updated = model.Property("ExampleProperty", model.datatypes.String, "updated-value") + + response = format_client.put( + self.elements_path(submodel.id, NESTED_PROPERTY), obj=updated + ) + + self.assertEqual(204, response.status_code) + retrieved = self._nested_property(self._stored_submodel(submodel.id), NESTED_PROPERTY) + self.assertEqual("updated-value", retrieved.value) + + @with_json_client + @with_xml_client + def test_submodel_element_put_not_found(self, format_client: FormatClient): + submodel = self.add_example_submodel() + updated = model.Property("DoesNotExist", model.datatypes.String, "v") + + response = format_client.put( + self.elements_path(submodel.id, "DoesNotExist"), obj=updated + ) + + self.assert_error(response, 404) + + # ------------------------------------------------------------------ DELETE .../submodel-elements/ + + @with_json_client + @with_xml_client + def test_submodel_element_delete_success(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.delete(self.elements_path(submodel.id, NESTED_PROPERTY)) + + self.assertEqual(204, response.status_code) + follow_up = format_client.get(self.elements_path(submodel.id, NESTED_PROPERTY)) + self.assert_error(follow_up, 404) + + @with_json_client + @with_xml_client + def test_submodel_element_delete_not_found(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.delete(self.elements_path(submodel.id, "DoesNotExist")) + + self.assert_error(response, 404) + + # ------------------------------------------------------------------ GET ...//$metadata + + @with_json_client + @with_xml_client + def test_submodel_element_metadata_get(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.get(f"{self.elements_path(submodel.id, NESTED_PROPERTY)}/$metadata") + + self.assert_ok(response) + self.assertEqual( + "ExampleProperty", format_client.field(format_client.parse_object(response), "idShort") + ) + + def test_submodel_element_metadata_get_rejects_capability(self): + submodel = self.add_example_submodel() + + response = self.client.get( + f"{self.elements_path(submodel.id, 'ExampleCapability')}/$metadata" + ) + + self.assert_error(response, 400) + + def test_submodel_element_metadata_get_rejects_level(self): + submodel = self.add_example_submodel() + + response = self.client.get( + f"{self.elements_path(submodel.id, NESTED_PROPERTY)}/$metadata?level=core" + ) + + self.assert_error(response, 400) + + # ------------------------------------------------------------------ GET ...//$reference + + @with_json_client + @with_xml_client + def test_submodel_element_reference_get(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.get(f"{self.elements_path(submodel.id, NESTED_PROPERTY)}/$reference") + + self.assert_ok(response) + self.assertEqual( + "ExampleProperty", format_client.reference_target(format_client.parse_object(response)) + ) + + @with_json_client + @with_xml_client + def test_submodel_element_reference_get_not_found(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.get( + f"{self.elements_path(submodel.id, 'DoesNotExist')}/$reference" + ) + + self.assert_error(response, 404) + + # ------------------------------------------------------------------ ...//attachment + + def test_submodel_element_attachment_get_blob(self): + submodel = self.add_example_submodel() + + response = self.client.get(f"{self.elements_path(submodel.id, NESTED_BLOB)}/attachment") + + self.assertEqual(200, response.status_code) + self.assertEqual("application/pdf", response.mimetype) + self.assertEqual(bytes([1, 2, 3, 4, 5]), response.get_data()) + + def test_submodel_element_attachment_get_file(self): + submodel = self.add_example_submodel() + self._nested_file(submodel, NESTED_FILE).value = "/TestFile.pdf" + self._nested_file(submodel, NESTED_FILE).content_type = "application/pdf" + + self.file_store.write_file.side_effect = lambda name, stream: stream.write(b"file-content") + + response = self.client.get(f"{self.elements_path(submodel.id, NESTED_FILE)}/attachment") + + self.assertEqual(200, response.status_code) + self.assertEqual("application/pdf", response.content_type) + self.file_store.write_file.assert_any_call("/TestFile.pdf", mock.ANY) + self.assertEqual(b"file-content", response.data) + + def test_submodel_element_attachment_get_on_non_file_returns_400(self): + submodel = self.add_example_submodel() + + response = self.client.get(f"{self.elements_path(submodel.id, NESTED_PROPERTY)}/attachment") + + self.assert_error(response, 400) + + def test_submodel_element_attachment_get_file_without_value_returns_404(self): + submodel = create_example_submodel() + self._nested_file(submodel, NESTED_FILE).value = None + self.object_store.add(submodel) + + response = self.client.get(f"{self.elements_path(submodel.id, NESTED_FILE)}/attachment") + + self.assert_error(response, 404) + + def test_submodel_element_attachment_get_external_file_return_400(self): + submodel = create_example_submodel() + self._nested_file(submodel, NESTED_FILE).value = "C:\\Users\\Test\\File.pdf" + self.object_store.add(submodel) + + response = self.client.get(f"{self.elements_path(submodel.id, NESTED_FILE)}/attachment") + + self.assert_error(response, 400) + + def test_submodel_element_attachment_get_missing_file_return_404(self): + submodel = create_example_submodel() + self.object_store.add(submodel) + + self.file_store.write_file.side_effect = KeyError + response = self.client.get(f"{self.elements_path(submodel.id, NESTED_FILE)}/attachment") + + self.assert_error(response, 404) + + def test_submodel_element_attachment_put_success(self): + submodel = create_example_submodel() + self._nested_file(submodel, NESTED_FILE).value = None + self.object_store.add(submodel) + self.file_store.add_file.return_value = "/uploaded.pdf" + + response = self.client.put( + f"{self.elements_path(submodel.id, NESTED_FILE)}/attachment", + data={ + "fileName": "/uploaded.pdf", + "file": (io.BytesIO(b"pdf-bytes"), "uploaded.pdf", "application/pdf"), + }, + content_type="multipart/form-data", + ) + + self.assertEqual(204, response.status_code) + self.file_store.add_file.assert_called_once_with("/uploaded.pdf", mock.ANY, "application/pdf") + self.assertEqual( + "/uploaded.pdf", self._nested_file(self._stored_submodel(submodel.id), NESTED_FILE).value + ) + + def test_submodel_element_attachment_put_conflict_when_value_present(self): + submodel = self.add_example_submodel() + + response = self.client.put( + f"{self.elements_path(submodel.id, NESTED_FILE)}/attachment", + data={ + "fileName": "/uploaded.pdf", + "file": (io.BytesIO(b"pdf-bytes"), "uploaded.pdf", "application/pdf"), + }, + content_type="multipart/form-data", + ) + + self.assert_error(response, 409) + + def test_submodel_element_attachment_put_on_non_file_returns_400(self): + submodel = self.add_example_submodel() + + response = self.client.put( + f"{self.elements_path(submodel.id, NESTED_PROPERTY)}/attachment", + data={"fileName": "/x.pdf", "file": (io.BytesIO(b"x"), "x.pdf", "application/pdf")}, + content_type="multipart/form-data", + ) + + self.assert_error(response, 400) + + def test_submodel_element_attachment_put_missing_filename_returns_400(self): + submodel = self.add_example_submodel() + self._nested_file(submodel, NESTED_FILE).value = None + + response = self.client.put( + f"{self.elements_path(submodel.id, NESTED_FILE)}/attachment", + data={"file": (io.BytesIO(b"x"), "x.pdf", "application/pdf")}, + content_type="multipart/form-data", + ) + + self.assert_error(response, 400) + + def test_submodel_element_attachment_put_external_filename_returns_400(self): + submodel = self.add_example_submodel() + self._nested_file(submodel, NESTED_FILE).value = None + + response = self.client.put( + f"{self.elements_path(submodel.id, NESTED_FILE)}/attachment", + data={"fileName": "C:\\Users\\Test\\x.pdf", "file": (io.BytesIO(b"x"), "x.pdf", "application/pdf")}, + content_type="multipart/form-data", + ) + + self.assert_error(response, 400) + + def test_submodel_element_attachment_put_missing_file_returns_400(self): + submodel = self.add_example_submodel() + self._nested_file(submodel, NESTED_FILE).value = None + + response = self.client.put( + f"{self.elements_path(submodel.id, NESTED_FILE)}/attachment", + data={"fileName": "/x.pdf"}, + content_type="multipart/form-data", + ) + + self.assert_error(response, 400) + + def test_submodel_element_attachment_put_mimetype_mismatch_returns_400(self): + submodel = self.add_example_submodel() + self._nested_file(submodel, NESTED_FILE).value = None + + response = self.client.put( + f"{self.elements_path(submodel.id, NESTED_FILE)}/attachment", + data={"fileName": "/x.pdf", "file": (io.BytesIO(b"x"), "x.pdf", "application/xml")}, + content_type="multipart/form-data", + ) + + self.assert_error(response, 415) + + def test_submodel_element_attachment_delete_blob(self): + submodel = self.add_example_submodel() + + response = self.client.delete(f"{self.elements_path(submodel.id, NESTED_BLOB)}/attachment") + + self.assertEqual(204, response.status_code) + self.assertIsNone(self._nested_blob(self._stored_submodel(submodel.id), NESTED_BLOB).value) + + def test_submodel_element_attachment_delete_file(self): + submodel = self.add_example_submodel() + self._nested_file(submodel, NESTED_FILE).value = "/TestFile.pdf" + + response = self.client.delete(f"{self.elements_path(submodel.id, NESTED_FILE)}/attachment") + + self.assertEqual(204, response.status_code) + self.file_store.delete_file.assert_any_call("/TestFile.pdf") + self.assertIsNone(self._nested_file(self._stored_submodel(submodel.id), NESTED_FILE).value) + + def test_submodel_element_attachment_delete_file_ignores_store_error(self): + submodel = self.add_example_submodel() + self._nested_file(submodel, NESTED_FILE).value = "/TestFile.pdf" + + self.file_store.delete_file.side_effect = KeyError + + response = self.client.delete(f"{self.elements_path(submodel.id, NESTED_FILE)}/attachment") + + self.assertEqual(204, response.status_code) + self.file_store.delete_file.assert_any_call("/TestFile.pdf") + self.assertIsNone(self._nested_file(self._stored_submodel(submodel.id), NESTED_FILE).value) + + def test_submodel_element_attachment_delete_on_non_attachment_returns_400(self): + submodel = self.add_example_submodel() + + response = self.client.delete( + f"{self.elements_path(submodel.id, NESTED_PROPERTY)}/attachment" + ) + + self.assert_error(response, 400) + + def test_submodel_element_attachment_delete_no_value_returns_404(self): + submodel = self.add_example_submodel() + self._nested_blob(submodel, NESTED_BLOB).value = None + + response = self.client.delete( + f"{self.elements_path(submodel.id, NESTED_BLOB)}/attachment" + ) + + self.assert_error(response, 404) + + def test_submodel_element_attachment_delete_external_file_returns_400(self): + submodel = self.add_example_submodel() + self._nested_file(submodel, NESTED_FILE).value = "C:\\Users\\Test\\x.pdf" + + response = self.client.delete(f"{self.elements_path(submodel.id, NESTED_FILE)}/attachment") + + self.assertEqual(400, response.status_code) + + # ------------------------------------------------------------------ ...//qualifiers + + @with_json_client + @with_xml_client + def test_submodel_element_qualifiers_get_list(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.get(f"{self.elements_path(submodel.id, NESTED_PROPERTY)}/qualifiers") + + self.assert_ok(response) + types = {format_client.field(node, "type") for node in format_client.parse_collection(response)} + self.assertEqual({EXAMPLE_QUALIFIER_TYPE}, types) + + @with_json_client + @with_xml_client + def test_submodel_element_qualifier_get_by_type(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.get( + f"{self.elements_path(submodel.id, NESTED_PROPERTY)}/qualifiers/" + f"{base64url_encode(EXAMPLE_QUALIFIER_TYPE)}" + ) + + self.assert_ok(response) + self.assertEqual( + EXAMPLE_QUALIFIER_TYPE, format_client.field(format_client.parse_object(response), "type") + ) + + @with_json_client + @with_xml_client + def test_submodel_element_qualifier_get_by_type_not_found(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.get( + f"{self.elements_path(submodel.id, NESTED_PROPERTY)}/qualifiers/" + f"{base64url_encode('urn:unknown-qualifier')}" + ) + + self.assert_error(response, 404) + + @with_json_client + @with_xml_client + def test_submodel_element_qualifiers_post_success(self, format_client: FormatClient): + submodel = self.add_example_submodel() + qualifier = model.Qualifier("AddedQualifier", model.datatypes.String, "v") + + response = format_client.post( + f"{self.elements_path(submodel.id, NESTED_PROPERTY)}/qualifiers", obj=qualifier + ) + + self.assertEqual(201, response.status_code) + retrieved = self._nested_element(self._stored_submodel(submodel.id), NESTED_PROPERTY) + self.assertTrue(retrieved.qualifier.contains_id("type", "AddedQualifier")) + + @with_json_client + @with_xml_client + def test_submodel_element_qualifiers_post_conflict(self, format_client: FormatClient): + submodel = self.add_example_submodel() + qualifier = model.Qualifier(EXAMPLE_QUALIFIER_TYPE, model.datatypes.String, "v") + + response = format_client.post( + f"{self.elements_path(submodel.id, NESTED_PROPERTY)}/qualifiers", obj=qualifier + ) + + self.assert_error(response, 409) + + @with_json_client + @with_xml_client + def test_submodel_element_qualifier_put_success(self, format_client: FormatClient): + submodel = self.add_example_submodel() + updated = model.Qualifier(EXAMPLE_QUALIFIER_TYPE, model.datatypes.String, "changed-value") + + response = format_client.put( + f"{self.elements_path(submodel.id, NESTED_PROPERTY)}/qualifiers/" + f"{base64url_encode(EXAMPLE_QUALIFIER_TYPE)}", + obj=updated, + ) + + self.assert_ok(response) + retrieved = self._nested_element(self._stored_submodel(submodel.id), NESTED_PROPERTY) + self.assertEqual("changed-value", retrieved.get_qualifier_by_type(EXAMPLE_QUALIFIER_TYPE).value) + + @with_json_client + @with_xml_client + def test_submodel_element_qualifier_put_changed_type_success(self, format_client: FormatClient): + submodel = self.add_example_submodel() + new_type = "http://example.org/Qualifier/ExampleQualifier_Changed" + updated = model.Qualifier(new_type, model.datatypes.String, "changed-value") + + response = format_client.put( + f"{self.elements_path(submodel.id, NESTED_PROPERTY)}/qualifiers/{base64url_encode(EXAMPLE_QUALIFIER_TYPE)}", + obj=updated, + ) + + self.assertEqual(201, response.status_code) + retrieved = self._nested_element(self._stored_submodel(submodel.id), NESTED_PROPERTY) + self.assertTrue(retrieved.qualifier.contains_id("type", new_type)) + self.assertEqual(new_type, format_client.field(format_client.parse_object(response), "type")) + + @with_json_client + @with_xml_client + def test_submodel_element_qualifier_put_conflict(self, format_client: FormatClient): + submodel = self.add_example_submodel() + new_type = "http://example.org/Qualifier/ExampleQualifier_Changed" + self._nested_property(submodel, NESTED_PROPERTY).qualifier.add( + model.Qualifier(type_=new_type, value_type=model.datatypes.String, value="test") + ) + self.object_store.commit(submodel) + + updated = model.Qualifier(new_type, model.datatypes.String, "changed-value") + response = format_client.put( + f"{self.elements_path(submodel.id, NESTED_PROPERTY)}/qualifiers/{base64url_encode(EXAMPLE_QUALIFIER_TYPE)}", + obj=updated, + ) + + self.assertEqual(409, response.status_code) + + @with_json_client + @with_xml_client + def test_submodel_element_qualifier_delete_success(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.delete( + f"{self.elements_path(submodel.id, NESTED_PROPERTY)}/qualifiers/" + f"{base64url_encode(EXAMPLE_QUALIFIER_TYPE)}" + ) + + self.assertEqual(204, response.status_code) + retrieved = self._nested_element(self._stored_submodel(submodel.id), NESTED_PROPERTY) + self.assertFalse(retrieved.qualifier.contains_id("type", EXAMPLE_QUALIFIER_TYPE)) + + @with_json_client + @with_xml_client + def test_submodel_element_qualifier_delete_not_found(self, format_client: FormatClient): + submodel = self.add_example_submodel() + + response = format_client.delete( + f"{self.elements_path(submodel.id, NESTED_PROPERTY)}/qualifiers/" + f"{base64url_encode('urn:unknown-qualifier')}" + ) + + self.assert_error(response, 404) diff --git a/server/test/interfaces/test_base.py b/server/test/interfaces/test_base.py new file mode 100644 index 000000000..fc695911e --- /dev/null +++ b/server/test/interfaces/test_base.py @@ -0,0 +1,265 @@ +import datetime +import json +import re +import unittest +from typing import Optional +from unittest import mock + +from app.interfaces import base +from basyx.aas import model +from basyx.aas.adapter._generic import XML_NS_MAP +from lxml import etree +from werkzeug.exceptions import BadRequest + + +class TestPagination(unittest.TestCase): + """ + Testing of the shared pagination strategy, shared by multiple endpoints. + """ + + __test__ = True + + @staticmethod + def _build_request(limit: Optional[str] = None, cursor: Optional[str] = None): + request = mock.Mock() + def mock_get(key, default = None): + if key == "limit": + return limit or default + elif key == "cursor": + return cursor or default + else: + return default + request.args.get.side_effect = mock_get + return request + + def test_pagination_on_empty_set(self): + page, metadata = base.BaseWSGIApp._get_slice(self._build_request('3'), []) + self.assertEqual(0, len(list(page))) + if not metadata: + self.fail("no metadata") + self.assertIsNone(metadata.cursor) + + def test_pagination_with_one_page(self): + page, metadata = base.BaseWSGIApp._get_slice(self._build_request("3"), [1, 2]) + self.assertEqual([1, 2], list(page)) + if not metadata: + self.fail("no metadata") + self.assertIsNone(metadata.cursor) + + def test_pagination_walks_all_items(self): + all_objects = [i for i in range(20)] + + pages: list[list] = [] + cursor = None + for _ in range(1000): + page, metadata = base.BaseWSGIApp._get_slice(self._build_request('6', cursor), all_objects) + result = list(page) + pages.append(result) + if not metadata: + self.fail("no paging_metadata returned") + cursor = metadata.cursor + if cursor is None: + break + if len(pages) > 4 or len(pages) < 1: + self.fail("cursor never signalled end") + + self.assertEqual([6, 6, 6, 2], [len(page) for page in pages]) + seen = [i for page in pages for i in page] + self.assertEqual(len(seen), len(set(seen)), "an item was returned on more than one page") + self.assertEqual(set(range(20)), set(seen)) + + def test_pagination_requires_positive_limit(self): + with self.assertRaises(BadRequest): + base.BaseWSGIApp._get_slice(self._build_request('-1'), []) + + def test_pagination_requires_positive_cursor(self): + with self.assertRaises(BadRequest): + base.BaseWSGIApp._get_slice(self._build_request('5', '-1'), []) + + +class TestJsonResponse(unittest.TestCase): + + __test__ = True + + def test_empty_response(self): + response = base.JsonResponse(None) + self.assertEqual(204, response.status_code) + self.assertEqual(0, len(response.get_data())) + + def test_example_single_object(self): + response = base.JsonResponse(model.Submodel(id_="https://example.org/Example_Submodel")) + + self.assertEqual("application/json", response.content_type) + self.assertEqual(200, response.status_code) + parsed_body = json.loads(response.get_data(as_text=True)) + if not isinstance(parsed_body, dict): + self.fail("Response is no JSON object") + self.assertEqual("Submodel", parsed_body["modelType"]) + self.assertEqual("https://example.org/Example_Submodel", parsed_body["id"]) + + def test_example_list(self): + response = base.JsonResponse([ + model.Submodel(id_="https://example.org/Example_Submodel"), + model.Submodel(id_="https://example.org/Second_Submodel") + ]) + + self.assertEqual("application/json", response.content_type) + self.assertEqual(200, response.status_code) + parsed_body = json.loads(response.get_data(as_text=True)) + if not isinstance(parsed_body, list): + self.fail("Response is no JSON list") + ids = [sm["id"] for sm in parsed_body] + self.assertEqual(2, len(ids)) + self.assertEqual(len(ids), len(set(ids))) + + def test_example_empty_list(self): + response = base.JsonResponse([]) + + self.assertEqual("application/json", response.content_type) + self.assertEqual(200, response.status_code) + parsed_body = json.loads(response.get_data(as_text=True)) + + if not isinstance(parsed_body, list): + self.fail("Response is no JSON list") + self.assertEqual(0, len(parsed_body)) + + def test_paging_metadata(self): + response = base.JsonResponse( + obj=[model.Submodel(id_="https://example.org/Example_Submodel")], + paging_metadata=base.PagingMetadata(cursor="asdf") + ) + + self.assertEqual("application/json", response.content_type) + self.assertEqual(200, response.status_code) + parsed_body = json.loads(response.get_data(as_text=True)) + if not isinstance(parsed_body, dict): + self.fail("Response is no JSON object") + self.assertEqual("asdf", parsed_body["paging_metadata"]["cursor"]) + if not isinstance(parsed_body["result"], list): + self.fail("Result part contains no list") + self.assertEqual("https://example.org/Example_Submodel", parsed_body["result"][0]["id"]) + + def test_paging_metadata_no_cursor(self): + response = base.JsonResponse( + obj=[model.Submodel(id_="https://example.org/Example_Submodel")], + paging_metadata=base.PagingMetadata(cursor=None), + ) + + self.assertEqual("application/json", response.content_type) + self.assertEqual(200, response.status_code) + parsed_body = json.loads(response.get_data(as_text=True)) + if not isinstance(parsed_body, dict): + self.fail("Response is no JSON object") + self.assertIn("paging_metadata", parsed_body) + self.assertNotIn("cursor", parsed_body["paging_metadata"]) + + def test_result(self): + response = base.JsonResponse( + base.Result(False, [ + base.Message("BAD_CODE", "test", base.MessageType.ERROR, datetime.datetime.now()) + ]) + ) + + self.assertEqual("application/json", response.content_type) + self.assertEqual(200, response.status_code) + parsed_body = json.loads(response.get_data(as_text=True)) + if not isinstance(parsed_body, dict): + self.fail("Response is no JSON object") + self.assertFalse(parsed_body["success"]) + if not isinstance(parsed_body["messages"], list): + self.fail("messages is no list") + message = parsed_body["messages"][0] + self.assertEqual("BAD_CODE", message["code"]) + self.assertEqual("test", message["text"]) + self.assertEqual("Error", message["messageType"]) + isodatetime = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}$") + self.assertIsNotNone(isodatetime.match(message["timestamp"])) + +class TestXmlResponse(unittest.TestCase): + __test__ = True + + def test_empty_response(self): + response = base.XmlResponse(None) + self.assertEqual(204, response.status_code) + self.assertEqual(0, len(response.get_data())) + + def test_example_single_object(self): + response = base.XmlResponse(model.Submodel(id_="https://example.org/Example_Submodel")) + + self.assertEqual("application/xml", response.content_type) + self.assertEqual(200, response.status_code) + parsed_body = etree.fromstring(response.get_data()) + + returned_id = parsed_body.findtext("aas:id", namespaces=XML_NS_MAP) + self.assertEqual("https://example.org/Example_Submodel", returned_id) + + def test_example_list(self): + response = base.XmlResponse( + [ + model.Submodel(id_="https://example.org/Example_Submodel"), + model.Submodel(id_="https://example.org/Second_Submodel"), + ] + ) + + self.assertEqual("application/xml", response.content_type) + self.assertEqual(200, response.status_code) + parsed_body = etree.fromstring(response.get_data()) + + ids = [elem.text for elem in parsed_body.findall("aas:submodel/aas:id", namespaces=XML_NS_MAP)] + self.assertEqual(2, len(ids)) + self.assertEqual(len(ids), len(set(ids))) + + def test_example_empty_list(self): + response = base.XmlResponse([]) + + self.assertEqual("application/xml", response.content_type) + self.assertEqual(200, response.status_code) + parsed_body = etree.fromstring(response.get_data()) + + self.assertEqual(0, len(list(parsed_body.iterchildren()))) + + def test_paging_metadata(self): + response = base.XmlResponse( + obj=[model.Submodel(id_="https://example.org/Example_Submodel")], + paging_metadata=base.PagingMetadata(cursor="asdf"), + ) + + self.assertEqual("application/xml", response.content_type) + self.assertEqual(200, response.status_code) + parsed_body = etree.fromstring(response.get_data()) + + self.assertEqual("asdf", parsed_body.get("cursor")) + ids = [elem.text for elem in parsed_body.findall("aas:submodel/aas:id", namespaces=XML_NS_MAP)] + self.assertEqual(["https://example.org/Example_Submodel"], ids) + + def test_paging_metadata_no_cursor(self): + response = base.XmlResponse( + obj=[model.Submodel(id_="https://example.org/Example_Submodel")], + paging_metadata=None, + ) + + self.assertEqual("application/xml", response.content_type) + self.assertEqual(200, response.status_code) + parsed_body = etree.fromstring(response.get_data()) + + self.assertIsNone(parsed_body.get("cursor")) + + def test_result(self): + response = base.XmlResponse( + base.Result(False, [base.Message("BAD_CODE", "test", base.MessageType.ERROR, datetime.datetime.now())]) + ) + + self.assertEqual("application/xml", response.content_type) + self.assertEqual(200, response.status_code) + parsed_body = etree.fromstring(response.get_data()) + + self.assertEqual("response", parsed_body.tag) + self.assertEqual("false", parsed_body.findtext("success")) + + messages = parsed_body.findall("messages/message") + self.assertEqual(1, len(messages)) + self.assertEqual("BAD_CODE", messages[0].findtext("code")) + self.assertEqual("test", messages[0].findtext("text")) + self.assertEqual("Error", messages[0].findtext("messageType")) + isodatetime = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}$") + self.assertIsNotNone(isodatetime.match(messages[0].findtext("timestamp") or "")) diff --git a/server/test/interfaces/test_discovery.py b/server/test/interfaces/test_discovery.py new file mode 100644 index 000000000..05c235095 --- /dev/null +++ b/server/test/interfaces/test_discovery.py @@ -0,0 +1,298 @@ +""" +Endpoint tests for :class:`~app.interfaces.discovery.DiscoveryAPI`. + +The routes follow the *Discovery Service Specification* (SSP-001, "full" profile) from ``aas-specs-api``. +Where this server deviates from the spec's documented status codes, the test asserts the *implemented* +behavior and the deviation is called out in a comment. + +Requests and responses go through the shared :class:`~..format_utils.JsonFormatClient` (as in +``test/interfaces/repository/test_shells.py``). Only ``GET /lookup/shells/{aasIdentifier}`` returns +AAS model objects (``SpecificAssetId``), so it is additionally run against the XML +:class:`~..format_utils.FormatClient`; every other route returns plain strings / JSON objects that the +SDK XML serializer cannot render, so those are JSON only. +""" + +import json +import unittest +from typing import List, Tuple + +from app.interfaces.discovery import DiscoveryAPI, DiscoveryStore +from app.util.converters import base64url_encode +from basyx.aas.model import SpecificAssetId +from werkzeug.test import Client, TestResponse + +from .format_utils import FormatClient, JsonFormatClient, inject_format_clients, with_json_client, with_xml_client + + +def _b64url_json(payload: object) -> str: + return base64url_encode(json.dumps(payload)) + + +class DiscoveryEndpointTestBase(unittest.TestCase): + __test__ = False + + AAS_ID = "https://example.org/aas/1" + AAS_ID_2 = "https://example.org/aas/2" + UNKNOWN_ID = "https://example.org/unknown" + + store: DiscoveryStore + discovery_server: DiscoveryAPI + client: Client + format_client: JsonFormatClient + + @classmethod + def setUpClass(cls) -> None: + super().setUpClass() + cls.store = DiscoveryStore() + cls.discovery_server = DiscoveryAPI(cls.store, base_path="") + cls.client = Client(cls.discovery_server) + cls.format_client = JsonFormatClient(cls.client) + + def setUp(self) -> None: + self.store.aas_id_to_asset_ids.clear() + self.store.asset_id_to_aas_ids.clear() + + # ------------------------------------------------------------------ helpers + + def register(self, aas_id: str, asset_ids: List[Tuple[str, str]]) -> None: + assets: list[SpecificAssetId] = [SpecificAssetId(name, value) for name, value in asset_ids] + self.store.add_specific_asset_ids_to_aas(aas_id, assets) + for asset in assets: + self.store._add_aas_id_to_specific_asset_id(asset, aas_id) + + def assert_ok(self, response: TestResponse) -> None: + self.assertEqual(200, response.status_code, msg=response.get_data(as_text=True)) + + def assert_error(self, response: TestResponse, status_code: int) -> None: + self.assertEqual(status_code, response.status_code, msg=response.get_data(as_text=True)) + self.assertIn("success", response.get_data(as_text=True), msg=response.get_data(as_text=True)) + + +# ====================================================================== /description + + +class DiscoveryServiceDescriptionTest(DiscoveryEndpointTestBase): + __test__ = True + + def test_description_ok(self) -> None: + response = self.format_client.get("/description") + + self.assert_ok(response) + profiles = self.format_client.parse_object(response)["profiles"] + self.assertIn("https://admin-shell.io/aas/API/3/1/DiscoveryServiceSpecification/SSP-001", profiles) + self.assertIn("https://admin-shell.io/aas/API/3/1/DiscoveryServiceSpecification/SSP-002", profiles) + + +# ====================================================================== POST /lookup/shellsByAssetLink + + +class SearchShellsByAssetLinkEndpointTest(DiscoveryEndpointTestBase): + __test__ = True + + def test_search_match(self) -> None: + self.register(self.AAS_ID, [("serial", "123")]) + + response = self.format_client.post("/lookup/shellsByAssetLink", obj=[{"name": "serial", "value": "123"}]) + + self.assert_ok(response) + self.assertEqual([self.AAS_ID], self.format_client.parse_collection(response)) + + def test_search_no_match_returns_empty(self) -> None: + self.register(self.AAS_ID, [("serial", "123")]) + + response = self.format_client.post( + "/lookup/shellsByAssetLink", obj=[{"name": "serial", "value": "does-not-exist"}] + ) + + self.assert_ok(response) + self.assertEqual([], self.format_client.parse_collection(response)) + + def test_search_empty_body_returns_empty(self) -> None: + self.register(self.AAS_ID, [("serial", "123")]) + + response = self.format_client.post("/lookup/shellsByAssetLink", obj=[]) + + self.assert_ok(response) + self.assertEqual([], self.format_client.parse_collection(response)) + + def test_search_multiple_links_unions_results(self) -> None: + self.register(self.AAS_ID, [("serial", "1")]) + self.register(self.AAS_ID_2, [("serial", "2")]) + + response = self.format_client.post( + "/lookup/shellsByAssetLink", + obj=[{"name": "serial", "value": "1"}, {"name": "serial", "value": "2"}], + ) + + self.assert_ok(response) + self.assertEqual({self.AAS_ID, self.AAS_ID_2}, set(self.format_client.parse_collection(response))) + + def test_search_supports_pagination(self) -> None: + self.register(self.AAS_ID, [("serial", "1")]) + self.register(self.AAS_ID_2, [("serial", "1")]) + + response = self.format_client.post("/lookup/shellsByAssetLink?limit=1", obj=[{"name": "serial", "value": "1"}]) + + self.assert_ok(response) + self.assertEqual(1, len(self.format_client.parse_collection(response))) + self.assertIsNotNone(self.format_client.next_cursor(response)) + + def test_search_malformed_json_returns_400(self) -> None: + response = self.format_client.post( + "/lookup/shellsByAssetLink", data=b"{not json", content_type="application/json" + ) + + self.assert_error(response, 400) + + def test_search_asset_link_missing_value_returns_400(self) -> None: + self.assert_error(self.format_client.post("/lookup/shellsByAssetLink", obj=[{"name": "serial"}]), 400) + + +# ====================================================================== GET /lookup/shells (deprecated) + + +class GetShellsByAssetLinkQueryEndpointTest(DiscoveryEndpointTestBase): + """The deprecated ``GET /lookup/shells?assetIds=...`` route (kept for BaSyx UI interoperability).""" + + __test__ = True + + def test_query_match(self) -> None: + self.register(self.AAS_ID, [("serial", "123")]) + + response = self.format_client.get(f"/lookup/shells?assetIds={_b64url_json({'name': 'serial', 'value': '123'})}") + + self.assert_ok(response) + self.assertEqual([self.AAS_ID], self.format_client.parse_collection(response)) + + def test_query_accepts_list_payload(self) -> None: + self.register(self.AAS_ID, [("serial", "123")]) + + response = self.format_client.get( + f"/lookup/shells?assetIds={_b64url_json([{'name': 'serial', 'value': '123'}])}" + ) + + self.assert_ok(response) + self.assertEqual([self.AAS_ID], self.format_client.parse_collection(response)) + + def test_query_missing_parameter_returns_400(self) -> None: + self.assert_error(self.format_client.get("/lookup/shells"), 400) + + def test_query_invalid_base64_returns_400(self) -> None: + self.assert_error(self.format_client.get("/lookup/shells?assetIds=not-base64!!!"), 400) + + def test_query_decoded_value_not_json_returns_400(self) -> None: + self.assert_error(self.format_client.get(f"/lookup/shells?assetIds={base64url_encode('not json')}"), 400) + + def test_query_payload_not_object_or_list_returns_400(self) -> None: + self.assert_error(self.format_client.get(f"/lookup/shells?assetIds={_b64url_json(123)}"), 400) + + def test_query_payload_item_not_object_returns_400(self) -> None: + self.assert_error(self.format_client.get(f"/lookup/shells?assetIds={_b64url_json([1, 2])}"), 400) + + +# ====================================================================== /lookup/shells/{aasIdentifier} + + +@inject_format_clients +class AssetLinksByIdEndpointTest(DiscoveryEndpointTestBase): + """Tests for GET/POST/DELETE on ``/lookup/shells/{aasIdentifier}``.""" + + __test__ = True + + # ------------------------------------------------------------------ GET (returns SpecificAssetId objects) + + @with_json_client + @with_xml_client + def test_get_returns_stored_asset_ids(self, format_client: FormatClient) -> None: + self.register(self.AAS_ID, [("serial", "123"), ("globalAssetId", "https://example.org/asset/1")]) + + response = format_client.get(f"/lookup/shells/{base64url_encode(self.AAS_ID)}") + + self.assertEqual(200, response.status_code, msg=response.get_data(as_text=True)) + pairs = { + (format_client.field(node, "name"), format_client.field(node, "value")) + for node in format_client.parse_collection(response) + } + self.assertEqual({("serial", "123"), ("globalAssetId", "https://example.org/asset/1")}, pairs) + + @with_json_client + @with_xml_client + def test_get_unknown_aas_returns_empty(self, format_client: FormatClient) -> None: + # Spec allows 404 here; this server returns 200 with an empty collection instead. + response = format_client.get(f"/lookup/shells/{base64url_encode(self.UNKNOWN_ID)}") + + self.assertEqual(200, response.status_code, msg=response.get_data(as_text=True)) + self.assertEqual([], format_client.parse_collection(response)) + + # ------------------------------------------------------------------ POST + + def test_post_creates_asset_links(self) -> None: + response = self.format_client.post( + f"/lookup/shells/{base64url_encode(self.AAS_ID)}", + obj=[{"name": "serial", "value": "123"}], + ) + + # Spec documents 201 for creation; this server responds 200 with the updated mapping. + self.assert_ok(response) + body = self.format_client.parse_object(response) + self.assertEqual([("serial", "123")], [(a["name"], a["value"]) for a in body[self.AAS_ID]]) + self.assertEqual( + [self.AAS_ID], + self.format_client.parse_collection( + self.format_client.post("/lookup/shellsByAssetLink", obj=[{"name": "serial", "value": "123"}]) + ), + ) + + def test_post_accepts_single_object_body(self) -> None: + response = self.format_client.post( + f"/lookup/shells/{base64url_encode(self.AAS_ID)}", obj={"name": "serial", "value": "123"} + ) + + self.assert_ok(response) + self.assertEqual( + [("serial", "123")], + [(a["name"], a["value"]) for a in self.format_client.parse_object(response)[self.AAS_ID]], + ) + + def test_post_is_additive(self) -> None: + self.register(self.AAS_ID, [("serial", "123")]) + + response = self.format_client.post( + f"/lookup/shells/{base64url_encode(self.AAS_ID)}", obj=[{"name": "batch", "value": "xyz"}] + ) + + self.assert_ok(response) + stored = {(a["name"], a["value"]) for a in self.format_client.parse_object(response)[self.AAS_ID]} + self.assertEqual({("serial", "123"), ("batch", "xyz")}, stored) + + def test_post_malformed_json_returns_400(self) -> None: + response = self.format_client.post( + f"/lookup/shells/{base64url_encode(self.AAS_ID)}", data=b"{not json", content_type="application/json" + ) + + self.assert_error(response, 400) + + # ------------------------------------------------------------------ DELETE + + def test_delete_removes_asset_links(self) -> None: + self.register(self.AAS_ID, [("serial", "123")]) + + response = self.format_client.delete(f"/lookup/shells/{base64url_encode(self.AAS_ID)}") + + self.assertEqual(204, response.status_code, msg=response.get_data(as_text=True)) + self.assertEqual( + [], + self.format_client.parse_object(self.format_client.get(f"/lookup/shells/{base64url_encode(self.AAS_ID)}")), + ) + self.assertEqual( + [], + self.format_client.parse_collection( + self.format_client.post("/lookup/shellsByAssetLink", obj=[{"name": "serial", "value": "123"}]) + ), + ) + + def test_delete_unknown_aas_returns_204(self) -> None: + # Spec allows 404 here; this server always returns 204. + response = self.format_client.delete(f"/lookup/shells/{base64url_encode(self.UNKNOWN_ID)}") + + self.assertEqual(204, response.status_code, msg=response.get_data(as_text=True)) diff --git a/server/test/interfaces/test_registry.py b/server/test/interfaces/test_registry.py new file mode 100644 index 000000000..c6d25726d --- /dev/null +++ b/server/test/interfaces/test_registry.py @@ -0,0 +1,540 @@ +""" +Endpoint tests for :class:`~app.interfaces.registry.RegistryAPI`. + +The routes and status codes follow the *Asset Administration Shell Registry* and *Submodel Registry* +Service Specifications (SSP-001, "full" profile) from ``aas-specs-api``. Only JSON is exercised: the +registry only stores :class:`~app.model.descriptor.Descriptor` objects, which the SDK XML serializer +cannot handle, and the spec only defines ``application/json`` for these routes. + +Requests and responses go through the shared :class:`~..format_utils.JsonFormatClient` (as in +``test/interfaces/repository/test_shells.py``), which serializes request bodies with the server's +``ServerAASToJsonEncoder`` so :class:`~app.model.descriptor.Descriptor` objects can be sent. +""" + +import unittest +from typing import Any + +from app.interfaces.registry import RegistryAPI +from app.model import DictDescriptorStore +from app.model.descriptor import AssetAdministrationShellDescriptor, SubmodelDescriptor +from app.model.endpoint import Endpoint, ProtocolInformation +from app.util.converters import base64url_encode +from basyx.aas import model +from werkzeug.test import Client, TestResponse + +from .format_utils import JsonFormatClient + + +class _InMemoryDescriptorStore(DictDescriptorStore): + """ + In-memory descriptor store with a no-op ``commit`` and a ``clear``. + + :class:`~app.model.provider.DictDescriptorStore` inherits ``commit`` from the SDK's + ``AbstractObjectStore``, where it raises ``NotImplementedError``; the registry calls it after every + write. This mirrors what ``SetIdentifiableStore`` provides for the repository tests. + """ + + def commit(self, x: Any) -> None: + pass + + def clear(self) -> None: + self._backend.clear() + + +def _endpoint(interface: str = "AAS-3.0", href: str = "https://example.org/endpoint") -> Endpoint: + return Endpoint(interface=interface, protocol_information=ProtocolInformation(href=href)) + + +def _aas_descriptor(id_: str, **kwargs: Any) -> AssetAdministrationShellDescriptor: + kwargs.setdefault("endpoints", [_endpoint("AAS-3.0")]) + return AssetAdministrationShellDescriptor(id_=id_, **kwargs) + + +def _submodel_descriptor(id_: str, **kwargs: Any) -> SubmodelDescriptor: + kwargs.setdefault("endpoints", [_endpoint("SUBMODEL-3.0")]) + return SubmodelDescriptor(id_=id_, **kwargs) + + +class RegistryEndpointTestBase(unittest.TestCase): + __test__ = False + + AAS_ID = "https://example.org/shell-descriptors/1" + AAS_ID_2 = "https://example.org/shell-descriptors/2" + SM_ID = "https://example.org/submodel-descriptors/1" + SM_ID_2 = "https://example.org/submodel-descriptors/2" + UNKNOWN_ID = "https://example.org/unknown" + + store: _InMemoryDescriptorStore + registry_server: RegistryAPI + client: Client + format_client: JsonFormatClient + + @classmethod + def setUpClass(cls) -> None: + super().setUpClass() + cls.store = _InMemoryDescriptorStore() + cls.registry_server = RegistryAPI(cls.store, base_path="") + cls.client = Client(cls.registry_server) + cls.format_client = JsonFormatClient(cls.client) + + def setUp(self) -> None: + self.store.clear() + + # ------------------------------------------------------------------ assertion helpers + + def assert_ok(self, response: TestResponse) -> None: + self.assertEqual(200, response.status_code, msg=response.get_data(as_text=True)) + + def assert_error(self, response: TestResponse, status_code: int) -> None: + self.assertEqual(status_code, response.status_code, msg=response.get_data(as_text=True)) + self.assertIn("success", response.get_data(as_text=True), msg=response.get_data(as_text=True)) + + def ids(self, response: TestResponse) -> list: + return [self.format_client.identifier(node) for node in self.format_client.parse_collection(response)] + + +# ====================================================================== /description + + +class RegistryServiceDescriptionTest(RegistryEndpointTestBase): + __test__ = True + + def test_description_ok(self) -> None: + response = self.format_client.get("/description") + + self.assert_ok(response) + profiles = self.format_client.parse_object(response)["profiles"] + self.assertIn( + "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellRegistryServiceSpecification/SSP-001", profiles + ) + self.assertIn("https://admin-shell.io/aas/API/3/1/SubmodelRegistryServiceSpecification/SSP-001", profiles) + self.assertIn( + "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellRegistryServiceSpecification/SSP-002", profiles + ) + self.assertIn("https://admin-shell.io/aas/API/3/1/SubmodelRegistryServiceSpecification/SSP-002", profiles) + + +# ====================================================================== /shell-descriptors + + +class ShellDescriptorsEndpointTest(RegistryEndpointTestBase): + """Tests for the ``/shell-descriptors`` and ``/shell-descriptors/{aasIdentifier}`` routes.""" + + __test__ = True + + # ------------------------------------------------------------------ GET /shell-descriptors + + def test_get_all_empty(self) -> None: + response = self.format_client.get("/shell-descriptors") + + self.assert_ok(response) + self.assertEqual([], self.format_client.parse_collection(response)) + + def test_get_all_returns_registered_descriptors(self) -> None: + self.store.add(_aas_descriptor(self.AAS_ID)) + self.store.add(_aas_descriptor(self.AAS_ID_2)) + + response = self.format_client.get("/shell-descriptors") + + self.assert_ok(response) + self.assertEqual({self.AAS_ID, self.AAS_ID_2}, set(self.ids(response))) + + def test_get_all_only_returns_aas_descriptors(self) -> None: + # The store is shared between the AAS- and Submodel-registry routes, so this route must filter by type. + self.store.add(_aas_descriptor(self.AAS_ID)) + self.store.add(_submodel_descriptor(self.SM_ID)) + + response = self.format_client.get("/shell-descriptors") + + self.assert_ok(response) + self.assertEqual([self.AAS_ID], self.ids(response)) + + def test_get_all_supports_pagination(self) -> None: + self.store.add(_aas_descriptor(self.AAS_ID)) + self.store.add(_aas_descriptor(self.AAS_ID_2)) + + pages = self.format_client.get_paginated("/shell-descriptors", limit=1, max_pages=2) + + self.assertEqual([1, 1], [len(page) for page in pages]) + seen = [self.format_client.identifier(node) for page in pages for node in page] + self.assertEqual({self.AAS_ID, self.AAS_ID_2}, set(seen)) + self.assertEqual(len(seen), len(set(seen)), "an item was returned on more than one page") + + def test_get_all_negative_limit_returns_400(self) -> None: + self.store.add(_aas_descriptor(self.AAS_ID)) + + self.assert_error(self.format_client.get("/shell-descriptors?limit=-1"), 400) + + def test_get_all_filter_by_asset_kind(self) -> None: + self.store.add(_aas_descriptor(self.AAS_ID, asset_kind=model.AssetKind.INSTANCE)) + self.store.add(_aas_descriptor(self.AAS_ID_2, asset_kind=model.AssetKind.TYPE)) + + response = self.format_client.get("/shell-descriptors?assetKind=INSTANCE") + + self.assert_ok(response) + self.assertEqual([self.AAS_ID], self.ids(response)) + + def test_get_all_filter_by_asset_kind_invalid_returns_400(self) -> None: + self.store.add(_aas_descriptor(self.AAS_ID, asset_kind=model.AssetKind.INSTANCE)) + + # The enum member names are upper-case; the serialized ("Instance") spelling is rejected. + self.assert_error(self.format_client.get("/shell-descriptors?assetKind=Instance"), 400) + + def test_get_all_filter_by_asset_type(self) -> None: + self.store.add(_aas_descriptor(self.AAS_ID, asset_type="https://example.org/type/a")) + self.store.add(_aas_descriptor(self.AAS_ID_2, asset_type="https://example.org/type/b")) + + response = self.format_client.get( + f"/shell-descriptors?assetType={base64url_encode('https://example.org/type/b')}" + ) + + self.assert_ok(response) + self.assertEqual([self.AAS_ID_2], self.ids(response)) + + def test_get_all_filter_by_asset_type_no_match(self) -> None: + self.store.add(_aas_descriptor(self.AAS_ID, asset_type="https://example.org/type/a")) + + response = self.format_client.get( + f"/shell-descriptors?assetType={base64url_encode('https://example.org/type/none')}" + ) + + self.assert_ok(response) + self.assertEqual([], self.format_client.parse_collection(response)) + + # ------------------------------------------------------------------ POST /shell-descriptors + + def test_post_success(self) -> None: + descriptor = _aas_descriptor(self.AAS_ID) + + response = self.format_client.post("/shell-descriptors", obj=descriptor) + + self.assertEqual(201, response.status_code, msg=response.get_data(as_text=True)) + self.assertIn(base64url_encode(self.AAS_ID), response.headers["Location"]) + self.assertEqual(self.AAS_ID, self.format_client.identifier(self.format_client.parse_object(response))) + self.assertIsNotNone(self.store.get(self.AAS_ID)) + + def test_post_missing_id_returns_400(self) -> None: + response = self.format_client.post( + "/shell-descriptors", data=b'{"endpoints": []}', content_type="application/json" + ) + + self.assert_error(response, 400) + + def test_post_malformed_json_returns_400(self) -> None: + response = self.format_client.post("/shell-descriptors", data=b"not json", content_type="application/json") + + self.assert_error(response, 400) + + def test_post_unsupported_content_type_returns_415(self) -> None: + response = self.format_client.post( + "/shell-descriptors", + data=self.format_client.serialize(_aas_descriptor(self.AAS_ID)), + content_type="text/plain", + ) + + self.assert_error(response, 415) + + def test_post_conflict_returns_409(self) -> None: + self.store.add(_aas_descriptor(self.AAS_ID)) + + self.assert_error(self.format_client.post("/shell-descriptors", obj=_aas_descriptor(self.AAS_ID)), 409) + + # ------------------------------------------------------------------ GET /shell-descriptors/{aasIdentifier} + + def test_get_by_id_success(self) -> None: + self.store.add(_aas_descriptor(self.AAS_ID)) + + response = self.format_client.get(f"/shell-descriptors/{base64url_encode(self.AAS_ID)}") + + self.assert_ok(response) + self.assertEqual(self.AAS_ID, self.format_client.identifier(self.format_client.parse_object(response))) + + def test_get_by_id_not_found_returns_404(self) -> None: + self.assert_error(self.format_client.get(f"/shell-descriptors/{base64url_encode(self.UNKNOWN_ID)}"), 404) + + # ------------------------------------------------------------------ PUT /shell-descriptors/{aasIdentifier} + + def test_put_creates_when_absent_returns_201(self) -> None: + response = self.format_client.put( + f"/shell-descriptors/{base64url_encode(self.AAS_ID)}", obj=_aas_descriptor(self.AAS_ID) + ) + + self.assertEqual(201, response.status_code, msg=response.get_data(as_text=True)) + self.assertIn(base64url_encode(self.AAS_ID), response.headers["Location"]) + self.assertIsNotNone(self.store.get(self.AAS_ID)) + + def test_put_updates_when_present_returns_204(self) -> None: + self.store.add(_aas_descriptor(self.AAS_ID, id_short="Original")) + updated = _aas_descriptor(self.AAS_ID, id_short="Updated") + + response = self.format_client.put(f"/shell-descriptors/{base64url_encode(self.AAS_ID)}", obj=updated) + + self.assertEqual(204, response.status_code, msg=response.get_data(as_text=True)) + stored = self.store.get(self.AAS_ID) + assert isinstance(stored, AssetAdministrationShellDescriptor) # make mypy happy + self.assertEqual("Updated", stored.id_short) + + # ------------------------------------------------------------------ DELETE /shell-descriptors/{aasIdentifier} + + def test_delete_success_returns_204(self) -> None: + self.store.add(_aas_descriptor(self.AAS_ID)) + + response = self.format_client.delete(f"/shell-descriptors/{base64url_encode(self.AAS_ID)}") + + self.assertEqual(204, response.status_code, msg=response.get_data(as_text=True)) + self.assertIsNone(self.store.get(self.AAS_ID)) + + def test_delete_not_found_returns_404(self) -> None: + self.assert_error(self.format_client.delete(f"/shell-descriptors/{base64url_encode(self.UNKNOWN_ID)}"), 404) + + +# ====================================================================== /shell-descriptors/{aasId}/submodel-descriptors + + +class SubmodelDescriptorsThroughSuperpathEndpointTest(RegistryEndpointTestBase): + """Tests for the ``/shell-descriptors/{aasIdentifier}/submodel-descriptors`` routes.""" + + __test__ = True + + def setUp(self) -> None: + super().setUp() + self.store.add(_aas_descriptor(self.AAS_ID)) + + @property + def _base(self) -> str: + return f"/shell-descriptors/{base64url_encode(self.AAS_ID)}/submodel-descriptors" + + @property + def _missing_base(self) -> str: + return f"/shell-descriptors/{base64url_encode(self.UNKNOWN_ID)}/submodel-descriptors" + + # ------------------------------------------------------------------ GET (collection) + + def test_get_all_empty(self) -> None: + response = self.format_client.get(self._base) + + self.assert_ok(response) + self.assertEqual([], self.format_client.parse_collection(response)) + + def test_get_all_returns_nested_descriptors(self) -> None: + self.format_client.post(self._base, obj=_submodel_descriptor(self.SM_ID)) + self.format_client.post(self._base, obj=_submodel_descriptor(self.SM_ID_2)) + + response = self.format_client.get(self._base) + + self.assert_ok(response) + self.assertEqual({self.SM_ID, self.SM_ID_2}, set(self.ids(response))) + + def test_get_all_supports_pagination(self) -> None: + self.format_client.post(self._base, obj=_submodel_descriptor(self.SM_ID)) + self.format_client.post(self._base, obj=_submodel_descriptor(self.SM_ID_2)) + + pages = self.format_client.get_paginated(self._base, limit=1, max_pages=2) + + self.assertEqual([1, 1], [len(page) for page in pages]) + + def test_get_all_aas_not_found_returns_404(self) -> None: + self.assert_error(self.format_client.get(self._missing_base), 404) + + # ------------------------------------------------------------------ POST + + def test_post_success(self) -> None: + response = self.format_client.post(self._base, obj=_submodel_descriptor(self.SM_ID)) + + self.assertEqual(201, response.status_code, msg=response.get_data(as_text=True)) + self.assertIn(base64url_encode(self.SM_ID), response.headers["Location"]) + self.assertEqual([self.SM_ID], self.ids(self.format_client.get(self._base))) + + def test_post_conflict_returns_409(self) -> None: + self.format_client.post(self._base, obj=_submodel_descriptor(self.SM_ID)) + + self.assert_error(self.format_client.post(self._base, obj=_submodel_descriptor(self.SM_ID)), 409) + + def test_post_aas_not_found_returns_404(self) -> None: + self.assert_error(self.format_client.post(self._missing_base, obj=_submodel_descriptor(self.SM_ID)), 404) + + # ------------------------------------------------------------------ GET (single) + + def test_get_by_id_success(self) -> None: + self.format_client.post(self._base, obj=_submodel_descriptor(self.SM_ID)) + + response = self.format_client.get(f"{self._base}/{base64url_encode(self.SM_ID)}") + + self.assert_ok(response) + self.assertEqual(self.SM_ID, self.format_client.identifier(self.format_client.parse_object(response))) + + def test_get_by_id_not_found_returns_404(self) -> None: + self.assert_error(self.format_client.get(f"{self._base}/{base64url_encode(self.UNKNOWN_ID)}"), 404) + + def test_get_by_id_aas_not_found_returns_404(self) -> None: + self.assert_error(self.format_client.get(f"{self._missing_base}/{base64url_encode(self.SM_ID)}"), 404) + + # ------------------------------------------------------------------ PUT + + def test_put_updates_when_present_returns_204(self) -> None: + self.format_client.post(self._base, obj=_submodel_descriptor(self.SM_ID, id_short="Original")) + updated = _submodel_descriptor(self.SM_ID, id_short="Updated") + + response = self.format_client.put(f"{self._base}/{base64url_encode(self.SM_ID)}", obj=updated) + + self.assertEqual(204, response.status_code, msg=response.get_data(as_text=True)) + fetched = self.format_client.parse_object( + self.format_client.get(f"{self._base}/{base64url_encode(self.SM_ID)}") + ) + self.assertEqual("Updated", self.format_client.field(fetched, "idShort")) + + def test_put_creates_when_absent_returns_201(self) -> None: + response = self.format_client.put( + f"{self._base}/{base64url_encode(self.SM_ID)}", obj=_submodel_descriptor(self.SM_ID) + ) + + self.assertEqual(201, response.status_code, msg=response.get_data(as_text=True)) + self.assertIn(base64url_encode(self.SM_ID), response.headers["Location"]) + self.assertEqual([self.SM_ID], self.ids(self.format_client.get(self._base))) + + def test_put_aas_not_found_returns_404(self) -> None: + self.assert_error( + self.format_client.put( + f"{self._missing_base}/{base64url_encode(self.SM_ID)}", obj=_submodel_descriptor(self.SM_ID) + ), + 404, + ) + + # ------------------------------------------------------------------ DELETE + + def test_delete_success_returns_204(self) -> None: + self.format_client.post(self._base, obj=_submodel_descriptor(self.SM_ID)) + + response = self.format_client.delete(f"{self._base}/{base64url_encode(self.SM_ID)}") + + self.assertEqual(204, response.status_code, msg=response.get_data(as_text=True)) + self.assertEqual([], self.format_client.parse_collection(self.format_client.get(self._base))) + + def test_delete_not_found_returns_404(self) -> None: + self.assert_error(self.format_client.delete(f"{self._base}/{base64url_encode(self.UNKNOWN_ID)}"), 404) + + def test_delete_aas_not_found_returns_404(self) -> None: + self.assert_error(self.format_client.delete(f"{self._missing_base}/{base64url_encode(self.SM_ID)}"), 404) + + +# ====================================================================== /submodel-descriptors + + +class SubmodelDescriptorsEndpointTest(RegistryEndpointTestBase): + """Tests for the standalone ``/submodel-descriptors`` and ``/submodel-descriptors/{submodelIdentifier}`` routes.""" + + __test__ = True + + # ------------------------------------------------------------------ GET /submodel-descriptors + + def test_get_all_empty(self) -> None: + response = self.format_client.get("/submodel-descriptors") + + self.assert_ok(response) + self.assertEqual([], self.format_client.parse_collection(response)) + + def test_get_all_returns_registered_descriptors(self) -> None: + self.store.add(_submodel_descriptor(self.SM_ID)) + self.store.add(_submodel_descriptor(self.SM_ID_2)) + + response = self.format_client.get("/submodel-descriptors") + + self.assert_ok(response) + self.assertEqual({self.SM_ID, self.SM_ID_2}, set(self.ids(response))) + + def test_get_all_only_returns_submodel_descriptors(self) -> None: + self.store.add(_submodel_descriptor(self.SM_ID)) + self.store.add(_aas_descriptor(self.AAS_ID)) + + response = self.format_client.get("/submodel-descriptors") + + self.assert_ok(response) + self.assertEqual([self.SM_ID], self.ids(response)) + + def test_get_all_supports_pagination(self) -> None: + self.store.add(_submodel_descriptor(self.SM_ID)) + self.store.add(_submodel_descriptor(self.SM_ID_2)) + + pages = self.format_client.get_paginated("/submodel-descriptors", limit=1, max_pages=2) + + self.assertEqual([1, 1], [len(page) for page in pages]) + + def test_get_all_negative_limit_returns_400(self) -> None: + self.store.add(_submodel_descriptor(self.SM_ID)) + + self.assert_error(self.format_client.get("/submodel-descriptors?limit=-1"), 400) + + # ------------------------------------------------------------------ POST /submodel-descriptors + + def test_post_success(self) -> None: + response = self.format_client.post("/submodel-descriptors", obj=_submodel_descriptor(self.SM_ID)) + + self.assertEqual(201, response.status_code, msg=response.get_data(as_text=True)) + self.assertIn(base64url_encode(self.SM_ID), response.headers["Location"]) + self.assertIsNotNone(self.store.get(self.SM_ID)) + + def test_post_missing_id_returns_400(self) -> None: + response = self.format_client.post( + "/submodel-descriptors", data=b'{"endpoints": []}', content_type="application/json" + ) + + self.assert_error(response, 400) + + def test_post_conflict_returns_409(self) -> None: + self.store.add(_submodel_descriptor(self.SM_ID)) + + self.assert_error(self.format_client.post("/submodel-descriptors", obj=_submodel_descriptor(self.SM_ID)), 409) + + # ------------------------------------------------------------------ GET /submodel-descriptors/{submodelIdentifier} + + def test_get_by_id_success(self) -> None: + self.store.add(_submodel_descriptor(self.SM_ID)) + + response = self.format_client.get(f"/submodel-descriptors/{base64url_encode(self.SM_ID)}") + + self.assert_ok(response) + self.assertEqual(self.SM_ID, self.format_client.identifier(self.format_client.parse_object(response))) + + def test_get_by_id_not_found_returns_404(self) -> None: + self.assert_error(self.format_client.get(f"/submodel-descriptors/{base64url_encode(self.UNKNOWN_ID)}"), 404) + + def test_get_by_id_wrong_type_returns_404(self) -> None: + # A descriptor with this id exists, but it is an AAS descriptor, not a submodel descriptor. + self.store.add(_aas_descriptor(self.AAS_ID)) + + self.assert_error(self.format_client.get(f"/submodel-descriptors/{base64url_encode(self.AAS_ID)}"), 404) + + # ------------------------------------------------------------------ PUT /submodel-descriptors/{submodelIdentifier} + + def test_put_creates_when_absent_returns_201(self) -> None: + response = self.format_client.put( + f"/submodel-descriptors/{base64url_encode(self.SM_ID)}", obj=_submodel_descriptor(self.SM_ID) + ) + + self.assertEqual(201, response.status_code, msg=response.get_data(as_text=True)) + self.assertIn(base64url_encode(self.SM_ID), response.headers["Location"]) + self.assertIsNotNone(self.store.get(self.SM_ID)) + + def test_put_updates_when_present_returns_204(self) -> None: + self.store.add(_submodel_descriptor(self.SM_ID, id_short="Original")) + updated = _submodel_descriptor(self.SM_ID, id_short="Updated") + + response = self.format_client.put(f"/submodel-descriptors/{base64url_encode(self.SM_ID)}", obj=updated) + + self.assertEqual(204, response.status_code, msg=response.get_data(as_text=True)) + stored = self.store.get(self.SM_ID) + assert isinstance(stored, SubmodelDescriptor) # make mypy happy + self.assertEqual("Updated", stored.id_short) + + # --------------------------------------------------------------- DELETE /submodel-descriptors/{submodelIdentifier} + + def test_delete_success_returns_204(self) -> None: + self.store.add(_submodel_descriptor(self.SM_ID)) + + response = self.format_client.delete(f"/submodel-descriptors/{base64url_encode(self.SM_ID)}") + + self.assertEqual(204, response.status_code, msg=response.get_data(as_text=True)) + self.assertIsNone(self.store.get(self.SM_ID)) + + def test_delete_not_found_returns_404(self) -> None: + self.assert_error(self.format_client.delete(f"/submodel-descriptors/{base64url_encode(self.UNKNOWN_ID)}"), 404) diff --git a/server/test/interfaces/test_repository.py b/server/test/interfaces/test_repository.py deleted file mode 100644 index d60926132..000000000 --- a/server/test/interfaces/test_repository.py +++ /dev/null @@ -1,133 +0,0 @@ -# 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 - -""" -This test uses the schemathesis package to perform automated stateful testing on the implemented http api. Requests -are created automatically based on the json schemata given in the api specification, responses are also validated -against said schemata. - -For data generation schemathesis uses hypothesis and hypothesis-jsonschema, hence the name. hypothesis is a library -for automated, property-based testing. It can generate test cases based on strategies. hypothesis-jsonschema is such -a strategy for generating data that matches a given JSON schema. - -schemathesis allows stateful testing by generating a statemachine based on the OAS links contained in the specification. -This is applied here with the APIWorkflowAAS and APIWorkflowSubmodel classes. They inherit the respective state machine -and offer an automatically generated python unittest TestCase. -""" - -# TODO: lookup schemathesis deps and add them to the readme -# TODO: implement official Plattform I4.0 HTTP API -# TODO: check required properties of schema -# TODO: add id_short format to schemata - -import os -import pathlib -import random -import urllib.parse -from typing import Set - -import hypothesis.strategies -import schemathesis -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 - - -def _encode_and_quote(identifier: model.Identifier) -> str: - return urllib.parse.quote(urllib.parse.quote(identifier, safe=""), safe="") - - -def _check_transformed(response, case): - """ - This helper function performs an additional checks on requests that have been *transformed*, i.e. requests, that - resulted from schemathesis using an OpenAPI Spec link. It asserts, that requests that are performed after a link has - been used, must be successful and result in a 2xx response. The exception are requests where hypothesis generates - invalid data (data, that validates against the schema, but is still semantically invalid). Such requests would - result in a 422 - Unprocessable Entity, which is why the 422 status code is ignored here. - """ - if case.source is not None: - assert 200 <= response.status_code < 300 or response.status_code == 422 - - -# define some settings for hypothesis, used in both api test cases -HYPOTHESIS_SETTINGS = hypothesis.settings( - max_examples=int(os.getenv("HYPOTHESIS_MAX_EXAMPLES", 10)), - stateful_step_count=5, - # disable the filter_too_much health check, which triggers if a strategy filters too much data, raising an error - suppress_health_check=[hypothesis.HealthCheck.filter_too_much], - # disable data generation deadlines, which would result in an error if data generation takes too much time - deadline=None, -) - -BASE_URL = "/api/v1" -IDENTIFIER_AAS: Set[str] = set() -IDENTIFIER_SUBMODEL: Set[str] = set() - -# register hypothesis strategy for generating valid idShorts -ID_SHORT_STRATEGY = hypothesis.strategies.from_regex(r"\A[A-Za-z_][0-9A-Za-z_]*\Z") -schemathesis.register_string_format("id_short", ID_SHORT_STRATEGY) - -# store identifiers of available AAS and Submodels -for obj in create_full_example(): - if isinstance(obj, model.AssetAdministrationShell): - IDENTIFIER_AAS.add(_encode_and_quote(obj.id)) - if isinstance(obj, model.Submodel): - IDENTIFIER_SUBMODEL.add(_encode_and_quote(obj.id)) - -# load aas and submodel api specs -AAS_SCHEMA = schemathesis.from_path( - pathlib.Path(__file__).parent / "http-api-oas-aas.yaml", - app=WSGIApp(create_full_example(), DictSupplementaryFileContainer()), -) - -SUBMODEL_SCHEMA = schemathesis.from_path( - pathlib.Path(__file__).parent / "http-api-oas-submodel.yaml", - app=WSGIApp(create_full_example(), DictSupplementaryFileContainer()), -) - - -class APIWorkflowAAS(AAS_SCHEMA.as_state_machine()): # type: ignore - def setup(self): - self.schema.app.identifiable_store = create_full_example() - # select random identifier for each test scenario - self.schema.base_url = BASE_URL + "/aas/" + random.choice(tuple(IDENTIFIER_AAS)) - - def transform(self, result, direction, case): - out = super().transform(result, direction, case) - print("transformed") - print(out) - print(result.response, direction.name) - return out - - def validate_response(self, response, case, additional_checks=()) -> None: - super().validate_response(response, case, additional_checks + (_check_transformed,)) - - -class APIWorkflowSubmodel(SUBMODEL_SCHEMA.as_state_machine()): # type: ignore - def setup(self): - self.schema.app.identifiable_store = create_full_example() - self.schema.base_url = BASE_URL + "/submodels/" + random.choice(tuple(IDENTIFIER_SUBMODEL)) - - def transform(self, result, direction, case): - out = super().transform(result, direction, case) - print("transformed") - print(out) - print(result.response, direction.name) - return out - - def validate_response(self, response, case, additional_checks=()) -> None: - super().validate_response(response, case, additional_checks + (_check_transformed,)) - - -# APIWorkflow.TestCase is a standard python unittest.TestCase -# TODO: Fix HTTP API Tests -# ApiTestAAS = APIWorkflowAAS.TestCase -# ApiTestAAS.settings = HYPOTHESIS_SETTINGS - -# ApiTestSubmodel = APIWorkflowSubmodel.TestCase -# ApiTestSubmodel.settings = HYPOTHESIS_SETTINGS diff --git a/server/test/interfaces/test_shells_asset_ids.py b/server/test/interfaces/test_shells_asset_ids.py deleted file mode 100644 index e45af1b90..000000000 --- a/server/test/interfaces/test_shells_asset_ids.py +++ /dev/null @@ -1,48 +0,0 @@ -# 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 unittest - -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 werkzeug.test import Client - -BASE_PATH = "/api/v3.1" - - -def _encode_asset_id(name: str, value: str) -> str: - payload = json.dumps({"name": name, "value": value}) - return base64.urlsafe_b64encode(payload.encode()).decode() - - -class ShellsAssetIdsTest(unittest.TestCase): - def setUp(self) -> None: - self.example_data = create_full_example() - app = WSGIApp(self.example_data, DictSupplementaryFileContainer()) - self.client = Client(app) - - def test_multiple_global_asset_ids_returns_matching_results(self) -> None: - aas_list = [obj for obj in self.example_data if isinstance(obj, model.AssetAdministrationShell)] - known_id = aas_list[0].asset_information.global_asset_id - assert known_id is not None - unknown_id = "http://example.org/nonexistent_asset" - id1 = _encode_asset_id("globalAssetId", known_id) - id2 = _encode_asset_id("globalAssetId", unknown_id) - response = self.client.get(f"{BASE_PATH}/shells?assetIds={id1}&assetIds={id2}") - self.assertEqual(200, response.status_code) - result = json.loads(response.data) - returned_ids = [r["id"] for r in result] - self.assertIn(aas_list[0].id, returned_ids) - - def test_malformed_asset_id_missing_field_returns_400(self) -> None: - bad_payload = base64.urlsafe_b64encode(b'{"name": "globalAssetId"}').decode() - response = self.client.get(f"{BASE_PATH}/shells?assetIds={bad_payload}") - self.assertEqual(400, response.status_code) diff --git a/server/test/test_api_base_path.py b/server/test/test_api_base_path.py index acf1ceed8..85bfddb56 100644 --- a/server/test/test_api_base_path.py +++ b/server/test/test_api_base_path.py @@ -21,8 +21,6 @@ SERVER_ROOT / "docker" / "discovery" / "Dockerfile", SERVER_ROOT / "docker" / "registry" / "Dockerfile", SERVER_ROOT / "docker" / "repository" / "Dockerfile", - # Tests - SERVER_ROOT / "test" / "interfaces" / "test_shells_asset_ids.py", ] diff --git a/server/test/test_config.default.ini b/server/test/test_config.default.ini new file mode 100644 index 000000000..048fdef86 --- /dev/null +++ b/server/test/test_config.default.ini @@ -0,0 +1,7 @@ +# Configurations for the unittest environment + +# For customizations, please create a new file in this directory, named `test_config.ini` and add the required entries +# to that file to override the defaults defined here. + +[server] +url = http://localhost:8080/api/v3.1