Implement testing for server package

.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 }}

server/pyproject.toml

@@ -42,6 +42,7 @@ dependencies = [
 
 [project.optional-dependencies]
 dev = [
+    "coverage",
     "mypy",
     "pycodestyle",
     "ruff==0.16.0",

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

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/<aas_id>
+
+    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()))

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)

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)