| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103 |
- """HttpClient 工具层的单元测试"""
- from unittest.mock import Mock, patch
- import pytest
- from src.tools.http_client import HttpClient
- # --- _parse_body ---
- def test_parse_body_json():
- resp = Mock()
- resp.json.return_value = {"a": 1}
- assert HttpClient()._parse_body(resp) == {"a": 1}
- def test_parse_body_fallback_to_text():
- resp = Mock()
- resp.json.side_effect = ValueError("not json")
- resp.text = "plain text"
- assert HttpClient()._parse_body(resp) == "plain text"
- # --- _error_result ---
- def test_error_result_shape():
- r = HttpClient()._error_result("boom")
- assert r == {"success": False, "status_code": None, "body": None, "elapsed": 0.0, "error": "boom"}
- # --- request ---
- def test_request_unsupported_method():
- r = HttpClient().request("TRACE", "http://x")
- assert r["success"] is False
- assert "不支持" in r["error"]
- def test_request_success():
- resp = Mock()
- resp.status_code = 200
- resp.json.return_value = {"ok": True}
- with patch("requests.request", return_value=resp):
- r = HttpClient(max_retries=0).request("GET", "http://x")
- assert r["success"] is True
- assert r["status_code"] == 200
- assert r["body"] == {"ok": True}
- def test_request_multipart_uses_files_not_json():
- resp = Mock()
- resp.status_code = 201
- resp.json.return_value = {"id": "img1"}
- files = {"file": ("a.png", b"pngbytes", "image/png")}
- with patch("requests.request", return_value=resp) as mock_req:
- r = HttpClient(max_retries=0).request(
- "POST", "http://x", body={"desc": "hi"}, files=files
- )
- assert r["success"] is True
- # multipart 场景:走 data + files,而不是 json=
- kwargs = mock_req.call_args.kwargs
- assert "json" not in kwargs
- assert kwargs["files"] == files
- assert kwargs["data"] == {"desc": "hi"}
- def test_request_multipart_no_file_uses_form_data():
- resp = Mock()
- resp.status_code = 200
- resp.json.return_value = {}
- with patch("requests.request", return_value=resp) as mock_req:
- HttpClient(max_retries=0).request(
- "POST", "http://x", body={"k": "v"}, content_type="multipart/form-data"
- )
- kwargs = mock_req.call_args.kwargs
- assert "json" not in kwargs
- assert kwargs["data"] == {"k": "v"}
- def test_request_retry_then_success(monkeypatch):
- import requests.exceptions
- # 去掉重试之间的 1 秒 sleep,加快测试
- monkeypatch.setattr("src.tools.http_client.time.sleep", lambda s: None)
- resp = Mock()
- resp.status_code = 200
- resp.json.return_value = {}
- with patch(
- "requests.request",
- side_effect=[requests.exceptions.ConnectionError("网络断开"), resp],
- ):
- r = HttpClient(max_retries=1).request("GET", "http://x")
- assert r["success"] is True
- assert r["status_code"] == 200
|