张翔
246 lines
9.2 KiB
Python
246 lines
9.2 KiB
Python
"""测试API客户端"""
|
|
|
|
import pytest
|
|
import time
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
import requests
|
|
from apitest.client.api_client import APIClient
|
|
from apitest.models.exceptions import RequestException
|
|
from apitest.models.test_models import HTTPMethod, PerformanceMetrics
|
|
|
|
|
|
class TestAPIClient:
|
|
"""测试APIClient类"""
|
|
|
|
def test_init(self):
|
|
"""测试初始化"""
|
|
client = APIClient("http://localhost:8080", timeout=5000, max_retries=3)
|
|
assert client.base_url == "http://localhost:8080"
|
|
assert client.timeout == 5.0
|
|
assert client.max_retries == 3
|
|
assert "Content-Type" in client._default_headers
|
|
assert "Accept" in client._default_headers
|
|
|
|
def test_init_with_logger(self):
|
|
"""测试带日志记录器的初始化"""
|
|
logger = Mock()
|
|
client = APIClient("http://localhost:8080", logger=logger)
|
|
assert client.logger == logger
|
|
|
|
def test_set_default_headers(self):
|
|
"""测试设置默认请求头"""
|
|
client = APIClient("http://localhost:8080")
|
|
client.set_default_headers({"X-Custom-Header": "value"})
|
|
assert "X-Custom-Header" in client._default_headers
|
|
assert client._default_headers["X-Custom-Header"] == "value"
|
|
|
|
def test_set_auth_token(self):
|
|
"""测试设置认证token"""
|
|
client = APIClient("http://localhost:8080")
|
|
client.set_auth_token("test-token")
|
|
assert "Authorization" in client._default_headers
|
|
assert client._default_headers["Authorization"] == "Bearer test-token"
|
|
|
|
def test_build_url(self):
|
|
"""测试构建URL"""
|
|
client = APIClient("http://localhost:8080")
|
|
url = client._build_url("/api/test")
|
|
assert url == "http://localhost:8080/api/test"
|
|
|
|
def test_build_url_with_leading_slash(self):
|
|
"""测试构建URL（带前导斜杠）"""
|
|
client = APIClient("http://localhost:8080")
|
|
url = client._build_url("api/test")
|
|
assert url == "http://localhost:8080/api/test"
|
|
|
|
def test_merge_headers(self):
|
|
"""测试合并请求头"""
|
|
client = APIClient("http://localhost:8080")
|
|
client.set_default_headers({"X-Default": "default"})
|
|
merged = client._merge_headers({"X-Custom": "custom"})
|
|
assert merged["X-Default"] == "default"
|
|
assert merged["X-Custom"] == "custom"
|
|
|
|
def test_merge_headers_override(self):
|
|
"""测试合并请求头（覆盖）"""
|
|
client = APIClient("http://localhost:8080")
|
|
client.set_default_headers({"Content-Type": "application/xml"})
|
|
merged = client._merge_headers({"Content-Type": "application/json"})
|
|
assert merged["Content-Type"] == "application/json"
|
|
|
|
@patch('requests.Session.get')
|
|
def test_request_success(self, mock_get):
|
|
"""测试成功请求"""
|
|
mock_response = Mock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"success": True}
|
|
mock_response.headers = {"Content-Type": "application/json"}
|
|
mock_get.return_value = mock_response
|
|
|
|
client = APIClient("http://localhost:8080")
|
|
result = client.request(HTTPMethod.GET, "/api/test")
|
|
|
|
assert result["status_code"] == 200
|
|
assert result["response_body"] == {"success": True}
|
|
assert "performance" in result
|
|
|
|
@patch('requests.Session.get')
|
|
def test_request_with_params(self, mock_get):
|
|
"""测试带参数的请求"""
|
|
mock_response = Mock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"success": True}
|
|
mock_response.headers = {"Content-Type": "application/json"}
|
|
mock_get.return_value = mock_response
|
|
|
|
client = APIClient("http://localhost:8080")
|
|
result = client.request(HTTPMethod.GET, "/api/test", params={"page": 1, "size": 10})
|
|
|
|
assert result["status_code"] == 200
|
|
assert result["response_body"] == {"success": True}
|
|
mock_get.assert_called_once()
|
|
|
|
@patch('requests.Session.post')
|
|
def test_request_with_body(self, mock_post):
|
|
"""测试带请求体的请求"""
|
|
mock_response = Mock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"success": True}
|
|
mock_response.headers = {"Content-Type": "application/json"}
|
|
mock_post.return_value = mock_response
|
|
|
|
client = APIClient("http://localhost:8080")
|
|
result = client.request(HTTPMethod.POST, "/api/test", body={"name": "test"})
|
|
|
|
assert result["status_code"] == 200
|
|
assert result["response_body"] == {"success": True}
|
|
mock_post.assert_called_once()
|
|
|
|
@patch('requests.Session.get')
|
|
def test_request_with_headers(self, mock_get):
|
|
"""测试带自定义请求头的请求"""
|
|
mock_response = Mock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"success": True}
|
|
mock_response.headers = {"Content-Type": "application/json"}
|
|
mock_get.return_value = mock_response
|
|
|
|
client = APIClient("http://localhost:8080")
|
|
result = client.request(HTTPMethod.GET, "/api/test", headers={"X-Custom": "value"})
|
|
|
|
assert result["status_code"] == 200
|
|
assert result["response_body"] == {"success": True}
|
|
mock_get.assert_called_once()
|
|
|
|
@patch('requests.Session.get')
|
|
def test_request_retry_on_timeout(self, mock_get):
|
|
"""测试超时重试"""
|
|
mock_response = Mock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"success": True}
|
|
mock_response.headers = {"Content-Type": "application/json"}
|
|
|
|
mock_get.side_effect = [
|
|
requests.Timeout(),
|
|
requests.Timeout(),
|
|
mock_response
|
|
]
|
|
|
|
client = APIClient("http://localhost:8080", max_retries=3)
|
|
result = client.request(HTTPMethod.GET, "/api/test")
|
|
|
|
assert result["status_code"] == 200
|
|
assert result["response_body"] == {"success": True}
|
|
assert mock_get.call_count == 3
|
|
|
|
@patch('requests.Session.get')
|
|
def test_request_failure_after_retries(self, mock_get):
|
|
"""测试重试后仍然失败"""
|
|
mock_get.side_effect = requests.Timeout()
|
|
|
|
client = APIClient("http://localhost:8080", max_retries=2)
|
|
|
|
with pytest.raises(RequestException):
|
|
client.request(HTTPMethod.GET, "/api/test")
|
|
|
|
@patch('requests.Session.get')
|
|
def test_request_invalid_json(self, mock_get):
|
|
"""测试无效JSON响应"""
|
|
mock_response = Mock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.side_effect = ValueError("Invalid JSON")
|
|
mock_response.text = "plain text"
|
|
mock_response.headers = {"Content-Type": "text/plain"}
|
|
mock_get.return_value = mock_response
|
|
|
|
client = APIClient("http://localhost:8080")
|
|
result = client.request(HTTPMethod.GET, "/api/test")
|
|
|
|
assert result["status_code"] == 200
|
|
assert result["response_body"] == "plain text"
|
|
|
|
@patch('requests.Session.get')
|
|
def test_get(self, mock_get):
|
|
"""测试GET请求"""
|
|
mock_response = Mock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"success": True}
|
|
mock_response.headers = {"Content-Type": "application/json"}
|
|
mock_get.return_value = mock_response
|
|
|
|
client = APIClient("http://localhost:8080")
|
|
result = client.get("/api/test")
|
|
|
|
assert result["status_code"] == 200
|
|
assert result["response_body"] == {"success": True}
|
|
|
|
@patch('requests.Session.post')
|
|
def test_post(self, mock_post):
|
|
"""测试POST请求"""
|
|
mock_response = Mock()
|
|
mock_response.status_code = 201
|
|
mock_response.json.return_value = {"success": True}
|
|
mock_response.headers = {"Content-Type": "application/json"}
|
|
mock_post.return_value = mock_response
|
|
|
|
client = APIClient("http://localhost:8080")
|
|
result = client.post("/api/test", {"name": "test"})
|
|
|
|
assert result["status_code"] == 201
|
|
assert result["response_body"] == {"success": True}
|
|
|
|
@patch('requests.Session.put')
|
|
def test_put(self, mock_put):
|
|
"""测试PUT请求"""
|
|
mock_response = Mock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"success": True}
|
|
mock_response.headers = {"Content-Type": "application/json"}
|
|
mock_put.return_value = mock_response
|
|
|
|
client = APIClient("http://localhost:8080")
|
|
result = client.put("/api/test/1", {"name": "updated"})
|
|
|
|
assert result["status_code"] == 200
|
|
assert result["response_body"] == {"success": True}
|
|
|
|
@patch('requests.Session.delete')
|
|
def test_delete(self, mock_delete):
|
|
"""测试DELETE请求"""
|
|
mock_response = Mock()
|
|
mock_response.status_code = 204
|
|
mock_response.headers = {"Content-Type": "application/json"}
|
|
mock_delete.return_value = mock_response
|
|
|
|
client = APIClient("http://localhost:8080")
|
|
result = client.delete("/api/test/1")
|
|
|
|
assert result["status_code"] == 204
|
|
|
|
def test_close(self):
|
|
"""测试关闭客户端"""
|
|
client = APIClient("http://localhost:8080")
|
|
session = client._session
|
|
client.close()
|
|
assert client._session == session
|