-
Notifications
You must be signed in to change notification settings - Fork 2
Expand file tree
/
Copy pathtest_server.py
More file actions
150 lines (109 loc) · 4.8 KB
/
Copy pathtest_server.py
File metadata and controls
150 lines (109 loc) · 4.8 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
"""Tests for the server module."""
from datetime import datetime
from unittest.mock import MagicMock
def test_data_models_import_without_textual():
"""Data models should import without Textual dependency."""
from uipath.dev.models.data import ChatData, LogData, TraceData
log = LogData(run_id="r1", level="INFO", message="test", timestamp=datetime.now())
assert log.run_id == "r1"
trace = TraceData(
run_id="r1", span_name="agent", span_id="s1", timestamp=datetime.now()
)
assert trace.status == "running"
assert trace.parent_span_id is None
chat = ChatData(run_id="r1")
assert chat.event is None
def test_server_class_import():
"""UiPathDeveloperServer should be importable."""
from uipath.dev.server import UiPathDeveloperServer
assert UiPathDeveloperServer is not None
def test_serializer_log():
"""serialize_log should produce a valid dict."""
from uipath.dev.models.data import LogData
from uipath.dev.server.serializers import serialize_log
ts = datetime(2025, 1, 15, 10, 30, 0)
log = LogData(run_id="r1", level="WARNING", message="oops", timestamp=ts)
result = serialize_log(log)
assert result["run_id"] == "r1"
assert result["level"] == "WARNING"
assert result["message"] == "oops"
assert "2025-01-15" in result["timestamp"]
def test_serializer_trace():
"""serialize_trace should produce a valid dict."""
from uipath.dev.models.data import TraceData
from uipath.dev.server.serializers import serialize_trace
trace = TraceData(
run_id="r1",
span_name="llm_call",
span_id="s1",
parent_span_id="s0",
trace_id="t1",
status="completed",
duration_ms=123.4,
timestamp=datetime.now(),
attributes={"model": "gpt-4"},
)
result = serialize_trace(trace)
assert result["span_name"] == "llm_call"
assert result["parent_span_id"] == "s0"
assert result["duration_ms"] == 123.4
assert result["attributes"]["model"] == "gpt-4"
def test_serializer_chat():
"""serialize_chat should omit event/message when None."""
from uipath.dev.models.data import ChatData
from uipath.dev.server.serializers import serialize_chat
chat = ChatData(run_id="r1")
result = serialize_chat(chat)
assert result["run_id"] == "r1"
assert "event" not in result
assert "message" not in result
def test_ws_protocol_enums():
"""WebSocket protocol enums should have correct values."""
from uipath.dev.server.ws.protocol import ClientCommand, ServerEvent
assert ServerEvent.RUN_UPDATED.value == "run.updated"
assert ServerEvent.LOG.value == "log"
assert ServerEvent.TRACE.value == "trace"
assert ServerEvent.CHAT.value == "chat"
assert ClientCommand.SUBSCRIBE.value == "subscribe"
assert ClientCommand.UNSUBSCRIBE.value == "unsubscribe"
assert ClientCommand.CHAT_MESSAGE.value == "chat.message"
assert ClientCommand.DEBUG_STEP.value == "debug.step"
assert ClientCommand.DEBUG_CONTINUE.value == "debug.continue"
assert ClientCommand.DEBUG_STOP.value == "debug.stop"
def test_connection_manager_init():
"""ConnectionManager should initialize with empty state."""
from uipath.dev.server.ws.manager import ConnectionManager
mgr = ConnectionManager()
assert mgr is not None
def test_app_factory_creates_fastapi():
"""create_app should return a FastAPI app with routes."""
from uipath.dev.server.app import create_app
mock_server = MagicMock()
mock_server.run_service = MagicMock()
mock_server.connection_manager = MagicMock()
mock_server.runtime_factory = MagicMock()
mock_server.runtime_factory.list_entrypoints.return_value = []
app = create_app(mock_server)
assert app.title == "UiPath Developer Server"
route_paths = [getattr(r, "path", "") for r in app.routes]
assert "/api/entrypoints" in route_paths
assert "/api/runs" in route_paths
def test_app_has_fallback_when_no_static():
"""App should serve fallback HTML when static files don't exist."""
from uipath.dev.server.app import create_app
mock_server = MagicMock()
mock_server.run_service = MagicMock()
mock_server.connection_manager = MagicMock()
mock_server.runtime_factory = MagicMock()
mock_server.runtime_factory.list_entrypoints.return_value = []
app = create_app(mock_server)
# Should have a "/" route (either static mount or fallback)
route_paths = [getattr(r, "path", "") for r in app.routes]
assert "/" in route_paths or any(
getattr(r, "path", "").startswith("/") for r in app.routes
)
def test_frontend_build_module():
"""Frontend build functions should be importable."""
from uipath.dev.server.frontend_build import ensure_frontend_built, needs_build
assert callable(needs_build)
assert callable(ensure_frontend_built)