From 36022ff2bde29c4cc1a364d431a09e32cd9553d9 Mon Sep 17 00:00:00 2001 From: Vlad Cimpeanu Date: Tue, 14 Jul 2026 10:22:44 +0300 Subject: [PATCH] test: add protocol conformance pins for runtime implementations Typed assignments that make mypy verify each implementation against the protocol surface of the installed uipath-runtime version. A dependency bump that adds or changes protocol members fails typecheck on these lines until the implementations catch up, without this file changing. Covers UiPathLangGraphRuntime, UiPathLangGraphRuntimeFactory and SqliteResumableStorage, wired through UiPathResumableRuntime exactly as in UiPathLangGraphRuntimeFactory._create_runtime_instance. Co-Authored-By: Claude Fable 5 --- tests/runtime/test_protocol_conformance.py | 68 ++++++++++++++++++++++ 1 file changed, 68 insertions(+) create mode 100644 tests/runtime/test_protocol_conformance.py diff --git a/tests/runtime/test_protocol_conformance.py b/tests/runtime/test_protocol_conformance.py new file mode 100644 index 000000000..26023f506 --- /dev/null +++ b/tests/runtime/test_protocol_conformance.py @@ -0,0 +1,68 @@ +"""Protocol conformance for uipath-runtime protocol implementations. + +Every protocol-annotated assignment below is a typed boundary: mypy verifies +the implementation against the protocol surface of the installed +uipath-runtime version. A dependency bump that adds or changes protocol +members fails typecheck on these lines until the implementations catch up. + +The wiring mirrors UiPathLangGraphRuntimeFactory._create_runtime_instance, +so the exact production composition is what gets checked. Deliberately +construction-only: no runtime behavior is exercised here. +""" + +from typing import TypedDict + +from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver +from langgraph.graph import END, START, StateGraph +from uipath.platform.resume_triggers import UiPathResumeTriggerHandler +from uipath.runtime import ( + UiPathResumableRuntime, + UiPathResumableStorageProtocol, + UiPathResumeTriggerProtocol, + UiPathRuntimeContext, + UiPathRuntimeFactoryProtocol, + UiPathRuntimeProtocol, +) + +from uipath_langchain.runtime.factory import UiPathLangGraphRuntimeFactory +from uipath_langchain.runtime.runtime import UiPathLangGraphRuntime +from uipath_langchain.runtime.storage import SqliteResumableStorage + + +class _State(TypedDict, total=False): + value: str + + +async def test_langgraph_wiring_satisfies_runtime_protocols() -> None: + graph = StateGraph(_State) + graph.add_node("noop", lambda state: state) + graph.add_edge(START, "noop") + graph.add_edge("noop", END) + + async with AsyncSqliteSaver.from_conn_string(":memory:") as memory: + compiled_graph = graph.compile(checkpointer=memory) + + delegate: UiPathRuntimeProtocol = UiPathLangGraphRuntime( + graph=compiled_graph, + runtime_id="protocol-conformance-test", + entrypoint="test", + ) + storage: UiPathResumableStorageProtocol = SqliteResumableStorage(memory) + trigger_manager: UiPathResumeTriggerProtocol = UiPathResumeTriggerHandler() + + runtime: UiPathRuntimeProtocol = UiPathResumableRuntime( + delegate=delegate, + storage=storage, + trigger_manager=trigger_manager, + runtime_id="protocol-conformance-test", + ) + + assert runtime is not None + + +def test_langgraph_factory_satisfies_factory_protocol() -> None: + factory: UiPathRuntimeFactoryProtocol = UiPathLangGraphRuntimeFactory( + context=UiPathRuntimeContext() + ) + + assert factory is not None