Add integration test for upserting graph in StateManager

integration-tests/test_fan_out.py

@@ -0,0 +1,94 @@
+import pytest
+import asyncio
+import threading
+import os
+from aiohttp import ClientSession
+from pydantic import BaseModel
+from exospherehost import BaseNode, Runtime, StateManager, GraphNodeModel
+
+@pytest.mark.asyncio
+async def test_upsert_graph(running_server):
+
+ class Node1(BaseNode):
+ class Inputs(BaseModel):
+ message: str
+
+ class Outputs(BaseModel):
+ count: str
+
+ async def execute(self):
+ return [self.Outputs(count=str(i)) for i in range(10)]
+
+
+ state_machine_url = running_server.base_url
+ runtime = Runtime(
+ namespace="test",
+ name="test",
+ nodes=[
+ Node1
+ ],
+ state_manager_uri=state_machine_url,
+ )
+
+ thread = threading.Thread(target=runtime.start, daemon=True)
+ thread.start()
+
+ await asyncio.sleep(2)
+
+ state_manager = StateManager(
+ namespace="test",
+ state_manager_uri=state_machine_url,
+ )
+
+ data = await state_manager.upsert_graph(
+ graph_name="test_graph",
+ graph_nodes=[
+ GraphNodeModel(
+ node_name="Node1",
+ namespace="test",
+ identifier="node1",
+ inputs={
+ "message": "Hello, world!",
+ },
+ )
+ ],
+ secrets={},
+ )
+
+ assert data is not None
+ assert data["validation_status"] == "VALID"
+
+ trigger = await state_manager.trigger(
+ graph_name="test_graph",
+ inputs={},
+ )
+
+ assert trigger is not None
+ assert "run_id" in trigger
+ run_id = trigger["run_id"]
+
+ await asyncio.sleep(30)
+
+ run_object = None
+
+ async with ClientSession() as session:
+ async with session.get(f"{state_machine_url}/v0/namespace/test/runs/1/100", headers={"x-api-key": os.getenv("EXOSPHERE_API_KEY")}) as response: # type: ignore
+ assert response.status == 200
+ data = await response.json()
+
+ assert "runs" in data
+
+ for run in data["runs"]:
+ if run["run_id"] == run_id:
+ run_object = run
+ break
+
+ assert run_object is not None
+ assert run_object["run_id"] == run_id
+ assert run_object["graph_name"] == "test_graph"
+ assert run_object["success_count"] == 10
+ assert run_object["pending_count"] == 0
+ assert run_object["errored_count"] == 0
+ assert run_object["retried_count"] == 0
+ assert run_object["total_count"] == 10
+ assert run_object["status"] == "SUCCESS"

integration-tests/test_upsert_graph.py

@@ -0,0 +1,54 @@
+import pytest
+import asyncio
+import threading
+from pydantic import BaseModel
+from exospherehost import BaseNode, Runtime, StateManager, GraphNodeModel
+
+@pytest.mark.asyncio
+async def test_upsert_graph(running_server):
+
+ class PrintNode(BaseNode):
+ class Inputs(BaseModel):
+ message: str
+
+ async def execute(self):
+ print(self.inputs.message) # type: ignore
+
+
+ state_machine_url = running_server.base_url
+ runtime = Runtime(
+ namespace="test",
+ name="test",
+ nodes=[
+ PrintNode
+ ],
+ state_manager_uri=state_machine_url,
+ )
+
+ thread = threading.Thread(target=runtime.start, daemon=True)
+ thread.start()
+
+ await asyncio.sleep(2)
+
+ state_manager = StateManager(
+ namespace="test",
+ state_manager_uri=state_machine_url,
+ )
+
+ data = await state_manager.upsert_graph(
+ graph_name="test_graph",
+ graph_nodes=[
+ GraphNodeModel(
+ node_name="PrintNode",
+ namespace="test",
+ identifier="node1",
+ inputs={
+ "message": "Hello, world!",
+ },
+ )
+ ],
+ secrets={},
+ )
+
+ assert data is not None
+ assert data["validation_status"] == "VALID"