Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 68 additions & 0 deletions tests/runtime/test_protocol_conformance.py
Original file line number Diff line number Diff line change
@@ -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()
Comment on lines +45 to +51

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
Loading