diff --git a/README.md b/README.md index 6f643dd..623057e 100644 --- a/README.md +++ b/README.md @@ -34,7 +34,78 @@ integration role: | Inspect Services and navigate their catalogs | `offering_protocol.agent` | | Publish an ODP Service | `offering_protocol.service` | -## Search the Directory +## Search Services and Collections + +`DirectoryClient.search()` searches indexed Services and explicitly submitted Collections. +The Directory does not crawl complete catalogs or index Offerings. + +```python +import asyncio + +from offering_protocol.directory import ( + CollectionResult, + DirectoryClient, + ResourceSearchRequest, + ServiceResult, + UnknownResult, +) + + +async def main() -> None: + async with DirectoryClient() as directory: + response = await directory.search(ResourceSearchRequest(query="weather forecast", limit=25)) + for item in response.items: + if isinstance(item, ServiceResult): + print("Service:", item.service.name, item.service.service_origin) + elif isinstance(item, CollectionResult): + print( + "Collection:", + item.collection.name, + item.collection.id, + item.service.service_origin, + ) + elif isinstance(item, UnknownResult): + print("Unsupported result type:", item.type) + for issue in response.issues: + print(f"Skipped result {issue.index}: {issue.message}") + + +asyncio.run(main()) +``` + +`types=["service"]` or `types=["collection"]` restricts the result types. Omission selects both; +an explicit list must be nonempty and distinct. Filters apply to the owning Service's metadata. +Omit the query to browse. The default and maximum result limit are 100. + +A Collection is identified by its owning Service origin and case-sensitive Collection ID. +Inspect that Service's live ODP document, then call `ServiceClient.get_collection()` for current +details. The result's `indexed_at` describes Collection freshness; `service.indexed_at` describes +parent freshness. `service.service_id` is the Directory's Service identifier. A Service result can +have `available_through` platform attribution; a Collection's attribution is its owning `service`. + +Malformed known results are omitted and reported in `issues` with their original response index. +Unknown future types retain their full JSON in `UnknownResult.raw`; do not treat them as Services +or execute their metadata. Additional fields are available through `additional`. Directory +metadata does not replace inspection of the Service's own document. + +Mixed search does not currently offer continuation. Missing `next` does **not** mean every match +was returned. Refine the query or filters when needed. `continue_search(next)` follows one opaque +same-origin reference if the server supplies one; the SDK does not invent continuations. Facets +count all matching targets, not just returned items: a Service and two Collections count as three. +Collection search does not require permission to display its card on the Directory landing page. + +`suggest(SuggestionRequest(prefix="we", filters=ServiceFilters(keywords=["weather"])))` +sends POST `/v1/directory/suggestions`. Optional filters use the same `ServiceFilters` as search; +Collection filters apply to the owning Service. `suggest_services()` remains GET and does not +accept filters. `suggest()` matches names, descriptions and keywords, but returns +the **names of matching Services and Collections**, not the text that matched. Matching uses +substrings and whitespace-separated alternative terms despite the parameter name `prefix`. +The server deduplicates names; the default and maximum suggestion limit are 25. These strings +are candidate search queries, not resource identifiers. + +See the [runnable canonical Directory example](examples/README.md#canonical-directory-discovery). + +## Search only Services `DirectoryClient` uses the one canonical production Directory. Pass `Environment.SANDBOX` when working against InFlow's sandbox; the endpoint itself is not configurable. @@ -47,7 +118,7 @@ from offering_protocol.directory import DirectoryClient, Environment, SearchRequ async def main() -> None: async with DirectoryClient(Environment.PRODUCTION) as directory: - page = await directory.search( + page = await directory.search_services( SearchRequest( query="indoor plants", filters=ServiceFilters(keywords=["plants"]), @@ -58,16 +129,29 @@ async def main() -> None: print(service.name, service.service_origin) if page.next: - next_page = await directory.continue_search(page.next) + next_page = await directory.continue_search_services(page.next) print(f"Next page contains {len(next_page.items)} Services") asyncio.run(main()) ``` -Use `search_services()` when the application wants bounded automatic pagination. Search responses +Use `collect_services()` when the application wants bounded automatic pagination. It stops at the +item or response limit without fetching another response. Search responses provide facets for enrollment protocols, keywords, operations, payment protocols, payment options, -and trust protocols. Use `suggest()` to discover keyword completions supported by the Directory. +and trust protocols. Use `suggest_services()` to discover Service-only keyword completions. + +### API migration + +- Service-only `search()` calls become `search_services()`, and `continue_search()` calls become + `continue_search_services()`. +- Aggregating `search_services(request, options)` calls become `collect_services(request, options)`. +- `search_pages()` is removed. Applications that need individual Service-only responses can call + `search_services()` and follow `continue_search_services()` with their own explicit limit. +- Service-only `suggest()` calls become `suggest_services()`. +- `search()`, `continue_search()`, and `suggest()` select mixed discovery. + +`Agent` federated Offering discovery remains Service-only and uses `collect_services()`. ## Inspect and navigate a Service diff --git a/examples/README.md b/examples/README.md index a12db49..7df0752 100644 --- a/examples/README.md +++ b/examples/README.md @@ -29,3 +29,16 @@ typed `Catalog` protocol over their own data source. `agent.py` prints the Service document, lists Collections and Offerings only when those operations are advertised, and fetches full details for the first Offering. It does not invoke an Action. + +## Canonical Directory discovery + +```sh +uv run python examples/directory.py sandbox weather +``` + +Use `production` for the production Directory. Omit `weather` to browse. This example requires a +deployment with `/v1/directory/search`. It requests at most five mixed results, displays Service +and Collection names, reports unusable or unknown results, and retrieves Collection details only +after inspecting the owning Service's advertised anonymous support. It does not enroll, pay or +invoke Actions. Unlike the local Service example above, this uses the real Directory. +The server's bounded result list does not promise every matching result is included. diff --git a/examples/directory.py b/examples/directory.py new file mode 100644 index 0000000..40de119 --- /dev/null +++ b/examples/directory.py @@ -0,0 +1,53 @@ +"""Discover mixed Directory results and retrieve anonymous Collection details.""" + +from __future__ import annotations + +import argparse +import asyncio + +from offering_protocol.agent import ServiceClient +from offering_protocol.core import AuthenticationRequirement, Operation +from offering_protocol.directory import ( + CollectionResult, + DirectoryClient, + Environment, + ResourceSearchRequest, + ServiceResult, + UnknownResult, +) + + +async def discover(environment: Environment, query: str) -> None: + async with DirectoryClient(environment) as directory: + response = await directory.search(ResourceSearchRequest(query=query, limit=5)) + for issue in response.issues: + print(f"Skipped result {issue.index}: {issue.message}") + for item in response.items: + if isinstance(item, ServiceResult): + print(f"Service: {item.service.name} ({item.service.service_origin})") + elif isinstance(item, CollectionResult): + print( + f"Collection: {item.collection.name} " + f"({item.collection.id}, through {item.service.service_origin})" + ) + async with ServiceClient(item.service.service_origin) as service: + inspection = await service.inspect() + if any( + operation.name == Operation.GET_COLLECTION + and operation.authentication != AuthenticationRequirement.REQUIRED + for operation in inspection.document.operations + ): + collection = await service.get_collection(item.collection.id) + print(collection.model_dump_json(indent=2)) + else: + print("The Service does not advertise anonymous Collection retrieval.") + elif isinstance(item, UnknownResult): + print(f"Unsupported result type: {item.type}") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("environment", choices=["production", "sandbox"]) + parser.add_argument("query", nargs="*", help="Omit to browse indexed results") + args = parser.parse_args() + asyncio.run(discover(Environment(args.environment), " ".join(args.query))) diff --git a/scripts/verify-consumer.sh b/scripts/verify-consumer.sh index f695b7f..36ac723 100755 --- a/scripts/verify-consumer.sh +++ b/scripts/verify-consumer.sh @@ -37,6 +37,8 @@ assert agent.__name__ == "offering_protocol.agent" assert core.__name__ == "offering_protocol.core" assert directory.__name__ == "offering_protocol.directory" assert service.__name__ == "offering_protocol.service" +request = directory.ResourceSearchRequest(types=["collection"]) +assert request.to_dict() == {"types": ["collection"]} document = core.parse_service_document( b'{"description":"Consumer smoke test","http":{"endpoint_base":"/odp"},' b'"language":"en","localizations":["en"],"name":"Consumer",' diff --git a/src/offering_protocol/agent/agent.py b/src/offering_protocol/agent/agent.py index 83a801a..c99a919 100644 --- a/src/offering_protocol/agent/agent.py +++ b/src/offering_protocol/agent/agent.py @@ -79,7 +79,7 @@ async def search_offerings_across_services( ) concurrency = _bounded(request.concurrency, 4, 16, "concurrency") try: - services = await self.directory.search_services( + services = await self.directory.collect_services( request.services, IterationOptions(max_items=maximum_services, max_pages=16), ) diff --git a/src/offering_protocol/directory/__init__.py b/src/offering_protocol/directory/__init__.py index 69241db..aa04856 100644 --- a/src/offering_protocol/directory/__init__.py +++ b/src/offering_protocol/directory/__init__.py @@ -6,6 +6,10 @@ DirectoryRequestError, ) from offering_protocol.directory.models import ( + CollectionResult, + CollectionSummary, + DirectoryIssue, + DirectoryResult, DirectoryService, Environment, Facet, @@ -14,10 +18,15 @@ OperationFilter, PaymentFilter, PaymentOptionFacetValue, + ResourceSearchRequest, SearchPage, SearchRequest, + SearchResponse, ServiceFilters, + ServiceReference, + ServiceResult, SuggestionRequest, + UnknownResult, ) from offering_protocol.directory.transport import ( HttpRequest, @@ -28,9 +37,13 @@ ) __all__ = [ + "CollectionResult", + "CollectionSummary", "DirectoryClient", "DirectoryError", + "DirectoryIssue", "DirectoryRequestError", + "DirectoryResult", "DirectoryService", "Environment", "Facet", @@ -42,10 +55,15 @@ "OperationFilter", "PaymentFilter", "PaymentOptionFacetValue", + "ResourceSearchRequest", "SearchPage", "SearchRequest", + "SearchResponse", "ServiceFilters", + "ServiceReference", + "ServiceResult", "SuggestionRequest", "Transport", "TransportError", + "UnknownResult", ] diff --git a/src/offering_protocol/directory/client.py b/src/offering_protocol/directory/client.py index 1f44c08..96a975d 100644 --- a/src/offering_protocol/directory/client.py +++ b/src/offering_protocol/directory/client.py @@ -16,10 +16,13 @@ DirectoryService, Environment, IterationOptions, + ResourceSearchRequest, SearchPage, SearchRequest, + SearchResponse, SuggestionRequest, ) +from offering_protocol.directory.results import parse_search_response from offering_protocol.directory.transport import ( HttpRequest, HttpResponse, @@ -63,7 +66,26 @@ async def aclose(self) -> None: if self._owns_transport: await self._transport.aclose() - async def search(self, request: SearchRequest) -> SearchPage: + async def search(self, request: ResourceSearchRequest) -> SearchResponse: + _validate_search_request(request) + if request.types is not None and ( + not request.types or len(set(request.types)) != len(request.types) + ): + raise DirectoryError("types must contain distinct service or collection values") + response = await self._request( + "POST", + f"{self.environment.origin}/v1/directory/search", + json.dumps( + request.model_dump(mode="json", exclude_defaults=True, exclude_none=True) + ).encode(), + ) + return _parse_mixed_response(response.body) + + async def continue_search(self, next_reference: str) -> SearchResponse: + response = await self._request("GET", self._continuation_url(next_reference)) + return _parse_mixed_response(response.body) + + async def search_services(self, request: SearchRequest) -> SearchPage: _validate_search_request(request) response = await self._request( "POST", @@ -72,54 +94,72 @@ async def search(self, request: SearchRequest) -> SearchPage: ) return _parse_search_page(response.body) - async def continue_search(self, next_reference: str) -> SearchPage: + def _continuation_url(self, next_reference: str) -> str: + if not next_reference.strip(): + raise DirectoryError("Directory continuation is empty") target = urljoin(f"{self.environment.origin}/", next_reference) if derive_service_origin(target) != self.environment.origin: raise DirectoryError("Directory continuation changed canonical origin") - response = await self._request("GET", target) - return _parse_search_page(response.body) + return target - async def search_pages( - self, request: SearchRequest, options: IterationOptions | None = None - ) -> list[SearchPage]: - options = options or IterationOptions() - maximum_pages = _bounded(options.max_pages, 16, 16, "max_pages") - pages: list[SearchPage] = [] - page = await self.search(request) - for page_number in range(maximum_pages): - pages.append(page) - if not page.next: - break - if page_number + 1 < maximum_pages: - page = await self.continue_search(page.next) - return pages + async def continue_search_services(self, next_reference: str) -> SearchPage: + response = await self._request("GET", self._continuation_url(next_reference)) + return _parse_search_page(response.body) - async def search_services( + async def collect_services( self, request: SearchRequest, options: IterationOptions | None = None ) -> list[DirectoryService]: options = options or IterationOptions() maximum_items = _bounded(options.max_items, 10_000, 10_000, "max_items") + maximum_responses = _bounded(options.max_pages, 16, 16, "max_pages") services: list[DirectoryService] = [] - for page in await self.search_pages(request, options): + page = await self.search_services(request) + response_count = 1 + while True: services.extend(page.items[: maximum_items - len(services)]) - if len(services) == maximum_items: + if ( + not page.next + or len(services) == maximum_items + or response_count == maximum_responses + ): break + page = await self.continue_search_services(page.next) + response_count += 1 return services async def suggest(self, request: SuggestionRequest) -> list[str]: + return await self._suggestions("/v1/directory/suggestions", request, mixed=True) + + async def suggest_services(self, request: SuggestionRequest) -> list[str]: + return await self._suggestions("/v1/services/suggestions", request) + + async def _suggestions( + self, path: str, request: SuggestionRequest, *, mixed: bool = False + ) -> list[str]: prefix = request.prefix.strip() if not prefix or len(prefix) > 128: raise DirectoryError("prefix must contain from 1 through 128 characters") - if request.limit > 25: + if request.limit < 0 or request.limit > 25: raise DirectoryError("limit must be from 1 through 25") - query = {"prefix": prefix} - if request.limit: - query["limit"] = str(request.limit) - response = await self._request( - "GET", f"{self.environment.origin}/v1/services/suggestions?{urlencode(query)}" - ) + if mixed: + _validate_search_request(SearchRequest(filters=request.filters)) + payload = request.model_dump(mode="json", exclude_defaults=True, exclude_none=True) + payload["prefix"] = prefix + response = await self._request( + "POST", f"{self.environment.origin}{path}", json.dumps(payload).encode() + ) + else: + if request.filters is not None: + raise DirectoryError("Service-only suggestions do not support filters") + query = {"prefix": prefix} + if request.limit: + query["limit"] = str(request.limit) + response = await self._request( + "GET", f"{self.environment.origin}{path}?{urlencode(query)}" + ) try: - suggestions = json.loads(response.body) + envelope = json.loads(response.body) + suggestions = envelope.get("items") if isinstance(envelope, dict) else None except (UnicodeDecodeError, json.JSONDecodeError) as error: raise DirectoryError(f"invalid Directory suggestions: {error}") from error if ( @@ -161,6 +201,13 @@ async def _request(self, method: str, target: str, body: bytes = b"") -> HttpRes raise DirectoryError("Directory response exceeded its redirect limit") +def _parse_mixed_response(body: bytes) -> SearchResponse: + try: + return parse_search_response(body) + except ValueError as error: + raise DirectoryError(f"invalid Directory response: {error}") from error + + def _parse_search_page(body: bytes) -> SearchPage: try: raw = json.loads(body) @@ -210,7 +257,7 @@ def _normalize_service_protocols(item: dict[str, object]) -> None: def _validate_search_request(request: SearchRequest) -> None: - if request.limit > 100: + if request.limit < 0 or request.limit > 100: raise DirectoryError("limit must be from 1 through 100") if request.query.strip() != request.query or len(request.query) > 512: raise DirectoryError( diff --git a/src/offering_protocol/directory/models.py b/src/offering_protocol/directory/models.py index 4483a2a..37e42c3 100644 --- a/src/offering_protocol/directory/models.py +++ b/src/offering_protocol/directory/models.py @@ -3,9 +3,9 @@ from __future__ import annotations from enum import StrEnum -from typing import Generic, TypeVar +from typing import Generic, Literal, TypeAlias, TypeVar -from pydantic import Field +from pydantic import Field, JsonValue from offering_protocol.core.models import ( AuthenticationRequirement, @@ -59,6 +59,10 @@ class SearchRequest(OdpModel): query: str = "" +class ResourceSearchRequest(SearchRequest): + types: list[Literal["service", "collection"]] | None = None + + class DirectoryService(OdpModel): description: str documentation_url: str = "" @@ -74,6 +78,50 @@ class DirectoryService(OdpModel): support_url: str = "" website_url: str = "" + @property + def service_id(self) -> str | None: + value = self.additional.get("service_id") + return value if isinstance(value, str) else None + + +class ServiceReference(OdpModel): + service_id: str + service_origin: str + name: str | None = None + + +class CollectionSummary(OdpModel): + id: str + name: str + description: str | None = None + + +class ServiceResult(OdpModel): + type: Literal["service"] + service: DirectoryService + indexed_at: str + available_through: ServiceReference | None = None + + +class CollectionResult(OdpModel): + type: Literal["collection"] + service: DirectoryService + indexed_at: str + collection: CollectionSummary + + +class UnknownResult(OdpModel): + type: str + raw: dict[str, JsonValue] + + +DirectoryResult: TypeAlias = ServiceResult | CollectionResult | UnknownResult + + +class DirectoryIssue(OdpModel): + index: int + message: str + class PaymentOptionFacetValue(OdpModel): name: Protocol @@ -100,7 +148,15 @@ class SearchPage(OdpModel): next: str = "" +class SearchResponse(OdpModel): + facets: Facets | None = None + items: list[DirectoryResult] + next: str | None = None + issues: list[DirectoryIssue] = Field(default_factory=list) + + class SuggestionRequest(OdpModel): + filters: ServiceFilters | None = None limit: int = 0 prefix: str diff --git a/src/offering_protocol/directory/results.py b/src/offering_protocol/directory/results.py new file mode 100644 index 0000000..7d9aed2 --- /dev/null +++ b/src/offering_protocol/directory/results.py @@ -0,0 +1,112 @@ +"""Decode mixed Directory results without treating unknown types as Services.""" + +from __future__ import annotations + +import json +from datetime import datetime + +from pydantic import JsonValue, TypeAdapter + +from offering_protocol.core import ( + derive_service_origin, + is_local_resource_identifier, + parse_agent_service_document, +) +from offering_protocol.directory.models import ( + CollectionResult, + DirectoryIssue, + DirectoryResult, + SearchResponse, + ServiceResult, + UnknownResult, +) + +_OBJECT = TypeAdapter(dict[str, JsonValue]) + + +def parse_search_response(body: bytes) -> SearchResponse: + raw = _OBJECT.validate_json(body) + candidates = raw.get("items") + if not isinstance(candidates, list) or len(candidates) > 100: + raise ValueError("items must be an array of at most 100 results") + items: list[DirectoryResult] = [] + issues: list[DirectoryIssue] = [] + for index, candidate in enumerate(candidates): + try: + items.append(_result(candidate)) + except ValueError as error: + issues.append(DirectoryIssue(index=index, message=str(error))) + response = SearchResponse.model_validate({**raw, "items": items, "issues": issues}) + if response.next is not None and not response.next.strip(): + raise ValueError("Directory continuation is empty") + if response.facets is not None and any( + facet.value.name.value != "tap" for facet in response.facets.trust + ): + raise ValueError("Directory trust facets are invalid") + return response + + +def _result(value: JsonValue) -> DirectoryResult: + raw = _OBJECT.validate_python(value) + kind = _text(raw, "type", 128) + if kind not in {"service", "collection"}: + return UnknownResult(type=kind, raw=raw) + _timestamp(raw) + service = _OBJECT.validate_python(raw.get("service")) + _text(service, "service_id", 128) + _origin(service) + _timestamp(service) + for name in ( + "branding", + "http", + "mcp", + "odp_version", + "payment_origins", + "search_capabilities", + ): + service.pop(name, None) + document = parse_agent_service_document( + json.dumps({**service, "odp_version": "1.0", "http": {"endpoint_base": "/"}}) + ) + service["operations"] = [operation.to_dict() for operation in document.operations] + if document.protocols is None: + service.pop("protocols", None) + else: + service["protocols"] = document.protocols.to_dict() + raw["service"] = service + if kind == "service": + if "available_through" in raw: + reference = _OBJECT.validate_python(raw["available_through"]) + _text(reference, "service_id", 128) + _origin(reference) + if "name" in reference: + _text(reference, "name", 128) + return ServiceResult.model_validate(raw) + collection = _OBJECT.validate_python(raw.get("collection")) + if not is_local_resource_identifier(_text(collection, "id", 128)): + raise ValueError("collection.id must be a local resource identifier") + _text(collection, "name", 128) + if "description" in collection: + description = collection["description"] + if not isinstance(description, str) or len(description) > 1024: + raise ValueError("collection.description must be a string of at most 1024 characters") + return CollectionResult.model_validate(raw) + + +def _origin(value: dict[str, JsonValue]) -> None: + origin = _text(value, "service_origin", 2048) + if not origin.startswith("https://") or derive_service_origin(origin) != origin: + raise ValueError("service_origin must be a canonical HTTPS origin") + + +def _timestamp(value: dict[str, JsonValue]) -> None: + timestamp = _text(value, "indexed_at", 64) + if datetime.fromisoformat(timestamp.replace("Z", "+00:00")).tzinfo is None: + raise ValueError("indexed_at must contain a timezone") + + +def _text(value: dict[str, JsonValue], field: str, maximum: int) -> str: + text = value.get(field) + if not isinstance(text, str) or not text.strip() or len(text) > maximum: + raise ValueError(f"{field} must be a nonempty string of at most {maximum} characters") + return text diff --git a/tests/test_directory.py b/tests/test_directory.py index 5419c8f..3377421 100644 --- a/tests/test_directory.py +++ b/tests/test_directory.py @@ -50,16 +50,17 @@ async def test_searches_continues_and_iterates_canonical_directory() -> None: response(DIRECTORY_PAGE, content_type="application/json"), ) client = DirectoryClient(transport=transport) - pages = await client.search_pages( + first_page = await client.search_services( SearchRequest( query="plants", filters=ServiceFilters(trust=[TrustProtocol(name=Protocol.TAP)]), ) ) - assert len(pages) == 2 - assert pages[0].items[0].name == "Indica Flowers" - assert pages[0].facets is not None - assert pages[0].facets.trust[0].value == TrustProtocol(name=Protocol.TAP) + second_page = await client.continue_search_services(first_page.next) + assert len(second_page.items) == 1 + assert first_page.items[0].name == "Indica Flowers" + assert first_page.facets is not None + assert first_page.facets.trust[0].value == TrustProtocol(name=Protocol.TAP) assert transport.requests[0].url == "https://api.inflowpay.ai/v1/services/search" assert transport.requests[0].method == "POST" assert transport.requests[1].method == "GET" @@ -75,14 +76,14 @@ async def test_search_services_is_bounded_and_suggestions_are_typed() -> None: Environment.SANDBOX, transport=QueueTransport( response(DIRECTORY_PAGE, content_type="application/json"), - response('["plant","planter"]', content_type="application/json"), + response('{"items":["plant","planter"]}', content_type="application/json"), ), ) - services = await client.search_services( + services = await client.collect_services( SearchRequest(), IterationOptions(max_items=1, max_pages=1) ) assert services[0].service_origin == "https://demo.inflowpay.ai" - assert await client.suggest(SuggestionRequest(prefix="pla", limit=2)) == [ + assert await client.suggest_services(SuggestionRequest(prefix="pla", limit=2)) == [ "plant", "planter", ] @@ -99,7 +100,7 @@ async def test_search_filters_unknown_protocols_and_rejects_malformed_known() -> candidate = DIRECTORY_PAGE.replace('"items":[{', '"items":[{' + protocols) page = await DirectoryClient( transport=QueueTransport(response(candidate, content_type="application/json")) - ).search(SearchRequest()) + ).search_services(SearchRequest()) parsed = page.items[0].protocols assert parsed is not None assert not parsed.enrollment @@ -116,14 +117,14 @@ async def test_search_filters_unknown_protocols_and_rejects_malformed_known() -> ).replace(',{"name":"tap"}', "") page = await DirectoryClient( transport=QueueTransport(response(unknown_only, content_type="application/json")) - ).search(SearchRequest()) + ).search_services(SearchRequest()) assert page.items[0].protocols is None malformed = candidate.replace('"name":"mpp"', '"name":"mpp","extra":true') with pytest.raises(DirectoryError): await DirectoryClient( transport=QueueTransport(response(malformed, content_type="application/json")) - ).search(SearchRequest()) + ).search_services(SearchRequest()) @pytest.mark.asyncio @@ -132,7 +133,7 @@ async def test_follows_same_origin_redirect_and_changes_post_to_get() -> None: response(b"", headers={"location": "/redirect"}, status=303), response(DIRECTORY_PAGE, content_type="application/json"), ) - await DirectoryClient(transport=transport).search(SearchRequest()) + await DirectoryClient(transport=transport).search_services(SearchRequest()) assert [request.method for request in transport.requests] == ["POST", "GET"] @@ -150,7 +151,7 @@ async def test_follows_same_origin_redirect_and_changes_post_to_get() -> None: ) async def test_rejects_invalid_searches(candidate: SearchRequest) -> None: with pytest.raises(DirectoryError): - await DirectoryClient(transport=QueueTransport()).search(candidate) + await DirectoryClient(transport=QueueTransport()).search_services(candidate) @pytest.mark.asyncio @@ -178,7 +179,9 @@ async def test_rejects_invalid_responses_and_continuations() -> None: ] for candidate in scenarios[:5]: with pytest.raises(DirectoryError): - await DirectoryClient(transport=QueueTransport(candidate)).search(SearchRequest()) + await DirectoryClient(transport=QueueTransport(candidate)).search_services( + SearchRequest() + ) with pytest.raises(DirectoryError): await DirectoryClient(transport=QueueTransport(scenarios[5])).suggest( SuggestionRequest(prefix="b") @@ -198,7 +201,7 @@ async def test_request_error_exposes_status_and_headers() -> None: "blocked", content_type="text/plain", headers={"retry-after": "1"}, status=429 ) ) - ).search(SearchRequest()) + ).search_services(SearchRequest()) assert caught.value.status == 429 assert caught.value.headers["retry-after"] == "1" @@ -208,22 +211,22 @@ async def test_rejects_redirect_failures_and_iteration_bounds() -> None: with pytest.raises(DirectoryError): await DirectoryClient( transport=QueueTransport(response(b"", headers={}, status=302)) - ).search(SearchRequest()) + ).search_services(SearchRequest()) with pytest.raises(DirectoryError): await DirectoryClient( transport=QueueTransport( response(b"", headers={"location": "https://other.example"}, status=302) ) - ).search(SearchRequest()) + ).search_services(SearchRequest()) redirects = [response(b"", headers={"location": "/again"}, status=307) for _ in range(6)] with pytest.raises(DirectoryError): - await DirectoryClient(transport=QueueTransport(*redirects)).search(SearchRequest()) + await DirectoryClient(transport=QueueTransport(*redirects)).search_services(SearchRequest()) with pytest.raises(DirectoryError): - await DirectoryClient(transport=QueueTransport()).search_pages( + await DirectoryClient(transport=QueueTransport()).collect_services( SearchRequest(), IterationOptions(max_pages=17) ) with pytest.raises(DirectoryError): - await DirectoryClient(transport=QueueTransport()).search_services( + await DirectoryClient(transport=QueueTransport()).collect_services( SearchRequest(), IterationOptions(max_items=10_001) ) @@ -235,11 +238,11 @@ async def test_iteration_can_stop_at_page_limit_and_redirect_loop_is_exhaustive( first = DIRECTORY_PAGE[:-2] + ',"next":"/v1/services/search?cursor=two"}' pages = await DirectoryClient( transport=QueueTransport(response(first, content_type="application/json")) - ).search_pages(SearchRequest(), IterationOptions(max_pages=1)) + ).collect_services(SearchRequest(), IterationOptions(max_pages=1)) assert len(pages) == 1 monkeypatch.setattr("offering_protocol.directory.client._MAXIMUM_REDIRECTS", -1) with pytest.raises(DirectoryError, match="redirect limit"): - await DirectoryClient(transport=QueueTransport()).search(SearchRequest()) + await DirectoryClient(transport=QueueTransport()).search_services(SearchRequest()) @pytest.mark.asyncio diff --git a/tests/test_directory_mixed.py b/tests/test_directory_mixed.py new file mode 100644 index 0000000..eaa9a6d --- /dev/null +++ b/tests/test_directory_mixed.py @@ -0,0 +1,315 @@ +from __future__ import annotations + +import asyncio +import json + +import pytest +from pydantic import JsonValue + +from helpers import QueueTransport, response +from offering_protocol.directory import ( + CollectionResult, + DirectoryClient, + DirectoryError, + DirectoryService, + Environment, + IterationOptions, + ResourceSearchRequest, + SearchRequest, + ServiceFilters, + ServiceResult, + SuggestionRequest, + UnknownResult, +) +from offering_protocol.directory.transport import HttpRequest, HttpResponse + + +def service() -> dict[str, JsonValue]: + return { + "service_id": "parent", + "service_origin": "https://api.example.com", + "indexed_at": "2026-09-18T11:00:00Z", + "name": "Data", + "description": "Data services.", + "language": "en", + "localizations": ["en"], + "operations": [ + {"name": "get-offering", "authentication": "not-required"}, + {"name": "list-offerings", "authentication": "not-required"}, + {"name": "future-operation", "authentication": "not-required"}, + ], + "protocols": {"trust": [{"name": "tap"}, {"name": "future"}]}, + } + + +def item(kind: str = "collection") -> dict[str, JsonValue]: + raw: dict[str, JsonValue] = { + "type": kind, + "service": service(), + "indexed_at": "2026-09-18T12:00:00Z", + } + if kind == "collection": + raw["collection"] = { + "id": "Weather", + "name": "Weather forecasts", + "description": "Forecasts.", + } + return raw + + +def transport_for(body: object) -> QueueTransport: + return QueueTransport(response(json.dumps(body), content_type="application/json")) + + +@pytest.mark.asyncio +async def test_mixed_results_metadata_unknown_types_and_requests() -> None: + first = item("service") + first["available_through"] = { + "service_id": "platform", + "service_origin": "https://platform.example", + "name": "Platform", + } + first["extra"] = True + future: dict[str, JsonValue] = {"type": "future", "nested": {"value": 42}} + transport = transport_for( + { + "items": [first, item(), future], + "extra": 42, + "facets": {"keywords": [{"value": "weather", "count": 12}]}, + } + ) + async with DirectoryClient(Environment.SANDBOX, transport=transport) as client: + result = await client.search( + ResourceSearchRequest(query="weather", limit=10, types=["collection", "service"]) + ) + assert result.issues == [] + assert result.additional["extra"] == 42 + assert result.facets is not None and result.facets.keywords[0].count == 12 + first_result, second, third = result.items + assert isinstance(first_result, ServiceResult) + assert first_result.service.service_id == "parent" + assert first_result.available_through is not None + assert first_result.available_through.name == "Platform" + assert first_result.additional["extra"] is True + assert len(first_result.service.operations) == 2 + assert first_result.service.protocols is not None + assert len(first_result.service.protocols.trust) == 1 + assert isinstance(second, CollectionResult) + assert second.collection.id == "Weather" + assert second.indexed_at != second.service.indexed_at + assert isinstance(third, UnknownResult) + assert third.type == "future" and third.raw == future + request = transport.requests[0] + assert request.method == "POST" + assert request.url == "https://sandbox.inflowpay.ai/v1/directory/search" + assert json.loads(request.body) == { + "query": "weather", + "limit": 10, + "types": ["collection", "service"], + } + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "path,value", + [ + ("type", None), + ("service/service_id", ""), + ("service/service_origin", "http://api.example.com"), + ("service/service_origin", "https://api.example.com/path"), + ("service/operations", []), + ("indexed_at", "yesterday"), + ("indexed_at", "2026-09-18T00:00:00"), + ("collection/id", "../bad"), + ("collection/name", ""), + ("collection/description", None), + ("collection/description", "x" * 1025), + ("service", None), + ], +) +async def test_bad_known_items_do_not_discard_valid_items(path: str, value: JsonValue) -> None: + invalid = item() + parent = invalid + parts = path.split("/") + for part in parts[:-1]: + child = parent[part] + assert isinstance(child, dict) + parent = child + parent[parts[-1]] = value + result = await DirectoryClient(transport=transport_for({"items": [invalid, item()]})).search( + ResourceSearchRequest() + ) + assert len(result.items) == 1 + assert len(result.issues) == 1 and result.issues[0].index == 0 + + +@pytest.mark.asyncio +async def test_optional_members_and_unverified_metadata() -> None: + first = item("service") + parent = service() + parent.pop("protocols") + parent["http"] = {"endpoint_base": "https://untrusted.example/"} + first["service"] = parent + first["available_through"] = { + "service_id": "platform", + "service_origin": "https://platform.example", + } + empty_description = item() + empty_description["collection"] = {"id": "Weather", "name": "Weather", "description": ""} + omitted_description = item() + omitted_description["collection"] = {"id": "Weather", "name": "Weather"} + result = await DirectoryClient( + transport=transport_for( + {"items": [first, item("service"), empty_description, omitted_description]} + ) + ).search(ResourceSearchRequest()) + assert not result.issues + parsed = result.items[0] + assert isinstance(parsed, ServiceResult) + assert "http" not in parsed.service.additional + assert parsed.service.protocols is None + assert parsed.available_through is not None and parsed.available_through.name is None + parent = parsed.service.to_dict() + parent.pop("service_id") + assert DirectoryService.model_validate(parent).service_id is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "body", + [ + None, + {}, + {"items": None}, + {"items": [item()] * 101}, + {"items": [], "next": " "}, + {"items": [], "next": False}, + {"items": [], "facets": {"trust": [{"value": {"name": "mpp"}, "count": 1}]}}, + ], +) +async def test_invalid_envelopes(body: object) -> None: + with pytest.raises(DirectoryError): + await DirectoryClient(transport=transport_for(body)).search(ResourceSearchRequest()) + + +@pytest.mark.asyncio +async def test_mixed_continuation_suggestions_and_request_validation() -> None: + transport = transport_for({"items": [], "next": "/v1/directory/search?cursor=opaque"}) + client = DirectoryClient(transport=transport) + page = await client.continue_search("/v1/directory/search?cursor=opaque") + assert page.next == "/v1/directory/search?cursor=opaque" + assert transport.requests[0].method == "GET" and not transport.requests[0].body + for request in [ + ResourceSearchRequest(types=[]), + ResourceSearchRequest(types=["service", "service"]), + ResourceSearchRequest(limit=-1), + ]: + with pytest.raises(DirectoryError): + await client.search(request) + with pytest.raises(DirectoryError, match="empty"): + await client.continue_search(" ") + with pytest.raises(DirectoryError): + await client.suggest(SuggestionRequest(prefix="we", limit=-1)) + assert len(transport.requests) == 1 + transport = transport_for({"items": ["Weather forecasts"]}) + names = await DirectoryClient(transport=transport).suggest(SuggestionRequest(prefix="we")) + assert names == ["Weather forecasts"] + assert transport.requests[0].url == "https://api.inflowpay.ai/v1/directory/suggestions" + assert transport.requests[0].method == "POST" + assert json.loads(transport.requests[0].body) == {"prefix": "we"} + transport = transport_for({"items": []}) + await DirectoryClient(transport=transport).search(ResourceSearchRequest()) + assert json.loads(transport.requests[0].body) == {} + + +@pytest.mark.asyncio +async def test_service_suggestions_without_limit() -> None: + transport = transport_for({"items": ["weather"]}) + assert await DirectoryClient(transport=transport).suggest_services( + SuggestionRequest(prefix="we") + ) == ["weather"] + assert transport.requests[0].method == "GET" + assert transport.requests[0].url.endswith("/v1/services/suggestions?prefix=we") + + +@pytest.mark.asyncio +async def test_suggestion_filters() -> None: + transport = transport_for({"items": ["Weather"]}) + client = DirectoryClient(transport=transport) + request = SuggestionRequest( + prefix=" we ", limit=5, filters=ServiceFilters(keywords=["weather"]) + ) + assert await client.suggest(request) == ["Weather"] + assert json.loads(transport.requests[0].body) == { + "prefix": "we", + "limit": 5, + "filters": {"keywords": ["weather"]}, + } + assert request.prefix == " we " + with pytest.raises(DirectoryError, match="do not support filters"): + await client.suggest_services(request) + with pytest.raises(DirectoryError, match="keywords"): + await client.suggest(SuggestionRequest(prefix="we", filters=ServiceFilters(keywords=[""]))) + assert len(transport.requests) == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "body", + [ + {"items": [" bad"]}, + {"items": ["name"] * 26}, + {"items": [False]}, + {"items": [""]}, + {"items": ["x" * 129]}, + ["name"], + ], +) +async def test_invalid_suggestion_envelopes(body: object) -> None: + with pytest.raises(DirectoryError): + await DirectoryClient(transport=transport_for(body)).suggest(SuggestionRequest(prefix="x")) + + +@pytest.mark.asyncio +async def test_collection_and_service_aggregation_are_separate_and_bounded() -> None: + parent = service() + operations = parent["operations"] + assert isinstance(operations, list) + operations.pop() + body = {"items": [parent], "next": "/v1/services/search?cursor=more"} + for options in [IterationOptions(max_items=1), IterationOptions(max_pages=1)]: + transport = transport_for(body) + items = await DirectoryClient(transport=transport).collect_services( + SearchRequest(), options + ) + assert len(items) == 1 and len(transport.requests) == 1 + transport = transport_for(body) + transport.responses.append( + response(json.dumps({"items": [parent]}), content_type="application/json") + ) + items = await DirectoryClient(transport=transport).collect_services(SearchRequest()) + assert len(items) == 2 and len(transport.requests) == 2 + + +@pytest.mark.asyncio +async def test_cancellation_propagates_without_followup_requests() -> None: + started = asyncio.Event() + requests: list[HttpRequest] = [] + + class WaitingTransport: + async def send(self, request: HttpRequest) -> HttpResponse: + requests.append(request) + started.set() + await asyncio.Future[None]() + raise AssertionError("cancelled request resumed") + + async def aclose(self) -> None: + pass + + client = DirectoryClient(transport=WaitingTransport()) + task = asyncio.create_task(client.search(ResourceSearchRequest())) + await started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert len(requests) == 1 diff --git a/tests/test_edges.py b/tests/test_edges.py index ef1c67e..2dfaf90 100644 --- a/tests/test_edges.py +++ b/tests/test_edges.py @@ -195,7 +195,7 @@ async def test_directory_response_edge_cases_and_real_transport_adapter( content_type="application/json", ) ) - ).search(SearchRequest()) + ).search_services(SearchRequest()) with pytest.raises(DirectoryError): await DirectoryClient( transport=QueueTransport( @@ -206,7 +206,7 @@ async def test_directory_response_edge_cases_and_real_transport_adapter( content_type="application/json", ) ) - ).search(SearchRequest()) + ).search_services(SearchRequest()) async def handler(request: httpx.Request) -> httpx.Response: assert request.url.host == "93.184.216.34"