[WIP] Add better tests

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,59 @@
+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
+
+
+# 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"]
+
+    # 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---")
+
+    return {"graph_state": state["graph_state"] + " I am"}
+
+
+def node_2(state):
+    print("---Node 2---")
+    return {"graph_state": state["graph_state"] + " happy!"}
+
+
+def node_3(state):
+    print("---Node 3---")
+    return {"graph_state": state["graph_state"] + " sad!"}
+
+
+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", END)
+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,23 @@
+import os
+import sys
+
+from dotenv import load_dotenv
+
+from uipath_langchain._cli.cli_run import langgraph_run_middleware
+
+load_dotenv()
+
+
+def test_dummy():
+    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)
+    result = langgraph_run_middleware(
+        entrypoint=None,
+        input='{ "graph_state": "GET Assets API does not enforce proper permissions Assets.View" }',
+        resume=False,
+    )
+
+    assert result.error_message is None

tests/conftest.py

@@ -1,95 +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")
-        if url:
-            return "/".join(url.split("/", 3)[:3])
-    except Exception:
-        return None
-    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
+# 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