test_executor_agent.py 2.7 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182
  1. """ExecutorAgent 路径参数替换与 URL 拼接的单元测试"""
  2. from src.agents.executor_agent import ExecutorAgent
  3. def _executor():
  4. return ExecutorAgent()
  5. # --- _replace_path_params ---
  6. def test_replace_path_params_basic():
  7. path, params = _executor()._replace_path_params("/status/{codes}", {"codes": "200"})
  8. assert path == "/status/200"
  9. assert params == {}
  10. def test_replace_path_params_keeps_query_params():
  11. path, params = _executor()._replace_path_params("/users/{id}", {"id": 1, "page": 2})
  12. assert path == "/users/1"
  13. assert params == {"page": 2}
  14. def test_replace_path_params_missing_value_keeps_placeholder():
  15. # error 用例缺少参数时,占位符应保留原样
  16. path, params = _executor()._replace_path_params("/status/{codes}", {})
  17. assert path == "/status/{codes}"
  18. assert params == {}
  19. def test_replace_path_params_multiple():
  20. path, _ = _executor()._replace_path_params("/x/{a}/y/{b}", {"a": 1, "b": 2})
  21. assert path == "/x/1/y/2"
  22. # --- execute_one:URL 拼接 ---
  23. def test_execute_one_url_join_and_param_removal():
  24. class FakeClient:
  25. def request(self, method, url, headers=None, params=None, body=None,
  26. files=None, content_type="application/json"):
  27. return {"method": method, "url": url, "params": params, "body": body}
  28. ex = _executor()
  29. ex.http_client = FakeClient()
  30. case = {"path": "/status/{codes}", "method": "GET", "params": {"codes": "200"}}
  31. out = ex.execute_one(case, "https://httpbin.org/")
  32. assert out["case"] is case
  33. assert out["result"]["url"] == "https://httpbin.org/status/200"
  34. assert out["result"]["params"] == {}
  35. assert out["result"]["method"] == "GET"
  36. def test_execute_one_passes_headers_and_files():
  37. """执行时全局认证头与 multipart 文件要原样透传给工具"""
  38. captured = {}
  39. class FakeClient:
  40. def request(self, method, url, headers=None, params=None, body=None,
  41. files=None, content_type="application/json"):
  42. captured["headers"] = headers
  43. captured["files"] = files
  44. captured["content_type"] = content_type
  45. return {"method": method, "url": url}
  46. ex = _executor()
  47. ex.http_client = FakeClient()
  48. case = {
  49. "path": "/api/files/images",
  50. "method": "POST",
  51. "params": {},
  52. "body": {},
  53. "files": {"file": ("a.png", b"x", "image/png")},
  54. "content_type": "multipart/form-data",
  55. }
  56. ex.execute_one(case, "http://x", headers={"Authorization": "Bearer token"})
  57. assert captured["headers"] == {"Authorization": "Bearer token"}
  58. assert captured["files"] == {"file": ("a.png", b"x", "image/png")}
  59. assert captured["content_type"] == "multipart/form-data"