Coverage for agentos/tests/test_a2a.py: 0%

1"""测试 A2A 协议 — Task, Message, Handoff, Client, Server。""" 

2 

3import time 

4import pytest 

5from agentos.protocols.a2a import ( 

6 TaskState, MessageRole, 

7 TextPart, FilePart, DataPart, part_from_dict, 

8 A2AArtifact, 

9 A2AMessage, 

10 A2ATask, 

11 A2AHandoff, 

12 A2ASession, 

13 A2AServer, 

14 new_task, new_handoff, 

15) 

16 

17 

18class TestA2AParts: 

19 def test_text_part_roundtrip(self): 

20 tp = TextPart(text="hello", meta={"lang": "en"}) 

21 d = tp.to_dict() 

22 assert d["type"] == "text" 

23 tp2 = TextPart.from_dict(d) 

24 assert tp2.text == "hello" 

25 assert tp2.meta == {"lang": "en"} 

26 

27 def test_file_part_roundtrip(self): 

28 fp = FilePart(url="https://ex.com/f.pdf", filename="report.pdf", mime_type="application/pdf", size=1024) 

29 d = fp.to_dict() 

30 fp2 = FilePart.from_dict(d) 

31 assert fp2.filename == "report.pdf" 

32 assert fp2.mime_type == "application/pdf" 

33 

34 def test_data_part_roundtrip(self): 

35 dp = DataPart(data={"count": 42}, schema_uri="https://schema.org/result") 

36 d = dp.to_dict() 

37 dp2 = DataPart.from_dict(d) 

38 assert dp2.data["count"] == 42 

39 

40 def test_part_from_dict_dispatcher(self): 

41 d = {"type": "text", "text": "hi"} 

42 p = part_from_dict(d) 

43 assert isinstance(p, TextPart) 

44 assert p.text == "hi" 

45 

46 d = {"type": "file", "filename": "x.txt"} 

47 p = part_from_dict(d) 

48 assert isinstance(p, FilePart) 

49 

50 d = {"type": "data", "data": {"a": 1}} 

51 p = part_from_dict(d) 

52 assert isinstance(p, DataPart) 

53 

54 

55class TestA2AArtifact: 

56 def test_roundtrip(self): 

57 art = A2AArtifact(name="result.json", mime_type="application/json", blob=b'{"a":1}', size=8) 

58 d = art.to_dict() 

59 art2 = A2AArtifact.from_dict(d) 

60 assert art2.name == "result.json" 

61 assert art2.blob == b'{"a":1}' 

62 

63 def test_url_artifact(self): 

64 art = A2AArtifact(name="image.png", url="https://cdn.ex/img.png") 

65 d = art.to_dict() 

66 assert "url" in d 

67 art2 = A2AArtifact.from_dict(d) 

68 assert art2.url == "https://cdn.ex/img.png" 

69 

70 

71class TestA2AMessage: 

72 def test_user_text(self): 

73 msg = A2AMessage.user_text("hello world") 

74 assert msg.role == MessageRole.USER 

75 assert len(msg.parts) == 1 

76 assert msg.parts[0].text == "hello world" 

77 

78 def test_agent_text(self): 

79 msg = A2AMessage.agent_text("done") 

80 assert msg.role == MessageRole.AGENT 

81 assert msg.get_text() == "done" 

82 

83 def test_multipart_roundtrip(self): 

84 msg = A2AMessage( 

85 role=MessageRole.USER, 

86 parts=[ 

87 TextPart(text="analyze this"), 

88 FilePart(filename="data.csv"), 

89 DataPart(data={"options": {"method": "pca"}}), 

90 ], 

91 ) 

92 d = msg.to_dict() 

93 msg2 = A2AMessage.from_dict(d) 

94 assert msg2.role == MessageRole.USER 

95 assert len(msg2.parts) == 3 

96 assert isinstance(msg2.parts[0], TextPart) 

97 assert isinstance(msg2.parts[1], FilePart) 

98 assert isinstance(msg2.parts[2], DataPart) 

99 assert msg2.get_text() == "analyze this" 

100 

101 

102class TestA2ATask: 

103 def test_lifecycle(self): 

104 task = A2ATask(input=A2AMessage.user_text("do something")) 

105 assert task.state == TaskState.SUBMITTED 

106 

107 task.start_working() 

108 assert task.state == TaskState.WORKING 

109 

110 task.complete(A2AMessage.agent_text("done")) 

111 assert task.state == TaskState.COMPLETED 

112 assert task.output.get_text() == "done" 

113 assert task.is_terminal() 

114 

115 def test_fail(self): 

116 task = A2ATask(input=A2AMessage.user_text("bad")) 

117 task.start_working() 

118 task.fail("something went wrong") 

119 assert task.state == TaskState.FAILED 

120 assert task.error == "something went wrong" 

121 assert task.is_terminal() 

122 

123 def test_cancel(self): 

124 task = A2ATask() 

125 assert not task.is_terminal() 

126 task.cancel() 

127 assert task.state == TaskState.CANCELLED 

128 assert task.is_terminal() 

129 

130 def test_cannot_start_non_submitted(self): 

131 task = A2ATask() 

132 task.start_working() 

133 with pytest.raises(ValueError): 

134 task.start_working() 

135 

136 def test_cannot_complete_non_working(self): 

137 task = A2ATask() 

138 with pytest.raises(ValueError): 

139 task.complete() 

140 

141 def test_cannot_cancel_completed(self): 

142 task = A2ATask() 

143 task.start_working() 

144 task.complete() 

145 with pytest.raises(ValueError): 

146 task.cancel() 

147 

148 def test_artifact_attachment(self): 

149 task = A2ATask() 

150 task.add_artifact(A2AArtifact(name="out.csv")) 

151 task.add_artifact(A2AArtifact(name="out.png")) 

152 assert len(task.artifacts) == 2 

153 

154 def test_json_roundtrip(self): 

155 task = A2ATask(input=A2AMessage.user_text("hello")) 

156 task.start_working() 

157 task.complete(A2AMessage.agent_text("result")) 

158 task.add_artifact(A2AArtifact(name="out.json", blob=b"{}")) 

159 

160 json_str = task.to_json() 

161 task2 = A2ATask.from_json(json_str) 

162 assert task2.task_id == task.task_id 

163 assert task2.state == TaskState.COMPLETED 

164 assert task2.input.get_text() == "hello" 

165 assert task2.artifacts[0].name == "out.json" 

166 

167 def test_state_history(self): 

168 task = A2ATask() 

169 task.start_working() 

170 task.complete() 

171 assert len(task._state_history) == 2 

172 assert task._state_history[0][0] == TaskState.SUBMITTED 

173 assert task._state_history[1][0] == TaskState.WORKING 

174 

175 

176class TestA2AHandoff: 

177 def test_roundtrip(self): 

178 task = A2ATask(input=A2AMessage.user_text("do x")) 

179 ho = A2AHandoff( 

180 source_agent="coordinator", 

181 target_agent="worker", 

182 task=task, 

183 reason="delegation", 

184 ) 

185 d = ho.to_dict() 

186 ho2 = A2AHandoff.from_dict(d) 

187 assert ho2.source_agent == "coordinator" 

188 assert ho2.target_agent == "worker" 

189 assert ho2.task.task_id == task.task_id 

190 assert ho2.reason == "delegation" 

191 

192 def test_json_roundtrip(self): 

193 task = A2ATask(input=A2AMessage.user_text("test")) 

194 ho = A2AHandoff(source_agent="a", target_agent="b", task=task) 

195 ho2 = A2AHandoff.from_json(ho.to_json()) 

196 assert ho2.source_agent == "a" 

197 assert ho2.handoff_id == ho.handoff_id 

198 

199 

200class TestA2ASession: 

201 def test_basic(self): 

202 sess = A2ASession() 

203 sess.add_message(A2AMessage.user_text("hi")) 

204 sess.add_message(A2AMessage.agent_text("hello")) 

205 sess.add_task(A2ATask()) 

206 assert len(sess.history) == 2 

207 assert len(sess.tasks) == 1 

208 

209 def test_get_last_n(self): 

210 sess = A2ASession() 

211 for i in range(5): 

212 sess.add_message(A2AMessage.user_text(f"msg{i}")) 

213 last3 = sess.get_last_n_messages(3) 

214 assert len(last3) == 3 

215 assert last3[-1].get_text() == "msg4" 

216 

217 

218class TestA2Server: 

219 @pytest.mark.asyncio 

220 async def test_process_task_success(self): 

221 server = A2AServer() 

222 async def handler(task: A2ATask): 

223 return A2AMessage.agent_text(f"processed: {task.input.get_text()}") 

224 

225 server.register_handler("worker", handler) 

226 task = new_task("hello test", target_agent="worker") 

227 result = await server.process_task(task.to_dict()) 

228 assert result["state"] == "completed" 

229 assert "processed: hello test" in result["output"]["parts"][0]["text"] 

230 

231 @pytest.mark.asyncio 

232 async def test_process_task_no_handler(self): 

233 server = A2AServer() 

234 task = new_task("hello", target_agent="nonexistent") 

235 result = await server.process_task(task.to_dict()) 

236 assert result["state"] == "failed" 

237 assert "No handler" in result["error"] 

238 

239 @pytest.mark.asyncio 

240 async def test_process_task_handler_error(self): 

241 server = A2AServer() 

242 async def bad_handler(task): 

243 raise ValueError("simulated error") 

244 

245 server.register_handler("bad", bad_handler) 

246 task = new_task("test", target_agent="bad") 

247 result = await server.process_task(task.to_dict()) 

248 assert result["state"] == "failed" 

249 assert "simulated error" in result["error"] 

250 

251 def test_get_task(self): 

252 server = A2AServer() 

253 task = A2ATask(task_id="task-001") 

254 server._tasks["task-001"] = task 

255 assert server.get_task("task-001").task_id == "task-001" 

256 assert server.get_task("nonexistent") is None 

257 

258 def test_list_tasks_by_state(self): 

259 server = A2AServer() 

260 t1 = A2ATask(task_id="t1") 

261 t2 = A2ATask(task_id="t2"); t2.start_working(); t2.complete() 

262 t3 = A2ATask(task_id="t3"); t3.fail("err") 

263 server._tasks = {"t1": t1, "t2": t2, "t3": t3} 

264 assert len(server.list_tasks()) == 3 

265 assert len(server.list_tasks(TaskState.COMPLETED)) == 1 

266 assert len(server.list_tasks(TaskState.FAILED)) == 1 

267 

268 def test_cleanup(self): 

269 server = A2AServer() 

270 old = A2ATask(task_id="old") 

271 old.start_working(); old.complete() 

272 old._updated = time.time() - 4000 # fake old 

273 fresh = A2ATask(task_id="fresh") 

274 server._tasks = {"old": old, "fresh": fresh} 

275 n = server.cleanup_old(max_age_seconds=3600) 

276 assert n == 1 

277 assert "old" not in server._tasks 

278 assert "fresh" in server._tasks 

279 

280 

281class TestConvenience: 

282 def test_new_task(self): 

283 t = new_task("my task", target_agent="worker", priority="high") 

284 assert t.input.get_text() == "my task" 

285 assert t.meta["target_agent"] == "worker" 

286 assert t.meta["priority"] == "high" 

287 

288 def test_new_handoff(self): 

289 t = new_task("delegate me") 

290 ho = new_handoff(t, source="a", target="b", reason="overload") 

291 assert ho.source_agent == "a" 

292 assert ho.target_agent == "b" 

293 assert ho.reason == "overload"