feat(tests): add small integration test

Author: ionmincu

.python-version

@@ -1 +1 @@
-3.10
+3.11

pyproject.toml

@@ -16,6 +16,7 @@ dependencies = [
     "python-dotenv>=1.0.1",
     "httpx>=0.27.0",
     "openai>=1.65.5",
+    "pytest-asyncio>=1.0.0",
 ]
 classifiers = [
     "Development Status :: 3 - Alpha",

tests/cli_run/samples/1-simple-graph/langgraph.json

@@ -0,0 +1,9 @@
+{
+  "dependencies": [
+    "."
+  ],
+  "graphs": {
+    "agent": "./main.py:graph"
+  },
+  "env": ".env"
+}

tests/cli_run/samples/1-simple-graph/main.py

@@ -0,0 +1,69 @@
+import random
+from typing import Literal
+
+from langgraph.checkpoint.memory import MemorySaver
+from langgraph.graph import END, START, StateGraph
+from langgraph.types import interrupt
+from typing_extensions import TypedDict
+from uipath.models import CreateAction
+
+
+# State
+class State(TypedDict):
+    graph_state: str
+
+
+# Conditional edge
+def decide_mood(state) -> Literal["node_2", "node_3"]:
+    # Often, we will use state to decide on the next node to visit
+    user_input = state["graph_state"]
+
+    # x = interrupt("Xyz")
+    x = interrupt(CreateAction(title="Xyz", description="Xyz description"))
+
+    # Here, let's just do a 50 / 50 split between nodes 2, 3
+    if random.random() < 0.5:
+        # 50% of the time, we return Node 2
+        return "node_2"
+
+    # 50% of the time, we return Node 3
+    return "node_3"
+
+
+# Nodes
+def node_1(state):
+    print("---Node 1---")
+    simple_interrupt = interrupt("question: Who are you?")
+
+    return {"graph_state": "Hello, I am " + simple_interrupt["answer"] + "!"}
+
+
+def node_2(state):
+    print("---Node 2---")
+    action_interrupt = interrupt(
+        CreateAction(
+            app_name="Test-app", title="Test-title", description="Test-description"
+        )
+    )
+    return {"graph_state": state["graph_state"] + action_interrupt["ActionData"]}
+
+
+def node_3(state):
+    print("---Node 3---")
+    return {"graph_state": state["graph_state"] + " end"}
+
+
+builder = StateGraph(State)
+builder.add_node("node_1", node_1)
+builder.add_node("node_2", node_2)
+builder.add_node("node_3", node_3)
+
+builder.add_edge(START, "node_1")
+builder.add_edge("node_1", "node_2")
+builder.add_edge("node_2", "node_3")
+builder.add_edge("node_3", END)
+
+
+memory = MemorySaver()
+
+graph = builder.compile(checkpointer=memory)

tests/cli_run/samples/1-simple-graph/pyproject.toml

@@ -0,0 +1,15 @@
+[project]
+name = "c-host-in-uipath"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+authors = [{ name = "Eduard Stanculet", email = "[email protected]" }]
+requires-python = ">=3.13"
+dependencies = [
+    "langchain-anthropic>=0.3.10",
+    "langchain-community>=0.3.21",
+    "langgraph>=0.3.29",
+    "tavily-python>=0.5.4",
+    "uipath>=2.0.8",
+    "uipath-langchain>=0.0.88",
+]

tests/cli_run/samples/1-simple-graph/uipath.json

@@ -0,0 +1,18 @@
+{
+  "entryPoints": [
+    {
+      "filePath": "agent",
+      "uniqueId": "dcc7a309-fbcc-4999-af4f-2a75a844b49a",
+      "type": "agent",
+      "input": {
+        "type": "string",
+        "title": "graph_state"
+      },
+      "output": {}
+    }
+  ],
+  "bindings": {
+    "version": "2.0",
+    "resources": []
+  }
+}

tests/cli_run/test_run_sample.py

@@ -0,0 +1,103 @@
+import os
+import sys
+import uuid
+from unittest.mock import MagicMock, patch
+
+import pytest
+from dotenv import load_dotenv
+from uipath._cli._runtime._contracts import UiPathTraceContext
+
+from uipath_langchain._cli._runtime._context import LangGraphRuntimeContext
+from uipath_langchain._cli._runtime._runtime import LangGraphRuntime
+from uipath_langchain._cli._utils._graph import LangGraphConfig
+
+load_dotenv()
+
+
+def _create_test_runtime_context(config: LangGraphConfig) -> LangGraphRuntimeContext:
+    """Helper function to create and configure LangGraphRuntimeContext for tests."""
+    context = LangGraphRuntimeContext.from_config(
+        os.environ.get("UIPATH_CONFIG_PATH", "uipath.json")
+    )
+
+    context.entrypoint = (
+        None  # Or a specific graph name if needed, None will pick the single one
+    )
+    context.input = '{ "graph_state": "GET Assets API does not enforce proper permissions Assets.View" }'
+    context.resume = False
+    context.langgraph_config = config
+    context.logs_min_level = os.environ.get("LOG_LEVEL", "INFO")
+    context.job_id = str(uuid.uuid4())
+    context.trace_id = str(uuid.uuid4())
+    # Convert string "True" or "False" to boolean for tracing_enabled
+    tracing_enabled_str = os.environ.get("UIPATH_TRACING_ENABLED", "True")
+    context.tracing_enabled = tracing_enabled_str.lower() == "true"
+    context.trace_context = UiPathTraceContext(
+        enabled=context.tracing_enabled,
+        trace_id=str(
+            uuid.uuid4()
+        ),  # Consider passing trace_id if it needs to match context.trace_id
+        parent_span_id=os.environ.get("UIPATH_PARENT_SPAN_ID"),
+        root_span_id=os.environ.get("UIPATH_ROOT_SPAN_ID"),
+        job_id=os.environ.get(
+            "UIPATH_JOB_KEY"
+        ),  # Consider passing job_id if it needs to match context.job_id
+        org_id=os.environ.get("UIPATH_ORGANIZATION_ID"),
+        tenant_id=os.environ.get("UIPATH_TENANT_ID"),
+        process_key=os.environ.get("UIPATH_PROCESS_UUID"),
+        folder_key=os.environ.get("UIPATH_FOLDER_KEY"),
+    )
+    # Convert string "True" or "False" to boolean
+    langsmith_tracing_enabled_str = os.environ.get("LANGSMITH_TRACING", "False")
+    context.langsmith_tracing_enabled = langsmith_tracing_enabled_str.lower() == "true"
+    return context
+
+
+@pytest.mark.asyncio
+async def test_langgraph_runtime():
+    test_folder_path = os.path.dirname(os.path.abspath(__file__))
+    sample_path = os.path.join(test_folder_path, "samples", "1-simple-graph")
+
+    sys.path.append(sample_path)
+    os.chdir(sample_path)
+
+    config = LangGraphConfig()
+    if not config.exists:
+        raise AssertionError("langgraph.json not found in sample path")
+
+    context = _create_test_runtime_context(config)
+
+    # Mocking UiPath SDK for action creation
+    with patch("uipath_langchain._cli._runtime._output.UiPath") as MockUiPathClass:
+        mock_uipath_sdk_instance = MagicMock()
+        MockUiPathClass.return_value = mock_uipath_sdk_instance
+        mock_actions_client = MagicMock()
+        mock_uipath_sdk_instance.actions = mock_actions_client
+
+        mock_created_action = MagicMock()
+        mock_created_action.key = "mock_action_key_from_test"
+        mock_actions_client.create.return_value = mock_created_action
+
+        result = None
+        async with LangGraphRuntime.from_context(context) as runtime:
+            result = await runtime.execute()
+            print("Result:", result)
+
+        context.resume = True
+        context.input = '{ "answer":  "John Doe"}'  # Simulate some resume data
+        async with LangGraphRuntime.from_context(context) as runtime:
+            result = await runtime.execute()
+            print("Result:", result)
+
+        context.resume = True
+        context.input = (
+            '{ "ActionData": "Test-ActionData" }'  # Simulate some resume data
+        )
+        async with LangGraphRuntime.from_context(context) as runtime:
+            result = await runtime.execute()
+            print("Result:", result)
+
+        assert result is not None, "Result should not be None after execution"
+        assert (
+            result.output["graph_state"] == "Hello, I am John Doe!Test-ActionData end"
+        )

tests/conftest.py

@@ -0,0 +1,95 @@
+# import logging
+# import os
+# from os import environ as env
+# from typing import Generator, Optional
+
+# import httpx
+# import pytest
+# from langchain.embeddings import CacheBackedEmbeddings
+# from langchain.globals import set_llm_cache
+# from langchain.storage import LocalFileStore
+# from langchain_community.cache import SQLiteCache
+
+# from uipath_langchain.embeddings import UiPathOpenAIEmbeddings
+# from uipath_langchain.utils._settings import UiPathCachedPathsSettings
+
+# test_cache_settings = UiPathCachedPathsSettings(
+#     CACHED_COMPLETION_DB="tests/llm_cache/tests_uipath_cache.sqlite",
+#     CACHED_EMBEDDINGS_DIR="tests/llm_cache/cached_embeddings",
+# )
+
+
+# def get_from_uipath_url():
+#     try:
+#         url = os.getenv("UIPATH_URL", "https://cloud.uipath.com/dummyOrg/dummyTennant/")
+#         if url:
+#             return "/".join(url.split("/", 3)[:3])
+#     except Exception:
+#         return "https://cloud.uipath.com/dummyOrg/dummyTennant/"
+#     return None
+
+
+# def get_token():
+#     url_get_token = f"{get_from_uipath_url().rstrip('/')}/identity_/connect/token"
+
+#     os.environ["UIPATH_REQUESTING_PRODUCT"] = "uipath-python-sdk"
+#     os.environ["UIPATH_REQUESTING_FEATURE"] = "langgraph-agent"
+#     os.environ["UIPATH_TESTS_CACHE_LLMGW"] = "true"
+
+#     token_credentials = {
+#         "client_id": env.get("UIPATH_CLIENT_ID"),
+#         "client_secret": env.get("UIPATH_CLIENT_SECRET"),
+#         "grant_type": "client_credentials",
+#     }
+
+#     try:
+#         with httpx.Client() as client:
+#             response = client.post(url_get_token, data=token_credentials)
+#             response.raise_for_status()
+#             res_json = response.json()
+#             token = res_json.get("access_token")
+
+#             if not token:
+#                 pytest.skip("Authentication token is empty or missing")
+#     except (httpx.HTTPError, ValueError, KeyError) as e:
+#         pytest.skip(f"Failed to obtain authentication token: {str(e)}")
+
+#     return token
+
+
+# @pytest.fixture(autouse=True)
+# def setup_test_env():
+#     env["UIPATH_ACCESS_TOKEN"] = get_token()
+
+
+# @pytest.fixture(scope="session")
+# def cached_llmgw_calls() -> Generator[Optional[SQLiteCache], None, None]:
+#     if not os.environ.get("UIPATH_TESTS_CACHE_LLMGW"):
+#         yield None
+#     else:
+#         logging.info("Setting up LLMGW cache")
+#         db_path = test_cache_settings.cached_completion_db
+#         os.makedirs(os.path.dirname(db_path), exist_ok=True)
+#         cache = SQLiteCache(database_path=db_path)
+#         set_llm_cache(cache)
+#         yield cache
+#     set_llm_cache(None)
+#     return
+
+
+# @pytest.fixture(scope="session")
+# def cached_embedder() -> Generator[Optional[CacheBackedEmbeddings], None, None]:
+#     if not os.environ.get("UIPATH_TESTS_CACHE_LLMGW"):
+#         yield None
+#     else:
+#         logging.info("Setting up embeddings cache")
+#         model = "text-embedding-3-large"
+#         embedder = CacheBackedEmbeddings.from_bytes_store(
+#             underlying_embeddings=UiPathOpenAIEmbeddings(model=model),
+#             document_embedding_cache=LocalFileStore(
+#                 test_cache_settings.cached_embeddings_dir
+#             ),
+#             namespace=model,
+#         )
+#         yield embedder
+#     return