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

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

2 

3import time 

4 

5import pytest 

6 

7from agentos.protocols.a2a import ( 

8 A2AArtifact, 

9 A2AHandoff, 

10 A2AMessage, 

11 A2AServer, 

12 A2ASession, 

13 A2ATask, 

14 DataPart, 

15 FilePart, 

16 MessageRole, 

17 TaskState, 

18 TextPart, 

19 new_handoff, 

20 new_task, 

21 part_from_dict, 

22) 

23 

24 

25class TestA2AParts: 

26 def test_text_part_roundtrip(self): 

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

28 d = tp.to_dict() 

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

30 tp2 = TextPart.from_dict(d) 

31 assert tp2.text == "hello" 

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

33 

34 def test_file_part_roundtrip(self): 

35 fp = FilePart( 

36 url="https://ex.com/f.pdf", 

37 filename="report.pdf", 

38 mime_type="application/pdf", 

39 size=1024, 

40 ) 

41 d = fp.to_dict() 

42 fp2 = FilePart.from_dict(d) 

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

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

45 

46 def test_data_part_roundtrip(self): 

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

48 d = dp.to_dict() 

49 dp2 = DataPart.from_dict(d) 

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

51 

52 def test_part_from_dict_dispatcher(self): 

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

54 p = part_from_dict(d) 

55 assert isinstance(p, TextPart) 

56 assert p.text == "hi" 

57 

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

59 p = part_from_dict(d) 

60 assert isinstance(p, FilePart) 

61 

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

63 p = part_from_dict(d) 

64 assert isinstance(p, DataPart) 

65 

66 

67class TestA2AArtifact: 

68 def test_roundtrip(self): 

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

70 d = art.to_dict() 

71 art2 = A2AArtifact.from_dict(d) 

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

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

74 

75 def test_url_artifact(self): 

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

77 d = art.to_dict() 

78 assert "url" in d 

79 art2 = A2AArtifact.from_dict(d) 

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

81 

82 

83class TestA2AMessage: 

84 def test_user_text(self): 

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

86 assert msg.role == MessageRole.USER 

87 assert len(msg.parts) == 1 

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

89 

90 def test_agent_text(self): 

91 msg = A2AMessage.agent_text("done") 

92 assert msg.role == MessageRole.AGENT 

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

94 

95 def test_multipart_roundtrip(self): 

96 msg = A2AMessage( 

97 role=MessageRole.USER, 

98 parts=[ 

99 TextPart(text="analyze this"), 

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

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

102 ], 

103 ) 

104 d = msg.to_dict() 

105 msg2 = A2AMessage.from_dict(d) 

106 assert msg2.role == MessageRole.USER 

107 assert len(msg2.parts) == 3 

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

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

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

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

112 

113 

114class TestA2ATask: 

115 def test_lifecycle(self): 

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

117 assert task.state == TaskState.SUBMITTED 

118 

119 task.start_working() 

120 assert task.state == TaskState.WORKING 

121 

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

123 assert task.state == TaskState.COMPLETED 

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

125 assert task.is_terminal() 

126 

127 def test_fail(self): 

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

129 task.start_working() 

130 task.fail("something went wrong") 

131 assert task.state == TaskState.FAILED 

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

133 assert task.is_terminal() 

134 

135 def test_cancel(self): 

136 task = A2ATask() 

137 assert not task.is_terminal() 

138 task.cancel() 

139 assert task.state == TaskState.CANCELLED 

140 assert task.is_terminal() 

141 

142 def test_cannot_start_non_submitted(self): 

143 task = A2ATask() 

144 task.start_working() 

145 with pytest.raises(ValueError): 

146 task.start_working() 

147 

148 def test_cannot_complete_non_working(self): 

149 task = A2ATask() 

150 with pytest.raises(ValueError): 

151 task.complete() 

152 

153 def test_cannot_cancel_completed(self): 

154 task = A2ATask() 

155 task.start_working() 

156 task.complete() 

157 with pytest.raises(ValueError): 

158 task.cancel() 

159 

160 def test_artifact_attachment(self): 

161 task = A2ATask() 

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

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

164 assert len(task.artifacts) == 2 

165 

166 def test_json_roundtrip(self): 

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

168 task.start_working() 

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

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

171 

172 json_str = task.to_json() 

173 task2 = A2ATask.from_json(json_str) 

174 assert task2.task_id == task.task_id 

175 assert task2.state == TaskState.COMPLETED 

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

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

178 

179 def test_state_history(self): 

180 task = A2ATask() 

181 task.start_working() 

182 task.complete() 

183 assert len(task._state_history) == 2 

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

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

186 

187 

188class TestA2AHandoff: 

189 def test_roundtrip(self): 

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

191 ho = A2AHandoff( 

192 source_agent="coordinator", 

193 target_agent="worker", 

194 task=task, 

195 reason="delegation", 

196 ) 

197 d = ho.to_dict() 

198 ho2 = A2AHandoff.from_dict(d) 

199 assert ho2.source_agent == "coordinator" 

200 assert ho2.target_agent == "worker" 

201 assert ho2.task.task_id == task.task_id 

202 assert ho2.reason == "delegation" 

203 

204 def test_json_roundtrip(self): 

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

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

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

208 assert ho2.source_agent == "a" 

209 assert ho2.handoff_id == ho.handoff_id 

210 

211 

212class TestA2ASession: 

213 def test_basic(self): 

214 sess = A2ASession() 

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

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

217 sess.add_task(A2ATask()) 

218 assert len(sess.history) == 2 

219 assert len(sess.tasks) == 1 

220 

221 def test_get_last_n(self): 

222 sess = A2ASession() 

223 for i in range(5): 

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

225 last3 = sess.get_last_n_messages(3) 

226 assert len(last3) == 3 

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

228 

229 

230class TestA2Server: 

231 @pytest.mark.asyncio 

232 async def test_process_task_success(self): 

233 server = A2AServer() 

234 

235 async def handler(task: A2ATask): 

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

237 

238 server.register_handler("worker", handler) 

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

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

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

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

243 

244 @pytest.mark.asyncio 

245 async def test_process_task_no_handler(self): 

246 server = A2AServer() 

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

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

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

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

251 

252 @pytest.mark.asyncio 

253 async def test_process_task_handler_error(self): 

254 server = A2AServer() 

255 

256 async def bad_handler(task): 

257 raise ValueError("simulated error") 

258 

259 server.register_handler("bad", bad_handler) 

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

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

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

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

264 

265 def test_get_task(self): 

266 server = A2AServer() 

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

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

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

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

271 

272 def test_list_tasks_by_state(self): 

273 server = A2AServer() 

274 t1 = A2ATask(task_id="t1") 

275 t2 = A2ATask(task_id="t2") 

276 t2.start_working() 

277 t2.complete() 

278 t3 = A2ATask(task_id="t3") 

279 t3.fail("err") 

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

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

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

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

284 

285 def test_cleanup(self): 

286 server = A2AServer() 

287 old = A2ATask(task_id="old") 

288 old.start_working() 

289 old.complete() 

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

291 fresh = A2ATask(task_id="fresh") 

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

293 n = server.cleanup_old(max_age_seconds=3600) 

294 assert n == 1 

295 assert "old" not in server._tasks 

296 assert "fresh" in server._tasks 

297 

298 

299class TestConvenience: 

300 def test_new_task(self): 

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

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

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

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

305 

306 def test_new_handoff(self): 

307 t = new_task("delegate me") 

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

309 assert ho.source_agent == "a" 

310 assert ho.target_agent == "b" 

311 assert ho.reason == "overload"