From f265b55ef8eadcb2a15109adc1e34490ef7336bf Mon Sep 17 00:00:00 2001 From: Clemens Schwaighofer Date: Wed, 4 Feb 2026 14:55:39 +0900 Subject: [PATCH] Move requests handling to corelibs_requests module --- pyproject.toml | 1 + src/corelibs/iterator_handling/dict_mask.py | 2 +- .../requests_handling/auth_helpers.py | 9 +- src/corelibs/requests_handling/caller.py | 255 +----- test-run/requests_handling/caller.py | 38 + tests/unit/requests_handling/__init__.py | 3 - .../requests_handling/test_auth_helpers.py | 308 ------- tests/unit/requests_handling/test_caller.py | 847 ------------------ 8 files changed, 59 insertions(+), 1404 deletions(-) create mode 100644 test-run/requests_handling/caller.py delete mode 100644 tests/unit/requests_handling/__init__.py delete mode 100644 tests/unit/requests_handling/test_auth_helpers.py delete mode 100644 tests/unit/requests_handling/test_caller.py diff --git a/pyproject.toml b/pyproject.toml index 1a7cac1..e36f6ec 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,6 +16,7 @@ dependencies = [ "corelibs-iterator>=1.0.0", "corelibs-json>=1.0.0", "corelibs-regex-checks>=1.0.0", + "corelibs-requests>=1.0.0", "corelibs-search>=1.0.0", "corelibs-stack-trace>=1.0.0", "corelibs-text-colors>=1.0.0", diff --git a/src/corelibs/iterator_handling/dict_mask.py b/src/corelibs/iterator_handling/dict_mask.py index 9ef1916..b27f5a4 100644 --- a/src/corelibs/iterator_handling/dict_mask.py +++ b/src/corelibs/iterator_handling/dict_mask.py @@ -21,7 +21,7 @@ def mask( mask_str: str = "***", mask_str_edges: str = '_', skip: bool = False -) -> dict[str, Any]: +) -> dict[str, Any] | list[Any]: """ mask data for output Checks if mask_keys list exist in any key in the data set either from the start or at the end diff --git a/src/corelibs/requests_handling/auth_helpers.py b/src/corelibs/requests_handling/auth_helpers.py index e46b8ea..47049d9 100644 --- a/src/corelibs/requests_handling/auth_helpers.py +++ b/src/corelibs/requests_handling/auth_helpers.py @@ -2,9 +2,11 @@ Various HTTP auth helpers """ -from base64 import b64encode +from warnings import deprecated +from corelibs_requests.auth_helpers import basic_auth as corelibs_basic_auth +@deprecated("use corelibs_requests.auth_helpers.basic_auth instead") def basic_auth(username: str, password: str) -> str: """ setup basic auth, for debug @@ -16,5 +18,6 @@ def basic_auth(username: str, password: str) -> str: Returns: str -- _description_ """ - token = b64encode(f"{username}:{password}".encode('utf-8')).decode("ascii") - return f'Basic {token}' + return corelibs_basic_auth(username, password) + +# __END__ diff --git a/src/corelibs/requests_handling/caller.py b/src/corelibs/requests_handling/caller.py index 9be2f90..3e615f8 100644 --- a/src/corelibs/requests_handling/caller.py +++ b/src/corelibs/requests_handling/caller.py @@ -3,264 +3,35 @@ requests lib interface V2 call type """ -from typing import Any, TypedDict, cast -import requests -from requests import exceptions +from warnings import warn +from corelibs_requests.caller import ( + Caller as CoreLibsCaller, + ProxyConfig as CoreLibsProxyConfig, + ErrorResponse as CoreLibsErrorResponse +) -class ErrorResponse: +class ErrorResponse(CoreLibsErrorResponse): """ Error response structure. This is returned if a request could not be completed """ - def __init__( - self, - code: int, - message: str, - action: str, - url: str, - exception: exceptions.InvalidSchema | exceptions.ReadTimeout | exceptions.ConnectionError | None = None - ) -> None: - self.code = code - self.message = message - self.action = action - self.url = url - self.exception_name = type(exception).__name__ if exception is not None else None - self.exception_trace = exception if exception is not None else None -class ProxyConfig(TypedDict): +class ProxyConfig(CoreLibsProxyConfig): """ Socks proxy settings """ - type: str - host: str - port: str -class Caller: +class Caller(CoreLibsCaller): """ requests lib interface """ - def __init__( - self, - header: dict[str, str], - timeout: int = 20, - proxy: ProxyConfig | None = None, - verify: bool = True, - ca_file: str | None = None - ): - self.headers = header - self.timeout: int = timeout - self.ca_file = ca_file - self.verify = verify - self.proxy = cast(dict[str, str], proxy) if proxy is not None else None - def __timeout(self, timeout: int | None) -> int: - if timeout is not None and timeout >= 0: - return timeout - return self.timeout - - def __call( - self, - action: str, - url: str, - data: dict[str, Any] | None = None, - params: dict[str, Any] | None = None, - timeout: int | None = None - ) -> requests.Response | ErrorResponse: - """ - call wrapper, on error returns None - - Args: - action (str): _description_ - url (str): _description_ - data (dict | None): _description_. Defaults to None. - params (dict | None): _description_. Defaults to None. - - Returns: - requests.Response | None: _description_ - """ - - if data is None: - data = {} - try: - if action == "get": - return requests.get( - url, - params=params, - headers=self.headers, - timeout=self.__timeout(timeout), - verify=self.verify, - proxies=self.proxy, - cert=self.ca_file - ) - if action == "post": - return requests.post( - url, - params=params, - json=data, - headers=self.headers, - timeout=self.__timeout(timeout), - verify=self.verify, - proxies=self.proxy, - cert=self.ca_file - ) - if action == "put": - return requests.put( - url, - params=params, - json=data, - headers=self.headers, - timeout=self.__timeout(timeout), - verify=self.verify, - proxies=self.proxy, - cert=self.ca_file - ) - if action == "patch": - return requests.patch( - url, - params=params, - json=data, - headers=self.headers, - timeout=self.__timeout(timeout), - verify=self.verify, - proxies=self.proxy, - cert=self.ca_file - ) - if action == "delete": - return requests.delete( - url, - params=params, - headers=self.headers, - timeout=self.__timeout(timeout), - verify=self.verify, - proxies=self.proxy, - cert=self.ca_file - ) - return ErrorResponse( - 100, - f"Unsupported action '{action}'", - action, - url - ) - except exceptions.InvalidSchema as e: - return ErrorResponse( - 200, - f"Invalid URL during '{action}' for {url}", - action, - url, - e - ) - except exceptions.ReadTimeout as e: - return ErrorResponse( - 300, - f"Timeout ({self.timeout}s) during '{action}' for {url}", - action, - url, - e - ) - except exceptions.ConnectionError as e: - return ErrorResponse( - 400, - f"Connection error during '{action}' for {url}", - action, - url, - e - ) - - def get( - self, - url: str, - params: dict[str, Any] | None = None, - timeout: int | None = None - ) -> requests.Response | ErrorResponse: - """ - get data - - Args: - url (str): _description_ - params (dict | None): _description_ - - Returns: - requests.Response: _description_ - """ - return self.__call('get', url, params=params, timeout=timeout) - - def post( - self, - url: str, - data: dict[str, Any] | None = None, - params: dict[str, Any] | None = None, - timeout: int | None = None - ) -> requests.Response | ErrorResponse: - """ - post data - - Args: - url (str): _description_ - data (dict | None): _description_ - params (dict | None): _description_ - - Returns: - requests.Response | None: _description_ - """ - return self.__call('post', url, data, params, timeout=timeout) - - def put( - self, - url: str, - data: dict[str, Any] | None = None, - params: dict[str, Any] | None = None, - timeout: int | None = None - ) -> requests.Response | ErrorResponse: - """_summary_ - - Args: - url (str): _description_ - data (dict | None): _description_ - params (dict | None): _description_ - - Returns: - requests.Response | None: _description_ - """ - return self.__call('put', url, data, params, timeout=timeout) - - def patch( - self, - url: str, - data: dict[str, Any] | None = None, - params: dict[str, Any] | None = None, - timeout: int | None = None - ) -> requests.Response | ErrorResponse: - """_summary_ - - Args: - url (str): _description_ - data (dict | None): _description_ - params (dict | None): _description_ - - Returns: - requests.Response | None: _description_ - """ - return self.__call('patch', url, data, params, timeout=timeout) - - def delete( - self, - url: str, - params: dict[str, Any] | None = None, - timeout: int | None = None - ) -> requests.Response | ErrorResponse: - """ - delete - - Args: - url (str): _description_ - params (dict | None): _description_ - - Returns: - requests.Response | None: _description_ - """ - return self.__call('delete', url, params=params, timeout=timeout) +warn( + "corelibs.requests_handling.caller is deprecated, use corelibs_requests.caller instead", + DeprecationWarning, stacklevel=2 +) # __END__ diff --git a/test-run/requests_handling/caller.py b/test-run/requests_handling/caller.py new file mode 100644 index 0000000..5604c37 --- /dev/null +++ b/test-run/requests_handling/caller.py @@ -0,0 +1,38 @@ +""" +Caller tests +""" + +from corelibs_dump_data.dump_data import dump_data +from corelibs.requests_handling.caller import Caller, ErrorResponse +from corelibs.requests_handling.auth_helpers import basic_auth + + +def test_basic_auth(): + """basic auth test""" + user = "user" + password = "pass" + auth_header = basic_auth(user, password) + print(f"Auth Header for '{user}' & '{password}': {auth_header}") + + +def test_caller(): + """Caller tests""" + caller = Caller() + response = caller.get("https://httpbin.org/get") + if isinstance(response, ErrorResponse): + print(f"Error: {response.message}") + else: + print(f"Response Status Code: {response.status_code}") + print(f"Response Content: {dump_data(response.json())}") + + +def main(): + """main""" + test_caller() + test_basic_auth() + + +if __name__ == "__main__": + main() + +# __END__ diff --git a/tests/unit/requests_handling/__init__.py b/tests/unit/requests_handling/__init__.py deleted file mode 100644 index ce0a14c..0000000 --- a/tests/unit/requests_handling/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -""" -PyTest: requests_handling tests -""" diff --git a/tests/unit/requests_handling/test_auth_helpers.py b/tests/unit/requests_handling/test_auth_helpers.py deleted file mode 100644 index f5a0d78..0000000 --- a/tests/unit/requests_handling/test_auth_helpers.py +++ /dev/null @@ -1,308 +0,0 @@ -""" -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 '""" - 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__ diff --git a/tests/unit/requests_handling/test_caller.py b/tests/unit/requests_handling/test_caller.py deleted file mode 100644 index b0730f2..0000000 --- a/tests/unit/requests_handling/test_caller.py +++ /dev/null @@ -1,847 +0,0 @@ -""" -PyTest: requests_handling/caller -""" - -from unittest.mock import Mock, patch -import pytest -import requests -from corelibs.requests_handling.caller import Caller, ErrorResponse, ProxyConfig - - -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.ca_file is None - - def test_init_with_all_params(self): - """Test Caller initialization with all parameters""" - header = {"Authorization": "Bearer token", "Content-Type": "application/json"} - proxy: ProxyConfig = { - "type": "socks5", - "host": "proxy.example.com:8080", - "port": "8080" - } - caller = Caller(header=header, timeout=30, proxy=proxy, verify=False) - - 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.ca_file == 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, - cert=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, - cert=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: ProxyConfig = { - "type": "socks5", - "host": "proxy.example.com:8080", - "port": "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): - """Test GET request with invalid URL schema returns ErrorResponse""" - mock_get.side_effect = requests.exceptions.InvalidSchema("Invalid URL") - - caller = Caller(header={}) - response = caller.get("invalid://example.com") - - assert isinstance(response, ErrorResponse) - assert response.code == 200 - assert "Invalid URL during 'get'" in response.message - assert response.action == "get" - assert response.url == "invalid://example.com" - - @patch('corelibs.requests_handling.caller.requests.get') - def test_get_timeout_returns_none(self, mock_get: Mock): - """Test GET request timeout returns ErrorResponse""" - mock_get.side_effect = requests.exceptions.ReadTimeout("Timeout") - - caller = Caller(header={}) - response = caller.get("https://api.example.com/data") - - assert isinstance(response, ErrorResponse) - assert response.code == 300 - assert "Timeout (20s) during 'get'" in response.message - assert response.action == "get" - assert response.url == "https://api.example.com/data" - - @patch('corelibs.requests_handling.caller.requests.get') - def test_get_connection_error_returns_none(self, mock_get: Mock): - """Test GET request connection error returns ErrorResponse""" - mock_get.side_effect = requests.exceptions.ConnectionError("Connection failed") - - caller = Caller(header={}) - response = caller.get("https://api.example.com/data") - - assert isinstance(response, ErrorResponse) - assert response.code == 400 - assert "Connection error during 'get'" in response.message - assert response.action == "get" - assert response.url == "https://api.example.com/data" - - -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, - cert=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): - """Test POST request with invalid URL schema returns ErrorResponse""" - mock_post.side_effect = requests.exceptions.InvalidSchema("Invalid URL") - - caller = Caller(header={}) - response = caller.post("invalid://example.com", data={"test": "data"}) - - assert isinstance(response, ErrorResponse) - assert response.code == 200 - assert "Invalid URL during 'post'" in response.message - assert response.action == "post" - assert response.url == "invalid://example.com" - - @patch('corelibs.requests_handling.caller.requests.post') - def test_post_timeout_returns_none(self, mock_post: Mock): - """Test POST request timeout returns ErrorResponse""" - mock_post.side_effect = requests.exceptions.ReadTimeout("Timeout") - - caller = Caller(header={}) - response = caller.post("https://api.example.com/data", data={"test": "data"}) - - assert isinstance(response, ErrorResponse) - assert response.code == 300 - assert "Timeout (20s) during 'post'" in response.message - assert response.action == "post" - assert response.url == "https://api.example.com/data" - - @patch('corelibs.requests_handling.caller.requests.post') - def test_post_connection_error_returns_none(self, mock_post: Mock): - """Test POST request connection error returns ErrorResponse""" - mock_post.side_effect = requests.exceptions.ConnectionError("Connection failed") - - caller = Caller(header={}) - response = caller.post("https://api.example.com/data", data={"test": "data"}) - - assert isinstance(response, ErrorResponse) - assert response.code == 400 - assert "Connection error during 'post'" in response.message - assert response.action == "post" - assert response.url == "https://api.example.com/data" - - -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, - cert=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): - """Test PUT request timeout returns ErrorResponse""" - mock_put.side_effect = requests.exceptions.ReadTimeout("Timeout") - - caller = Caller(header={}) - response = caller.put("https://api.example.com/data/1", data={"test": "data"}) - - assert isinstance(response, ErrorResponse) - assert response.code == 300 - assert "Timeout (20s) during 'put'" in response.message - assert response.action == "put" - assert response.url == "https://api.example.com/data/1" - - -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, - cert=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): - """Test PATCH request connection error returns ErrorResponse""" - 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 isinstance(response, ErrorResponse) - assert response.code == 400 - assert "Connection error during 'patch'" in response.message - assert response.action == "patch" - assert response.url == "https://api.example.com/data/1" - - -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, - cert=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): - """Test DELETE request with invalid URL schema returns ErrorResponse""" - mock_delete.side_effect = requests.exceptions.InvalidSchema("Invalid URL") - - caller = Caller(header={}) - response = caller.delete("invalid://example.com/data/1") - - assert isinstance(response, ErrorResponse) - assert response.code == 200 - assert "Invalid URL during 'delete'" in response.message - assert response.action == "delete" - assert response.url == "invalid://example.com/data/1" - - -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 - ): - """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 isinstance(response, ErrorResponse) - assert expected_message in response.message - - -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, - cert=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, - cert=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, - cert=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: ProxyConfig = { - "type": "socks5", - "host": "proxy.example.com:8080", - "port": "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: ProxyConfig = { - "type": "socks5", - "host": "proxy.example.com:8080", - "port": "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 not isinstance(response, ErrorResponse) - 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 not isinstance(response, ErrorResponse) - assert response.status_code == status_code - - -# __END__