fix(service): provide defaults for optional catalog operations

Author: nkavian

README.md

@@ -238,7 +238,9 @@ service = (
 Every Service integration must implement `list-offerings` and `get-offering`. `StaticCatalog` is the
 small-Service implementation: it adds Collection operations when Collections are provided and uses
 integrity-protected, stateless continuations that expire after one hour. Larger Services can
-implement the typed `Catalog` protocol over their existing indexed catalog and search infrastructure.
+subclass the typed `Catalog` protocol over their existing indexed catalog and search infrastructure.
+The protocol supplies rejecting defaults for optional operations; override an optional method and
+include its matching `Operation` only when the Service implements it.
 
 Service responses are validated against the bundled normative schemas before they are returned.
 The handler enforces fixed operation paths and methods, ODP media types, request and response byte

src/offering_protocol/service/service.py

@@ -98,21 +98,29 @@ async def get_offering(self, identifier: str, request: CatalogRequest) -> Offeri
 
     async def search_offerings(
         self, query: OfferingSearchRequest, request: CatalogRequest
-    ) -> OfferingPage[Offering]: ...
+    ) -> OfferingPage[Offering]:
+        del query, request
+        raise CatalogError("Offering search is not supported")
 
-    async def list_collections(self, request: CatalogRequest) -> Page[Collection]: ...
+    async def list_collections(self, request: CatalogRequest) -> Page[Collection]:
+        del request
+        raise CatalogError("Collection listing is not supported")
 
-    async def get_collection(
-        self, identifier: str, request: CatalogRequest
-    ) -> Collection | None: ...
+    async def get_collection(self, identifier: str, request: CatalogRequest) -> Collection | None:
+        del identifier, request
+        raise CatalogError("Collection retrieval is not supported")
 
     async def search_collections(
         self, query: CollectionSearchRequest, request: CatalogRequest
-    ) -> Page[Collection]: ...
+    ) -> Page[Collection]:
+        del query, request
+        raise CatalogError("Collection search is not supported")
 
     async def list_collection_offerings(
         self, collection_id: str, request: CatalogRequest
-    ) -> OfferingPage[Offering]: ...
+    ) -> OfferingPage[Offering]:
+        del collection_id, request
+        raise CatalogError("Collection Offering listing is not supported")
 
 
 class ServiceBuilder:

tests/test_service.py

@@ -283,7 +283,7 @@ async def get_collection(
     assert (await service.handle(Request("GET", "/odp/offerings/rubber-plant"))).status == 500
     assert (await service.handle(Request("GET", "/odp/collections/plants"))).status == 500
 
-    class ApplicationCatalog:
+    class ApplicationCatalog(Catalog):
         def operations(self) -> list[Operation]:
             return [Operation.GET_OFFERING, Operation.LIST_OFFERINGS]
 
@@ -296,9 +296,25 @@ async def get_offering(self, identifier: str, request: CatalogRequest) -> Offeri
             del identifier, request
             return None
 
-    application_service = _service(cast(Catalog, ApplicationCatalog()))
+    application_service = _service(ApplicationCatalog())
     assert (await application_service.handle(Request("GET", "/odp/offerings"))).status == 500
 
+    application_catalog = ApplicationCatalog()
+    with pytest.raises(CatalogError):
+        await application_catalog.search_offerings(
+            OfferingSearchRequest(query="plant"), CatalogRequest()
+        )
+    with pytest.raises(CatalogError):
+        await application_catalog.list_collections(CatalogRequest())
+    with pytest.raises(CatalogError):
+        await application_catalog.get_collection("plants", CatalogRequest())
+    with pytest.raises(CatalogError):
+        await application_catalog.search_collections(
+            CollectionSearchRequest(query="plant"), CatalogRequest()
+        )
+    with pytest.raises(CatalogError):
+        await application_catalog.list_collection_offerings("plants", CatalogRequest())
+
     enrollment = EnrollmentProtocol(name=Protocol.AEP)
     payment = PaymentProtocol(
         authentication=AuthenticationRequirement.NOT_REQUIRED, name=Protocol.MPP