test_http_client.py 2.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103
  1. """HttpClient 工具层的单元测试"""
  2. from unittest.mock import Mock, patch
  3. import pytest
  4. from src.tools.http_client import HttpClient
  5. # --- _parse_body ---
  6. def test_parse_body_json():
  7. resp = Mock()
  8. resp.json.return_value = {"a": 1}
  9. assert HttpClient()._parse_body(resp) == {"a": 1}
  10. def test_parse_body_fallback_to_text():
  11. resp = Mock()
  12. resp.json.side_effect = ValueError("not json")
  13. resp.text = "plain text"
  14. assert HttpClient()._parse_body(resp) == "plain text"
  15. # --- _error_result ---
  16. def test_error_result_shape():
  17. r = HttpClient()._error_result("boom")
  18. assert r == {"success": False, "status_code": None, "body": None, "elapsed": 0.0, "error": "boom"}
  19. # --- request ---
  20. def test_request_unsupported_method():
  21. r = HttpClient().request("TRACE", "http://x")
  22. assert r["success"] is False
  23. assert "不支持" in r["error"]
  24. def test_request_success():
  25. resp = Mock()
  26. resp.status_code = 200
  27. resp.json.return_value = {"ok": True}
  28. with patch("requests.request", return_value=resp):
  29. r = HttpClient(max_retries=0).request("GET", "http://x")
  30. assert r["success"] is True
  31. assert r["status_code"] == 200
  32. assert r["body"] == {"ok": True}
  33. def test_request_multipart_uses_files_not_json():
  34. resp = Mock()
  35. resp.status_code = 201
  36. resp.json.return_value = {"id": "img1"}
  37. files = {"file": ("a.png", b"pngbytes", "image/png")}
  38. with patch("requests.request", return_value=resp) as mock_req:
  39. r = HttpClient(max_retries=0).request(
  40. "POST", "http://x", body={"desc": "hi"}, files=files
  41. )
  42. assert r["success"] is True
  43. # multipart 场景:走 data + files,而不是 json=
  44. kwargs = mock_req.call_args.kwargs
  45. assert "json" not in kwargs
  46. assert kwargs["files"] == files
  47. assert kwargs["data"] == {"desc": "hi"}
  48. def test_request_multipart_no_file_uses_form_data():
  49. resp = Mock()
  50. resp.status_code = 200
  51. resp.json.return_value = {}
  52. with patch("requests.request", return_value=resp) as mock_req:
  53. HttpClient(max_retries=0).request(
  54. "POST", "http://x", body={"k": "v"}, content_type="multipart/form-data"
  55. )
  56. kwargs = mock_req.call_args.kwargs
  57. assert "json" not in kwargs
  58. assert kwargs["data"] == {"k": "v"}
  59. def test_request_retry_then_success(monkeypatch):
  60. import requests.exceptions
  61. # 去掉重试之间的 1 秒 sleep,加快测试
  62. monkeypatch.setattr("src.tools.http_client.time.sleep", lambda s: None)
  63. resp = Mock()
  64. resp.status_code = 200
  65. resp.json.return_value = {}
  66. with patch(
  67. "requests.request",
  68. side_effect=[requests.exceptions.ConnectionError("网络断开"), resp],
  69. ):
  70. r = HttpClient(max_retries=1).request("GET", "http://x")
  71. assert r["success"] is True
  72. assert r["status_code"] == 200