server: Accept spec-compliant SpecificAssetId in GET /shells?assetIds

server/app/interfaces/repository.py

@@ -9,7 +9,6 @@
 """
 
 import io
-import json
 from typing import Callable, Dict, Iterable, Iterator, List, Optional, Tuple, Type, Union
 
 import werkzeug.exceptions
@@ -25,7 +24,7 @@
 from app._config import API_BASE_PATH
 from app.interfaces.base import PagingMetadata
 from app.model import ServiceDescription, ServiceSpecificationProfileEnum
-from app.util.converters import IdentifierToBase64URLConverter, IdShortPathConverter, base64url_decode
+from app.util.converters import IdentifierToBase64URLConverter, IdShortPathConverter
 
 from .base import APIResponse, HTTPApiDecoder, ObjectStoreWSGIApp, T, is_stripped_request
 
@@ -450,28 +449,29 @@ def _get_shells(
             global_asset_ids = []
 
             for asset_id in asset_ids:
-                asset_id_json = base64url_decode(asset_id)
-                asset_dict = json.loads(asset_id_json)
-                try:
-                    name = asset_dict["name"]
-                    value = asset_dict["value"]
-                except KeyError as e:
-                    raise BadRequest(f"Invalid assetId format: missing field {e}") from e
-
-                if name == "specificAssetId":
-                    decoded_specific_id = HTTPApiDecoder.json_list(value, model.SpecificAssetId, False, True)[0]
-                    specific_asset_ids.append(decoded_specific_id)
-                elif name == "globalAssetId":
-                    global_asset_ids.append(value)
-
-            # Filter AAS based on both SpecificAssetIds and globalAssetIds
+                # Each assetIds value is a base64url-encoded SpecificAssetId (AAS Part 2, v3.1.1).
+                # The reserved name "globalAssetId" matches AssetInformation/globalAssetId, any other
+                # name matches an entry of AssetInformation/specific_asset_id. Malformed input
+                # raises BadRequest (400) instead of escaping as a 500.
+                decoded_asset_id = HTTPApiDecoder.base64url_json(asset_id, model.SpecificAssetId, False)
+                if decoded_asset_id.name == "globalAssetId":
+                    global_asset_ids.append(decoded_asset_id.value)
+                else:
+                    specific_asset_ids.append(decoded_asset_id)
+
+            # Filter AAS based on both SpecificAssetIds and globalAssetIds. A queried SpecificAssetId
+            # matches by (name, value): full object equality would also require matching
+            # external_subject_id/semantic_id, which a plain {"name", "value"} query does not carry.
             aas = filter(
                 lambda shell: (
                     (
                         not specific_asset_ids
                         or all(
-                            specific_asset_id in shell.asset_information.specific_asset_id
-                            for specific_asset_id in specific_asset_ids
+                            any(
+                                stored.name == queried.name and stored.value == queried.value
+                                for stored in shell.asset_information.specific_asset_id
+                            )
+                            for queried in specific_asset_ids
                         )
                     )
                     and (not global_asset_ids or shell.asset_information.global_asset_id in global_asset_ids)

server/test/interfaces/test_shells_asset_ids.py

@@ -22,6 +22,10 @@ def _encode_asset_id(name: str, value: str) -> str:
     return base64.urlsafe_b64encode(payload.encode()).decode()
 
 
+def _result_ids(response) -> list:
+    return [shell["id"] for shell in json.loads(response.data)["result"]]
+
+
 class ShellsAssetIdsTest(unittest.TestCase):
     def setUp(self) -> None:
         self.example_data = create_full_example()
@@ -37,11 +41,55 @@ def test_multiple_global_asset_ids_returns_matching_results(self) -> None:
         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]
+        returned_ids = _result_ids(response)
         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)
+
+    def test_spec_specific_asset_id_without_match_returns_empty(self) -> None:
+        # AAS Part 2, v3.1.1: each assetIds value is a base64url-encoded SpecificAssetId.
+        # A query for an id no shell carries must return an empty result, not all shells.
+        spec_id = _encode_asset_id("serialNumber", "does-not-exist")
+        response = self.client.get(f"{BASE_PATH}/shells?assetIds={spec_id}")
+        self.assertEqual(200, response.status_code)
+        self.assertEqual([], _result_ids(response))
+
+    def test_spec_specific_asset_id_with_match_returns_shell(self) -> None:
+        aas_list = [obj for obj in self.example_data if isinstance(obj, model.AssetAdministrationShell)]
+        known_specific_id = next(iter(aas_list[0].asset_information.specific_asset_id))
+        spec_id = _encode_asset_id(known_specific_id.name, known_specific_id.value)
+        response = self.client.get(f"{BASE_PATH}/shells?assetIds={spec_id}")
+        self.assertEqual(200, response.status_code)
+        self.assertEqual([aas_list[0].id], _result_ids(response))
+
+    def test_spec_specific_asset_ids_all_must_match(self) -> None:
+        aas_list = [obj for obj in self.example_data if isinstance(obj, model.AssetAdministrationShell)]
+        known_specific_id = next(iter(aas_list[0].asset_information.specific_asset_id))
+        matching_id = _encode_asset_id(known_specific_id.name, known_specific_id.value)
+        non_matching_id = _encode_asset_id("serialNumber", "does-not-exist")
+        response = self.client.get(f"{BASE_PATH}/shells?assetIds={matching_id}&assetIds={non_matching_id}")
+        self.assertEqual(200, response.status_code)
+        self.assertEqual([], _result_ids(response))
+
+    def test_malformed_asset_id_json_returns_400(self) -> None:
+        # A value that is not valid JSON must be rejected with 400, not escape as a 500.
+        bad_payload = base64.urlsafe_b64encode(b"foo").decode()
+        response = self.client.get(f"{BASE_PATH}/shells?assetIds={bad_payload}")
+        self.assertEqual(400, response.status_code)
+
+    def test_old_specific_asset_id_wrapper_no_longer_matches(self) -> None:
+        # The pre-spec wrapper {"name": "specificAssetId", "value": "<SpecificAssetId as JSON string>"}
+        # is no longer understood: it decodes as a SpecificAssetId named "specificAssetId", which no
+        # shell carries, so the result is empty. This documents the breaking change of this fix.
+        aas_list = [obj for obj in self.example_data if isinstance(obj, model.AssetAdministrationShell)]
+        known_specific_id = next(iter(aas_list[0].asset_information.specific_asset_id))
+        wrapper = _encode_asset_id(
+            "specificAssetId",
+            json.dumps({"name": known_specific_id.name, "value": known_specific_id.value}),
+        )
+        response = self.client.get(f"{BASE_PATH}/shells?assetIds={wrapper}")
+        self.assertEqual(200, response.status_code)
+        self.assertEqual([], _result_ids(response))