Tests for requests handling
This commit is contained in:
Clemens Schwaighofer
@@ -18,11 +18,12 @@ class Caller:
|
||||
header: dict[str, str],
|
||||
verify: bool = True,
|
||||
timeout: int = 20,
|
||||
proxy: dict[str, str] | None = None
|
||||
proxy: dict[str, str] | None = None,
|
||||
ca_file: str | None = None
|
||||
):
|
||||
self.headers = header
|
||||
self.timeout: int = timeout
|
||||
self.cafile = "/Library/Application Support/Netskope/STAgent/data/nscacert.pem"
|
||||
self.cafile = ca_file
|
||||
self.verify = verify
|
||||
self.proxy = proxy
|
||||
|
||||
|
||||
3
tests/unit/requests_handling/__init__.py
Normal file
3
tests/unit/requests_handling/__init__.py
Normal file
@@ -0,0 +1,3 @@
|
||||
"""
|
||||
PyTest: requests_handling tests
|
||||
"""
|
||||
308
tests/unit/requests_handling/test_auth_helpers.py
Normal file
308
tests/unit/requests_handling/test_auth_helpers.py
Normal file
@@ -0,0 +1,308 @@
|
||||
"""
|
||||
PyTest: requests_handling/auth_helpers
|
||||
"""
|
||||
|
||||
from base64 import b64decode
|
||||
import pytest
|
||||
from corelibs.requests_handling.auth_helpers import basic_auth
|
||||
|
||||
|
||||
class TestBasicAuth:
|
||||
"""Tests for basic_auth function"""
|
||||
|
||||
def test_basic_credentials(self):
|
||||
"""Test basic auth with simple username and password"""
|
||||
result = basic_auth("user", "pass")
|
||||
assert result.startswith("Basic ")
|
||||
|
||||
# Decode and verify the credentials
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
assert decoded == "user:pass"
|
||||
|
||||
def test_username_with_special_characters(self):
|
||||
"""Test basic auth with special characters in username"""
|
||||
result = basic_auth("user@example.com", "password123")
|
||||
assert result.startswith("Basic ")
|
||||
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
assert decoded == "user@example.com:password123"
|
||||
|
||||
def test_password_with_special_characters(self):
|
||||
"""Test basic auth with special characters in password"""
|
||||
result = basic_auth("admin", "p@ssw0rd!#$%")
|
||||
assert result.startswith("Basic ")
|
||||
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
assert decoded == "admin:p@ssw0rd!#$%"
|
||||
|
||||
def test_both_with_special_characters(self):
|
||||
"""Test basic auth with special characters in both username and password"""
|
||||
result = basic_auth("user@domain.com", "p@ss:w0rd!")
|
||||
assert result.startswith("Basic ")
|
||||
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
assert decoded == "user@domain.com:p@ss:w0rd!"
|
||||
|
||||
def test_empty_username(self):
|
||||
"""Test basic auth with empty username"""
|
||||
result = basic_auth("", "password")
|
||||
assert result.startswith("Basic ")
|
||||
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
assert decoded == ":password"
|
||||
|
||||
def test_empty_password(self):
|
||||
"""Test basic auth with empty password"""
|
||||
result = basic_auth("username", "")
|
||||
assert result.startswith("Basic ")
|
||||
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
assert decoded == "username:"
|
||||
|
||||
def test_both_empty(self):
|
||||
"""Test basic auth with both username and password empty"""
|
||||
result = basic_auth("", "")
|
||||
assert result.startswith("Basic ")
|
||||
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
assert decoded == ":"
|
||||
|
||||
def test_colon_in_username(self):
|
||||
"""Test basic auth with colon in username (edge case)"""
|
||||
result = basic_auth("user:name", "password")
|
||||
assert result.startswith("Basic ")
|
||||
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
assert decoded == "user:name:password"
|
||||
|
||||
def test_colon_in_password(self):
|
||||
"""Test basic auth with colon in password"""
|
||||
result = basic_auth("username", "pass:word")
|
||||
assert result.startswith("Basic ")
|
||||
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
assert decoded == "username:pass:word"
|
||||
|
||||
def test_unicode_characters(self):
|
||||
"""Test basic auth with unicode characters"""
|
||||
result = basic_auth("用户", "密码")
|
||||
assert result.startswith("Basic ")
|
||||
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
assert decoded == "用户:密码"
|
||||
|
||||
def test_long_credentials(self):
|
||||
"""Test basic auth with very long credentials"""
|
||||
long_user = "a" * 100
|
||||
long_pass = "b" * 100
|
||||
result = basic_auth(long_user, long_pass)
|
||||
assert result.startswith("Basic ")
|
||||
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
assert decoded == f"{long_user}:{long_pass}"
|
||||
|
||||
def test_whitespace_in_credentials(self):
|
||||
"""Test basic auth with whitespace in credentials"""
|
||||
result = basic_auth("user name", "pass word")
|
||||
assert result.startswith("Basic ")
|
||||
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
assert decoded == "user name:pass word"
|
||||
|
||||
def test_newlines_in_credentials(self):
|
||||
"""Test basic auth with newlines in credentials"""
|
||||
result = basic_auth("user\nname", "pass\nword")
|
||||
assert result.startswith("Basic ")
|
||||
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
assert decoded == "user\nname:pass\nword"
|
||||
|
||||
def test_return_type(self):
|
||||
"""Test that return type is string"""
|
||||
result = basic_auth("user", "pass")
|
||||
assert isinstance(result, str)
|
||||
|
||||
def test_format_consistency(self):
|
||||
"""Test that the format is always 'Basic <token>'"""
|
||||
result = basic_auth("user", "pass")
|
||||
parts = result.split(" ")
|
||||
assert len(parts) == 2
|
||||
assert parts[0] == "Basic"
|
||||
# Verify the second part is valid base64
|
||||
try:
|
||||
b64decode(parts[1])
|
||||
except (ValueError, TypeError) as e:
|
||||
pytest.fail(f"Invalid base64 encoding: {e}")
|
||||
|
||||
def test_known_value(self):
|
||||
"""Test against a known basic auth value"""
|
||||
# "user:pass" in base64 is "dXNlcjpwYXNz"
|
||||
result = basic_auth("user", "pass")
|
||||
assert result == "Basic dXNlcjpwYXNz"
|
||||
|
||||
def test_case_sensitivity(self):
|
||||
"""Test that username and password are case sensitive"""
|
||||
result1 = basic_auth("User", "Pass")
|
||||
result2 = basic_auth("user", "pass")
|
||||
assert result1 != result2
|
||||
|
||||
def test_ascii_encoding(self):
|
||||
"""Test that the result is ASCII encoded"""
|
||||
result = basic_auth("user", "pass")
|
||||
# Should not raise exception
|
||||
result.encode('ascii')
|
||||
|
||||
|
||||
# Parametrized tests
|
||||
@pytest.mark.parametrize("username,password,expected_decoded", [
|
||||
("admin", "admin123", "admin:admin123"),
|
||||
("user@example.com", "password", "user@example.com:password"),
|
||||
("test", "test!@#", "test:test!@#"),
|
||||
("", "password", ":password"),
|
||||
("username", "", "username:"),
|
||||
("", "", ":"),
|
||||
("user name", "pass word", "user name:pass word"),
|
||||
])
|
||||
def test_basic_auth_parametrized(username: str, password: str, expected_decoded: str):
|
||||
"""Parametrized test for basic_auth"""
|
||||
result = basic_auth(username, password)
|
||||
assert result.startswith("Basic ")
|
||||
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
assert decoded == expected_decoded
|
||||
|
||||
|
||||
@pytest.mark.parametrize("username,password", [
|
||||
("user", "pass"),
|
||||
("admin", "secret"),
|
||||
("test@example.com", "complex!@#$%^&*()"),
|
||||
("a" * 50, "b" * 50),
|
||||
])
|
||||
def test_basic_auth_roundtrip(username: str, password: str):
|
||||
"""Test that we can encode and decode credentials correctly"""
|
||||
result = basic_auth(username, password)
|
||||
|
||||
# Extract the encoded part
|
||||
encoded = result.split(" ")[1]
|
||||
|
||||
# Decode and verify
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
decoded_username, decoded_password = decoded.split(":", 1)
|
||||
|
||||
assert decoded_username == username
|
||||
assert decoded_password == password
|
||||
|
||||
|
||||
class TestBasicAuthIntegration:
|
||||
"""Integration tests for basic_auth"""
|
||||
|
||||
def test_http_header_format(self):
|
||||
"""Test that the output can be used as HTTP Authorization header"""
|
||||
auth_header = basic_auth("user", "pass")
|
||||
|
||||
# Simulate HTTP header
|
||||
headers = {"Authorization": auth_header}
|
||||
|
||||
assert "Authorization" in headers
|
||||
assert headers["Authorization"].startswith("Basic ")
|
||||
|
||||
def test_multiple_calls_consistency(self):
|
||||
"""Test that multiple calls with same credentials produce same result"""
|
||||
result1 = basic_auth("user", "pass")
|
||||
result2 = basic_auth("user", "pass")
|
||||
result3 = basic_auth("user", "pass")
|
||||
|
||||
assert result1 == result2 == result3
|
||||
|
||||
def test_different_credentials_different_results(self):
|
||||
"""Test that different credentials produce different results"""
|
||||
result1 = basic_auth("user1", "pass1")
|
||||
result2 = basic_auth("user2", "pass2")
|
||||
result3 = basic_auth("user1", "pass2")
|
||||
result4 = basic_auth("user2", "pass1")
|
||||
|
||||
results = [result1, result2, result3, result4]
|
||||
# All should be unique
|
||||
assert len(results) == len(set(results))
|
||||
|
||||
|
||||
# Edge cases and security considerations
|
||||
class TestBasicAuthEdgeCases:
|
||||
"""Edge case tests for basic_auth"""
|
||||
|
||||
def test_null_bytes(self):
|
||||
"""Test basic auth with null bytes (security consideration)"""
|
||||
result = basic_auth("user\x00", "pass\x00")
|
||||
assert result.startswith("Basic ")
|
||||
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
assert "user\x00" in decoded
|
||||
assert "pass\x00" in decoded
|
||||
|
||||
def test_very_long_username(self):
|
||||
"""Test with extremely long username"""
|
||||
long_username = "a" * 1000
|
||||
result = basic_auth(long_username, "pass")
|
||||
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
assert decoded.startswith(long_username)
|
||||
|
||||
def test_very_long_password(self):
|
||||
"""Test with extremely long password"""
|
||||
long_password = "b" * 1000
|
||||
result = basic_auth("user", long_password)
|
||||
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
assert decoded.endswith(long_password)
|
||||
|
||||
def test_emoji_in_credentials(self):
|
||||
"""Test with emoji characters"""
|
||||
result = basic_auth("user🔒", "pass🔑")
|
||||
assert result.startswith("Basic ")
|
||||
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
assert decoded == "user🔒:pass🔑"
|
||||
|
||||
def test_multiple_colons(self):
|
||||
"""Test with multiple colons in credentials"""
|
||||
result = basic_auth("user:name:test", "pass:word:test")
|
||||
assert result.startswith("Basic ")
|
||||
|
||||
encoded = result.split(" ")[1]
|
||||
decoded = b64decode(encoded).decode("utf-8")
|
||||
# Only first colon is separator, rest are part of credentials
|
||||
assert decoded == "user:name:test:pass:word:test"
|
||||
|
||||
def test_base64_special_chars(self):
|
||||
"""Test credentials that might produce base64 with padding"""
|
||||
# These lengths should produce different padding
|
||||
result1 = basic_auth("a", "a")
|
||||
result2 = basic_auth("ab", "ab")
|
||||
result3 = basic_auth("abc", "abc")
|
||||
|
||||
# All should be valid
|
||||
for result in [result1, result2, result3]:
|
||||
assert result.startswith("Basic ")
|
||||
encoded = result.split(" ")[1]
|
||||
b64decode(encoded) # Should not raise
|
||||
|
||||
|
||||
# __END__
|
||||
812
tests/unit/requests_handling/test_caller.py
Normal file
812
tests/unit/requests_handling/test_caller.py
Normal file
@@ -0,0 +1,812 @@
|
||||
"""
|
||||
PyTest: requests_handling/caller
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
from unittest.mock import Mock, patch
|
||||
import pytest
|
||||
import requests
|
||||
from corelibs.requests_handling.caller import Caller
|
||||
|
||||
|
||||
class TestCallerInit:
|
||||
"""Tests for Caller initialization"""
|
||||
|
||||
def test_init_with_required_params_only(self):
|
||||
"""Test Caller initialization with only required parameters"""
|
||||
header = {"Authorization": "Bearer token"}
|
||||
caller = Caller(header=header)
|
||||
|
||||
assert caller.headers == header
|
||||
assert caller.timeout == 20
|
||||
assert caller.verify is True
|
||||
assert caller.proxy is None
|
||||
assert caller.cafile is None
|
||||
|
||||
def test_init_with_all_params(self):
|
||||
"""Test Caller initialization with all parameters"""
|
||||
header = {"Authorization": "Bearer token", "Content-Type": "application/json"}
|
||||
proxy = {"http": "http://proxy.example.com:8080", "https": "https://proxy.example.com:8080"}
|
||||
caller = Caller(header=header, verify=False, timeout=30, proxy=proxy)
|
||||
|
||||
assert caller.headers == header
|
||||
assert caller.timeout == 30
|
||||
assert caller.verify is False
|
||||
assert caller.proxy == proxy
|
||||
|
||||
def test_init_with_empty_header(self):
|
||||
"""Test Caller initialization with empty header"""
|
||||
caller = Caller(header={})
|
||||
|
||||
assert caller.headers == {}
|
||||
assert caller.timeout == 20
|
||||
|
||||
def test_init_custom_timeout(self):
|
||||
"""Test Caller initialization with custom timeout"""
|
||||
caller = Caller(header={}, timeout=60)
|
||||
|
||||
assert caller.timeout == 60
|
||||
|
||||
def test_init_verify_false(self):
|
||||
"""Test Caller initialization with verify=False"""
|
||||
caller = Caller(header={}, verify=False)
|
||||
|
||||
assert caller.verify is False
|
||||
|
||||
def test_init_with_ca_file(self):
|
||||
"""Test Caller initialization with ca_file parameter"""
|
||||
ca_file_path = "/path/to/ca/cert.pem"
|
||||
caller = Caller(header={}, ca_file=ca_file_path)
|
||||
|
||||
assert caller.cafile == ca_file_path
|
||||
|
||||
|
||||
class TestCallerGet:
|
||||
"""Tests for Caller.get method"""
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.get')
|
||||
def test_get_basic(self, mock_get: Mock):
|
||||
"""Test basic GET request"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_response.status_code = 200
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
caller = Caller(header={"Authorization": "Bearer token"})
|
||||
response = caller.get("https://api.example.com/data")
|
||||
|
||||
assert response == mock_response
|
||||
mock_get.assert_called_once_with(
|
||||
"https://api.example.com/data",
|
||||
params=None,
|
||||
headers={"Authorization": "Bearer token"},
|
||||
timeout=20,
|
||||
verify=True,
|
||||
proxies=None
|
||||
)
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.get')
|
||||
def test_get_with_params(self, mock_get: Mock):
|
||||
"""Test GET request with query parameters"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
caller = Caller(header={})
|
||||
params = {"page": 1, "limit": 10}
|
||||
response = caller.get("https://api.example.com/data", params=params)
|
||||
|
||||
assert response == mock_response
|
||||
mock_get.assert_called_once_with(
|
||||
"https://api.example.com/data",
|
||||
params=params,
|
||||
headers={},
|
||||
timeout=20,
|
||||
verify=True,
|
||||
proxies=None
|
||||
)
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.get')
|
||||
def test_get_with_custom_timeout(self, mock_get: Mock):
|
||||
"""Test GET request uses default timeout from instance"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
caller = Caller(header={}, timeout=45)
|
||||
caller.get("https://api.example.com/data")
|
||||
|
||||
mock_get.assert_called_once()
|
||||
assert mock_get.call_args[1]["timeout"] == 45
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.get')
|
||||
def test_get_with_verify_false(self, mock_get: Mock):
|
||||
"""Test GET request with verify=False"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
caller = Caller(header={}, verify=False)
|
||||
caller.get("https://api.example.com/data")
|
||||
|
||||
mock_get.assert_called_once()
|
||||
assert mock_get.call_args[1]["verify"] is False
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.get')
|
||||
def test_get_with_proxy(self, mock_get: Mock):
|
||||
"""Test GET request with proxy"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
proxy = {"http": "http://proxy.example.com:8080"}
|
||||
caller = Caller(header={}, proxy=proxy)
|
||||
caller.get("https://api.example.com/data")
|
||||
|
||||
mock_get.assert_called_once()
|
||||
assert mock_get.call_args[1]["proxies"] == proxy
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.get')
|
||||
def test_get_invalid_schema_returns_none(self, mock_get: Mock, capsys: Any):
|
||||
"""Test GET request with invalid URL schema returns None"""
|
||||
mock_get.side_effect = requests.exceptions.InvalidSchema("Invalid URL")
|
||||
|
||||
caller = Caller(header={})
|
||||
response = caller.get("invalid://example.com")
|
||||
|
||||
assert response is None
|
||||
captured = capsys.readouterr()
|
||||
assert "Invalid URL during 'get'" in captured.out
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.get')
|
||||
def test_get_timeout_returns_none(self, mock_get: Mock, capsys: Any):
|
||||
"""Test GET request timeout returns None"""
|
||||
mock_get.side_effect = requests.exceptions.ReadTimeout("Timeout")
|
||||
|
||||
caller = Caller(header={})
|
||||
response = caller.get("https://api.example.com/data")
|
||||
|
||||
assert response is None
|
||||
captured = capsys.readouterr()
|
||||
assert "Timeout (20s) during 'get'" in captured.out
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.get')
|
||||
def test_get_connection_error_returns_none(self, mock_get: Mock, capsys: Any):
|
||||
"""Test GET request connection error returns None"""
|
||||
mock_get.side_effect = requests.exceptions.ConnectionError("Connection failed")
|
||||
|
||||
caller = Caller(header={})
|
||||
response = caller.get("https://api.example.com/data")
|
||||
|
||||
assert response is None
|
||||
captured = capsys.readouterr()
|
||||
assert "Connection error during 'get'" in captured.out
|
||||
|
||||
|
||||
class TestCallerPost:
|
||||
"""Tests for Caller.post method"""
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.post')
|
||||
def test_post_basic(self, mock_post: Mock):
|
||||
"""Test basic POST request"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_response.status_code = 201
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
caller = Caller(header={"Content-Type": "application/json"})
|
||||
data = {"name": "test", "value": 123}
|
||||
response = caller.post("https://api.example.com/data", data=data)
|
||||
|
||||
assert response == mock_response
|
||||
mock_post.assert_called_once_with(
|
||||
"https://api.example.com/data",
|
||||
params=None,
|
||||
json=data,
|
||||
headers={"Content-Type": "application/json"},
|
||||
timeout=20,
|
||||
verify=True,
|
||||
proxies=None
|
||||
)
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.post')
|
||||
def test_post_without_data(self, mock_post: Mock):
|
||||
"""Test POST request without data"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
caller = Caller(header={})
|
||||
response = caller.post("https://api.example.com/data")
|
||||
|
||||
assert response == mock_response
|
||||
mock_post.assert_called_once()
|
||||
# Data defaults to None, which becomes {} in __call
|
||||
assert mock_post.call_args[1]["json"] == {}
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.post')
|
||||
def test_post_with_params(self, mock_post: Mock):
|
||||
"""Test POST request with query parameters"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
caller = Caller(header={})
|
||||
data = {"key": "value"}
|
||||
params = {"version": "v1"}
|
||||
response = caller.post("https://api.example.com/data", data=data, params=params)
|
||||
|
||||
assert response == mock_response
|
||||
mock_post.assert_called_once()
|
||||
assert mock_post.call_args[1]["params"] == params
|
||||
assert mock_post.call_args[1]["json"] == data
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.post')
|
||||
def test_post_invalid_schema_returns_none(self, mock_post: Mock, capsys: Any):
|
||||
"""Test POST request with invalid URL schema returns None"""
|
||||
mock_post.side_effect = requests.exceptions.InvalidSchema("Invalid URL")
|
||||
|
||||
caller = Caller(header={})
|
||||
response = caller.post("invalid://example.com", data={"test": "data"})
|
||||
|
||||
assert response is None
|
||||
captured = capsys.readouterr()
|
||||
assert "Invalid URL during 'post'" in captured.out
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.post')
|
||||
def test_post_timeout_returns_none(self, mock_post: Mock, capsys: Any):
|
||||
"""Test POST request timeout returns None"""
|
||||
mock_post.side_effect = requests.exceptions.ReadTimeout("Timeout")
|
||||
|
||||
caller = Caller(header={})
|
||||
response = caller.post("https://api.example.com/data", data={"test": "data"})
|
||||
|
||||
assert response is None
|
||||
captured = capsys.readouterr()
|
||||
assert "Timeout (20s) during 'post'" in captured.out
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.post')
|
||||
def test_post_connection_error_returns_none(self, mock_post: Mock, capsys: Any):
|
||||
"""Test POST request connection error returns None"""
|
||||
mock_post.side_effect = requests.exceptions.ConnectionError("Connection failed")
|
||||
|
||||
caller = Caller(header={})
|
||||
response = caller.post("https://api.example.com/data", data={"test": "data"})
|
||||
|
||||
assert response is None
|
||||
captured = capsys.readouterr()
|
||||
assert "Connection error during 'post'" in captured.out
|
||||
|
||||
|
||||
class TestCallerPut:
|
||||
"""Tests for Caller.put method"""
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.put')
|
||||
def test_put_basic(self, mock_put: Mock):
|
||||
"""Test basic PUT request"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_response.status_code = 200
|
||||
mock_put.return_value = mock_response
|
||||
|
||||
caller = Caller(header={"Content-Type": "application/json"})
|
||||
data = {"id": 1, "name": "updated"}
|
||||
response = caller.put("https://api.example.com/data/1", data=data)
|
||||
|
||||
assert response == mock_response
|
||||
mock_put.assert_called_once_with(
|
||||
"https://api.example.com/data/1",
|
||||
params=None,
|
||||
json=data,
|
||||
headers={"Content-Type": "application/json"},
|
||||
timeout=20,
|
||||
verify=True,
|
||||
proxies=None
|
||||
)
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.put')
|
||||
def test_put_with_params(self, mock_put: Mock):
|
||||
"""Test PUT request with query parameters"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_put.return_value = mock_response
|
||||
|
||||
caller = Caller(header={})
|
||||
data = {"name": "test"}
|
||||
params = {"force": "true"}
|
||||
response = caller.put("https://api.example.com/data/1", data=data, params=params)
|
||||
|
||||
assert response == mock_response
|
||||
mock_put.assert_called_once()
|
||||
assert mock_put.call_args[1]["params"] == params
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.put')
|
||||
def test_put_timeout_returns_none(self, mock_put: Mock, capsys: Any):
|
||||
"""Test PUT request timeout returns None"""
|
||||
mock_put.side_effect = requests.exceptions.ReadTimeout("Timeout")
|
||||
|
||||
caller = Caller(header={})
|
||||
response = caller.put("https://api.example.com/data/1", data={"test": "data"})
|
||||
|
||||
assert response is None
|
||||
captured = capsys.readouterr()
|
||||
assert "Timeout (20s) during 'put'" in captured.out
|
||||
|
||||
|
||||
class TestCallerPatch:
|
||||
"""Tests for Caller.patch method"""
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.patch')
|
||||
def test_patch_basic(self, mock_patch: Mock):
|
||||
"""Test basic PATCH request"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_response.status_code = 200
|
||||
mock_patch.return_value = mock_response
|
||||
|
||||
caller = Caller(header={"Content-Type": "application/json"})
|
||||
data = {"status": "active"}
|
||||
response = caller.patch("https://api.example.com/data/1", data=data)
|
||||
|
||||
assert response == mock_response
|
||||
mock_patch.assert_called_once_with(
|
||||
"https://api.example.com/data/1",
|
||||
params=None,
|
||||
json=data,
|
||||
headers={"Content-Type": "application/json"},
|
||||
timeout=20,
|
||||
verify=True,
|
||||
proxies=None
|
||||
)
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.patch')
|
||||
def test_patch_with_params(self, mock_patch: Mock):
|
||||
"""Test PATCH request with query parameters"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_patch.return_value = mock_response
|
||||
|
||||
caller = Caller(header={})
|
||||
data = {"field": "value"}
|
||||
params = {"notify": "false"}
|
||||
response = caller.patch("https://api.example.com/data/1", data=data, params=params)
|
||||
|
||||
assert response == mock_response
|
||||
mock_patch.assert_called_once()
|
||||
assert mock_patch.call_args[1]["params"] == params
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.patch')
|
||||
def test_patch_connection_error_returns_none(self, mock_patch: Mock, capsys: Any):
|
||||
"""Test PATCH request connection error returns None"""
|
||||
mock_patch.side_effect = requests.exceptions.ConnectionError("Connection failed")
|
||||
|
||||
caller = Caller(header={})
|
||||
response = caller.patch("https://api.example.com/data/1", data={"test": "data"})
|
||||
|
||||
assert response is None
|
||||
captured = capsys.readouterr()
|
||||
assert "Connection error during 'patch'" in captured.out
|
||||
|
||||
|
||||
class TestCallerDelete:
|
||||
"""Tests for Caller.delete method"""
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.delete')
|
||||
def test_delete_basic(self, mock_delete: Mock):
|
||||
"""Test basic DELETE request"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_response.status_code = 204
|
||||
mock_delete.return_value = mock_response
|
||||
|
||||
caller = Caller(header={"Authorization": "Bearer token"})
|
||||
response = caller.delete("https://api.example.com/data/1")
|
||||
|
||||
assert response == mock_response
|
||||
mock_delete.assert_called_once_with(
|
||||
"https://api.example.com/data/1",
|
||||
params=None,
|
||||
headers={"Authorization": "Bearer token"},
|
||||
timeout=20,
|
||||
verify=True,
|
||||
proxies=None
|
||||
)
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.delete')
|
||||
def test_delete_with_params(self, mock_delete: Mock):
|
||||
"""Test DELETE request with query parameters"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_delete.return_value = mock_response
|
||||
|
||||
caller = Caller(header={})
|
||||
params = {"force": "true"}
|
||||
response = caller.delete("https://api.example.com/data/1", params=params)
|
||||
|
||||
assert response == mock_response
|
||||
mock_delete.assert_called_once()
|
||||
assert mock_delete.call_args[1]["params"] == params
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.delete')
|
||||
def test_delete_invalid_schema_returns_none(self, mock_delete: Mock, capsys: Any):
|
||||
"""Test DELETE request with invalid URL schema returns None"""
|
||||
mock_delete.side_effect = requests.exceptions.InvalidSchema("Invalid URL")
|
||||
|
||||
caller = Caller(header={})
|
||||
response = caller.delete("invalid://example.com/data/1")
|
||||
|
||||
assert response is None
|
||||
captured = capsys.readouterr()
|
||||
assert "Invalid URL during 'delete'" in captured.out
|
||||
|
||||
|
||||
class TestCallerParametrized:
|
||||
"""Parametrized tests for all HTTP methods"""
|
||||
|
||||
@pytest.mark.parametrize("method,http_method", [
|
||||
("get", "get"),
|
||||
("post", "post"),
|
||||
("put", "put"),
|
||||
("patch", "patch"),
|
||||
("delete", "delete"),
|
||||
])
|
||||
@patch('corelibs.requests_handling.caller.requests')
|
||||
def test_all_methods_use_correct_headers(self, mock_requests: Mock, method: str, http_method: str):
|
||||
"""Test that all HTTP methods use the headers correctly"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_http_method = getattr(mock_requests, http_method)
|
||||
mock_http_method.return_value = mock_response
|
||||
|
||||
headers = {"Authorization": "Bearer token", "X-Custom": "value"}
|
||||
caller = Caller(header=headers)
|
||||
|
||||
# Call the method
|
||||
caller_method = getattr(caller, method)
|
||||
if method in ["get", "delete"]:
|
||||
caller_method("https://api.example.com/data")
|
||||
else:
|
||||
caller_method("https://api.example.com/data", data={"key": "value"})
|
||||
|
||||
# Verify headers were passed
|
||||
mock_http_method.assert_called_once()
|
||||
assert mock_http_method.call_args[1]["headers"] == headers
|
||||
|
||||
@pytest.mark.parametrize("method,http_method", [
|
||||
("get", "get"),
|
||||
("post", "post"),
|
||||
("put", "put"),
|
||||
("patch", "patch"),
|
||||
("delete", "delete"),
|
||||
])
|
||||
@patch('corelibs.requests_handling.caller.requests')
|
||||
def test_all_methods_use_timeout(self, mock_requests: Mock, method: str, http_method: str):
|
||||
"""Test that all HTTP methods use the timeout correctly"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_http_method = getattr(mock_requests, http_method)
|
||||
mock_http_method.return_value = mock_response
|
||||
|
||||
timeout = 45
|
||||
caller = Caller(header={}, timeout=timeout)
|
||||
|
||||
# Call the method
|
||||
caller_method = getattr(caller, method)
|
||||
if method in ["get", "delete"]:
|
||||
caller_method("https://api.example.com/data")
|
||||
else:
|
||||
caller_method("https://api.example.com/data", data={"key": "value"})
|
||||
|
||||
# Verify timeout was passed
|
||||
mock_http_method.assert_called_once()
|
||||
assert mock_http_method.call_args[1]["timeout"] == timeout
|
||||
|
||||
@pytest.mark.parametrize("exception_class,expected_message", [
|
||||
(requests.exceptions.InvalidSchema, "Invalid URL during"),
|
||||
(requests.exceptions.ReadTimeout, "Timeout"),
|
||||
(requests.exceptions.ConnectionError, "Connection error during"),
|
||||
])
|
||||
@patch('corelibs.requests_handling.caller.requests.get')
|
||||
def test_exception_handling(
|
||||
self, mock_get: Mock, exception_class: type, expected_message: str, capsys: Any
|
||||
):
|
||||
"""Test exception handling for all exception types"""
|
||||
mock_get.side_effect = exception_class("Test error")
|
||||
|
||||
caller = Caller(header={})
|
||||
response = caller.get("https://api.example.com/data")
|
||||
|
||||
assert response is None
|
||||
captured = capsys.readouterr()
|
||||
assert expected_message in captured.out
|
||||
|
||||
|
||||
class TestCallerIntegration:
|
||||
"""Integration tests for Caller"""
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests')
|
||||
def test_multiple_requests_maintain_state(self, mock_requests: Mock):
|
||||
"""Test that multiple requests maintain caller state"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_requests.get.return_value = mock_response
|
||||
mock_requests.post.return_value = mock_response
|
||||
|
||||
headers = {"Authorization": "Bearer token"}
|
||||
caller = Caller(header=headers, timeout=30, verify=False)
|
||||
|
||||
# Make multiple requests
|
||||
caller.get("https://api.example.com/data1")
|
||||
caller.post("https://api.example.com/data2", data={"key": "value"})
|
||||
|
||||
# Verify both used same configuration
|
||||
assert mock_requests.get.call_args[1]["headers"] == headers
|
||||
assert mock_requests.get.call_args[1]["timeout"] == 30
|
||||
assert mock_requests.get.call_args[1]["verify"] is False
|
||||
|
||||
assert mock_requests.post.call_args[1]["headers"] == headers
|
||||
assert mock_requests.post.call_args[1]["timeout"] == 30
|
||||
assert mock_requests.post.call_args[1]["verify"] is False
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.post')
|
||||
def test_post_with_complex_data(self, mock_post: Mock):
|
||||
"""Test POST request with complex nested data"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
caller = Caller(header={})
|
||||
complex_data = {
|
||||
"user": {
|
||||
"name": "John Doe",
|
||||
"email": "john@example.com",
|
||||
"preferences": {
|
||||
"notifications": True,
|
||||
"theme": "dark"
|
||||
}
|
||||
},
|
||||
"tags": ["important", "urgent"],
|
||||
"count": 42
|
||||
}
|
||||
response = caller.post("https://api.example.com/users", data=complex_data)
|
||||
|
||||
assert response == mock_response
|
||||
mock_post.assert_called_once()
|
||||
assert mock_post.call_args[1]["json"] == complex_data
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests')
|
||||
def test_all_http_methods_work_together(self, mock_requests: Mock):
|
||||
"""Test that all HTTP methods can be used with the same Caller instance"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
for method in ['get', 'post', 'put', 'patch', 'delete']:
|
||||
getattr(mock_requests, method).return_value = mock_response
|
||||
|
||||
caller = Caller(header={"Authorization": "Bearer token"})
|
||||
|
||||
# Test all methods
|
||||
caller.get("https://api.example.com/data")
|
||||
caller.post("https://api.example.com/data", data={"new": "data"})
|
||||
caller.put("https://api.example.com/data/1", data={"updated": "data"})
|
||||
caller.patch("https://api.example.com/data/1", data={"field": "value"})
|
||||
caller.delete("https://api.example.com/data/1")
|
||||
|
||||
# Verify all were called
|
||||
mock_requests.get.assert_called_once()
|
||||
mock_requests.post.assert_called_once()
|
||||
mock_requests.put.assert_called_once()
|
||||
mock_requests.patch.assert_called_once()
|
||||
mock_requests.delete.assert_called_once()
|
||||
|
||||
|
||||
class TestCallerEdgeCases:
|
||||
"""Edge case tests for Caller"""
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.get')
|
||||
def test_empty_url(self, mock_get: Mock):
|
||||
"""Test with empty URL"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
caller = Caller(header={})
|
||||
response = caller.get("")
|
||||
|
||||
assert response == mock_response
|
||||
mock_get.assert_called_once_with(
|
||||
"",
|
||||
params=None,
|
||||
headers={},
|
||||
timeout=20,
|
||||
verify=True,
|
||||
proxies=None
|
||||
)
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.post')
|
||||
def test_post_with_empty_data(self, mock_post: Mock):
|
||||
"""Test POST with explicitly empty data dict"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
caller = Caller(header={})
|
||||
response = caller.post("https://api.example.com/data", data={})
|
||||
|
||||
assert response == mock_response
|
||||
mock_post.assert_called_once()
|
||||
assert mock_post.call_args[1]["json"] == {}
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.get')
|
||||
def test_get_with_empty_params(self, mock_get: Mock):
|
||||
"""Test GET with explicitly empty params dict"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
caller = Caller(header={})
|
||||
response = caller.get("https://api.example.com/data", params={})
|
||||
|
||||
assert response == mock_response
|
||||
mock_get.assert_called_once()
|
||||
assert mock_get.call_args[1]["params"] == {}
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.post')
|
||||
def test_post_with_none_values_in_data(self, mock_post: Mock):
|
||||
"""Test POST with None values in data"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
caller = Caller(header={})
|
||||
data = {"key1": None, "key2": "value", "key3": None}
|
||||
response = caller.post("https://api.example.com/data", data=data)
|
||||
|
||||
assert response == mock_response
|
||||
mock_post.assert_called_once()
|
||||
assert mock_post.call_args[1]["json"] == data
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.get')
|
||||
def test_very_long_url(self, mock_get: Mock):
|
||||
"""Test with very long URL"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
caller = Caller(header={})
|
||||
long_url = "https://api.example.com/" + "a" * 1000
|
||||
response = caller.get(long_url)
|
||||
|
||||
assert response == mock_response
|
||||
mock_get.assert_called_once_with(
|
||||
long_url,
|
||||
params=None,
|
||||
headers={},
|
||||
timeout=20,
|
||||
verify=True,
|
||||
proxies=None
|
||||
)
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.get')
|
||||
def test_special_characters_in_url(self, mock_get: Mock):
|
||||
"""Test URL with special characters"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
caller = Caller(header={})
|
||||
url = "https://api.example.com/data?query=test%20value&id=123"
|
||||
response = caller.get(url)
|
||||
|
||||
assert response == mock_response
|
||||
mock_get.assert_called_once_with(
|
||||
url,
|
||||
params=None,
|
||||
headers={},
|
||||
timeout=20,
|
||||
verify=True,
|
||||
proxies=None
|
||||
)
|
||||
|
||||
def test_timeout_zero(self):
|
||||
"""Test Caller with timeout of 0"""
|
||||
caller = Caller(header={}, timeout=0)
|
||||
assert caller.timeout == 0
|
||||
|
||||
def test_negative_timeout(self):
|
||||
"""Test Caller with negative timeout"""
|
||||
caller = Caller(header={}, timeout=-1)
|
||||
assert caller.timeout == -1
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.get')
|
||||
def test_unicode_in_headers(self, mock_get: Mock):
|
||||
"""Test headers with unicode characters"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
headers = {"X-Custom": "测试", "Authorization": "Bearer token"}
|
||||
caller = Caller(header=headers)
|
||||
response = caller.get("https://api.example.com/data")
|
||||
|
||||
assert response == mock_response
|
||||
mock_get.assert_called_once()
|
||||
assert mock_get.call_args[1]["headers"] == headers
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.post')
|
||||
def test_unicode_in_data(self, mock_post: Mock):
|
||||
"""Test data with unicode characters"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
caller = Caller(header={})
|
||||
data = {"name": "用户", "message": "こんにちは", "emoji": "🚀"}
|
||||
response = caller.post("https://api.example.com/data", data=data)
|
||||
|
||||
assert response == mock_response
|
||||
mock_post.assert_called_once()
|
||||
assert mock_post.call_args[1]["json"] == data
|
||||
|
||||
|
||||
class TestCallerProxyHandling:
|
||||
"""Tests for proxy handling"""
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.get')
|
||||
def test_proxy_configuration(self, mock_get: Mock):
|
||||
"""Test that proxy configuration is passed to requests"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
proxy = {
|
||||
"http": "http://proxy.example.com:8080",
|
||||
"https": "https://proxy.example.com:8080"
|
||||
}
|
||||
caller = Caller(header={}, proxy=proxy)
|
||||
caller.get("https://api.example.com/data")
|
||||
|
||||
mock_get.assert_called_once()
|
||||
assert mock_get.call_args[1]["proxies"] == proxy
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.post')
|
||||
def test_proxy_with_auth(self, mock_post: Mock):
|
||||
"""Test proxy with authentication"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
proxy = {
|
||||
"http": "http://user:pass@proxy.example.com:8080",
|
||||
"https": "https://user:pass@proxy.example.com:8080"
|
||||
}
|
||||
caller = Caller(header={}, proxy=proxy)
|
||||
caller.post("https://api.example.com/data", data={"test": "data"})
|
||||
|
||||
mock_post.assert_called_once()
|
||||
assert mock_post.call_args[1]["proxies"] == proxy
|
||||
|
||||
|
||||
class TestCallerTimeoutHandling:
|
||||
"""Tests for timeout parameter handling"""
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.get')
|
||||
def test_timeout_parameter_none_uses_default(self, mock_get: Mock):
|
||||
"""Test that None timeout uses the instance default"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
caller = Caller(header={}, timeout=30)
|
||||
# The private __timeout method is called internally
|
||||
caller.get("https://api.example.com/data")
|
||||
|
||||
mock_get.assert_called_once()
|
||||
assert mock_get.call_args[1]["timeout"] == 30
|
||||
|
||||
|
||||
class TestCallerResponseHandling:
|
||||
"""Tests for response handling"""
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.get')
|
||||
def test_response_object_returned_correctly(self, mock_get: Mock):
|
||||
"""Test that response object is returned correctly"""
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_response.status_code = 200
|
||||
mock_response.text = "Success"
|
||||
mock_response.json.return_value = {"status": "ok"}
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
caller = Caller(header={})
|
||||
response = caller.get("https://api.example.com/data")
|
||||
|
||||
assert response is not None
|
||||
assert response.status_code == 200
|
||||
assert response.text == "Success"
|
||||
assert response.json() == {"status": "ok"}
|
||||
|
||||
@patch('corelibs.requests_handling.caller.requests.get')
|
||||
def test_response_with_different_status_codes(self, mock_get: Mock):
|
||||
"""Test response handling with different status codes"""
|
||||
for status_code in [200, 201, 204, 400, 401, 404, 500]:
|
||||
mock_response = Mock(spec=requests.Response)
|
||||
mock_response.status_code = status_code
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
caller = Caller(header={})
|
||||
response = caller.get("https://api.example.com/data")
|
||||
|
||||
assert response is not None
|
||||
assert response.status_code == status_code
|
||||
|
||||
|
||||
# __END__
|
||||
Reference in New Issue
Block a user