# coding=utf-8 """ Unit tests for CallbackProvider @author: GitHub Copilot """ import os import sys import logging import random import platform from time import sleep from base_test import BaseProviderTestCase, unittest, patch from ddns.provider.callback import CallbackProvider class TestCallbackProvider(BaseProviderTestCase): """Test cases for CallbackProvider""" def setUp(self): """Set up test fixtures""" super(TestCallbackProvider, self).setUp() self.id = "https://example.com/callback?domain=__DOMAIN__&ip=__IP__" self.token = "" # Use empty string instead of None for token def test_init_with_basic_config(self): """Test CallbackProvider initialization with basic configuration""" provider = CallbackProvider(self.id, self.token) self.assertEqual(provider.id, self.id) self.assertEqual(provider.token, self.token) self.assertFalse(provider.decode_response) def test_init_with_token_config(self): """Test CallbackProvider initialization with token configuration""" token = '{"api_key": "__DOMAIN__", "value": "__IP__"}' provider = CallbackProvider(self.id, token) self.assertEqual(provider.token, token) def test_validate_success(self): """Test _validate method with valid configuration""" provider = CallbackProvider(self.id, self.token) # Should not raise any exception since we have a valid id provider._validate() def test_validate_failure_no_id(self): """Test _validate method with missing id""" # _validate is called in __init__, so we need to test it directly with self.assertRaises(ValueError) as cm: CallbackProvider(None, self.token) # type: ignore self.assertIn("id must be configured", str(cm.exception)) def test_validate_failure_empty_id(self): """Test _validate method with empty id""" # _validate is called in __init__, so we need to test it directly with self.assertRaises(ValueError) as cm: CallbackProvider("", self.token) self.assertIn("id must be configured", str(cm.exception)) def test_replace_vars_basic(self): """Test _replace_vars method with basic replacements""" provider = CallbackProvider(self.id, self.token) test_str = "Hello __NAME__, your IP is __IP__" mapping = {"__NAME__": "World", "__IP__": "192.168.1.1"} result = provider._replace_vars(test_str, mapping) expected = "Hello World, your IP is 192.168.1.1" self.assertEqual(result, expected) def test_replace_vars_no_matches(self): """Test _replace_vars method with no matching variables""" provider = CallbackProvider(self.id, self.token) test_str = "No variables here" mapping = {"__NAME__": "World"} result = provider._replace_vars(test_str, mapping) self.assertEqual(result, test_str) def test_replace_vars_partial_matches(self): """Test _replace_vars method with partial matches""" provider = CallbackProvider(self.id, self.token) test_str = "__DOMAIN__ and __UNKNOWN__ and __IP__" mapping = {"__DOMAIN__": "example.com", "__IP__": "1.2.3.4"} result = provider._replace_vars(test_str, mapping) expected = "example.com and __UNKNOWN__ and 1.2.3.4" self.assertEqual(result, expected) def test_replace_vars_empty_string(self): """Test _replace_vars method with empty string""" provider = CallbackProvider(self.id, self.token) result = provider._replace_vars("", {"__TEST__": "value"}) self.assertEqual(result, "") def test_replace_vars_empty_mapping(self): """Test _replace_vars method with empty mapping""" provider = CallbackProvider(self.id, self.token) test_str = "__DOMAIN__ test" result = provider._replace_vars(test_str, {}) self.assertEqual(result, test_str) def test_replace_vars_none_values(self): """Test _replace_vars method with None values (should convert to string)""" provider = CallbackProvider(self.id, self.token) test_str = "TTL: __TTL__, Line: __LINE__" mapping = {"__TTL__": None, "__LINE__": None} result = provider._replace_vars(test_str, mapping) expected = "TTL: None, Line: None" self.assertEqual(result, expected) def test_replace_vars_numeric_values(self): """Test _replace_vars method with numeric values (should convert to string)""" provider = CallbackProvider(self.id, self.token) test_str = "Port: __PORT__, TTL: __TTL__" mapping = {"__PORT__": 8080, "__TTL__": 300} result = provider._replace_vars(test_str, mapping) expected = "Port: 8080, TTL: 300" self.assertEqual(result, expected) @patch("ddns.provider.callback.time") @patch.object(CallbackProvider, "_http") def test_set_record_get_method(self, mock_http, mock_time): """Test set_record method using GET method (no token)""" mock_time.return_value = 1634567890.123 mock_http.return_value = "Success" provider = CallbackProvider(self.id, None) # type: ignore result = provider.set_record("example.com", "192.168.1.1", "A", 300, "default") # Verify the result self.assertTrue(result) # Verify _http was called with correct parameters mock_http.assert_called_once() args, kwargs = mock_http.call_args self.assertEqual(args[0], "GET") # method # Check that URL contains replaced variables url = args[1] self.assertIn("example.com", url) self.assertIn("192.168.1.1", url) @patch("ddns.provider.callback.time") @patch.object(CallbackProvider, "_http") def test_set_record_post_method_dict_token(self, mock_http, mock_time): """Test set_record method using POST method with dict token""" mock_time.return_value = 1634567890.123 mock_http.return_value = "Success" token = {"api_key": "test_key", "domain": "__DOMAIN__", "ip": "__IP__"} provider = CallbackProvider(self.id, token) # type: ignore result = provider.set_record("example.com", "192.168.1.1", "A", 300, "default") # Verify the result self.assertTrue(result) # Verify _http was called with correct parameters mock_http.assert_called_once() args, kwargs = mock_http.call_args self.assertEqual(args[0], "POST") # method # URL should be replaced with actual values even for POST url = args[1] self.assertIn("example.com", url) self.assertIn("192.168.1.1", url) # Check params were properly replaced params = kwargs["body"] self.assertEqual(params["api_key"], "test_key") self.assertEqual(params["domain"], "example.com") self.assertEqual(params["ip"], "192.168.1.1") @patch("ddns.provider.callback.time") @patch.object(CallbackProvider, "_http") def test_set_record_post_method_json_token(self, mock_http, mock_time): """Test set_record method using POST method with JSON string token""" mock_time.return_value = 1634567890.123 mock_http.return_value = "Success" token = '{"api_key": "test_key", "domain": "__DOMAIN__", "ip": "__IP__"}' provider = CallbackProvider(self.id, token) result = provider.set_record("example.com", "192.168.1.1", "A", 300, "default") # Verify the result self.assertTrue(result) # Verify _http was called with correct parameters mock_http.assert_called_once() args, kwargs = mock_http.call_args self.assertEqual(args[0], "POST") # method # URL should be replaced with actual values even for POST url = args[1] self.assertIn("example.com", url) self.assertIn("192.168.1.1", url) # Check params were properly replaced params = kwargs["body"] self.assertEqual(params["api_key"], "test_key") self.assertEqual(params["domain"], "example.com") self.assertEqual(params["ip"], "192.168.1.1") @patch("ddns.provider.callback.time") @patch.object(CallbackProvider, "_http") def test_set_record_post_method_mixed_types(self, mock_http, mock_time): """Test set_record method with mixed type values in POST parameters""" mock_time.return_value = 1634567890.123 mock_http.return_value = "Success" token = {"api_key": 12345, "domain": "__DOMAIN__", "timeout": 30, "enabled": True} provider = CallbackProvider(self.id, token) # type: ignore result = provider.set_record("example.com", "192.168.1.1") # Verify the result self.assertTrue(result) # Verify _http was called with correct parameters mock_http.assert_called_once() args, kwargs = mock_http.call_args self.assertEqual(args[0], "POST") # method # Check that non-string values were not processed, but string values were replaced params = kwargs["body"] self.assertEqual(params["api_key"], 12345) # unchanged (not a string) self.assertEqual(params["domain"], "example.com") # replaced (was a string) self.assertEqual(params["timeout"], 30) # unchanged (not a string) self.assertEqual(params["enabled"], True) # unchanged (not a string) @patch("ddns.provider.callback.time") @patch.object(CallbackProvider, "_http") def test_set_record_http_failure(self, mock_http, mock_time): """Test set_record method when HTTP request fails""" mock_time.return_value = 1634567890.123 mock_http.return_value = None # Simulate failure provider = CallbackProvider(self.id, None) # type: ignore result = provider.set_record("example.com", "192.168.1.1") # Verify the result is False on failure self.assertFalse(result) @patch("ddns.provider.callback.time") @patch.object(CallbackProvider, "_http") def test_set_record_http_none_response(self, mock_http, mock_time): """Test set_record method with None HTTP response""" mock_time.return_value = 1634567890.123 mock_http.return_value = None # None response provider = CallbackProvider(self.id, None) # type: ignore result = provider.set_record("example.com", "192.168.1.1") # Empty string is falsy, so result should be False self.assertFalse(result) @patch("ddns.provider.callback.jsondecode") def test_json_decode_error_handling(self, mock_jsondecode): """Test handling of JSON decode errors in POST method""" mock_jsondecode.side_effect = ValueError("Invalid JSON") token = "invalid json" provider = CallbackProvider(self.id, token) # This should raise an exception when trying to decode invalid JSON with self.assertRaises(ValueError): provider.set_record("example.com", "192.168.1.1") class TestCallbackProviderRealIntegration(BaseProviderTestCase): """Real integration tests for CallbackProvider using httpbin.org""" def setUp(self): """Set up real test fixtures and skip on unsupported CI environments""" super(TestCallbackProviderRealIntegration, self).setUp() # Skip on Python 3.10/3.13 or 32bit in CI is_ci = os.environ.get("CI") or os.environ.get("GITHUB_ACTIONS") or os.environ.get("GITHUB_REF_NAME") pyver = sys.version_info sys_platform = sys.platform.lower() machine = platform.machine().lower() is_mac = sys_platform == "darwin" # On macOS CI, require arm64; on others, require amd64/x86_64 if is_ci: if is_mac: if not ("arm" in machine or "aarch64" in machine): self.skipTest("On macOS CI, only arm64 is supported for integration tests.") else: if not ("amd64" in machine or "x86_64" in machine): self.skipTest("On non-macOS CI, only amd64/x86_64 is supported for integration tests.") if pyver[:2] in [(3, 10), (3, 13)] or platform.architecture()[0] == "32bit": self.skipTest("Skip real HTTP integration on CI for Python 3.10/3.13 or 32bit platform") def _setup_provider_with_mock_logger(self, provider): """Helper method to setup provider with a mock logger.""" mock_logger = self.mock_logger(provider) # Ensure the logger is configured to capture info calls mock_logger.setLevel(logging.INFO) return mock_logger def _random_delay(self): """Add a random delay of 0-3 seconds to avoid rate limiting""" if os.environ.get("CI") or os.environ.get("GITHUB_ACTIONS") or os.environ.get("GITHUB_REF_NAME"): # In CI environments, use a shorter delay to speed up tests delay = random.uniform(0, 2) else: delay = random.uniform(0, 1) sleep(delay) def _assert_callback_result_logged(self, mock_logger, *expected_strings): """ Helper to assert that 'Callback result: %s' was logged with expected content. """ info_calls = mock_logger.info.call_args_list response_logged = False for call in info_calls: if len(call[0]) >= 2 and call[0][0] == "Callback result: %s": response_content = str(call[0][1]) # Check if the response contains the expected strings if all(expected in response_content for expected in expected_strings): response_logged = True break # Also check if this is a firewall/network blocking response blocking_keywords = ["firewall", "deny", "blocked", "policy", "rule"] if any(keyword.lower() in response_content.lower() for keyword in blocking_keywords): # Skip test if network is blocked raise unittest.SkipTest("Network request blocked by firewall/policy: {}".format(response_content)) self.assertTrue( response_logged, "Expected logger.info to log 'Callback result' containing: {}".format(", ".join(expected_strings)), ) def test_real_callback_get_method(self): """Test real callback using GET method with httpbin/httpbingo and verify logger calls""" # 尝试多个测试端点以提高可靠性 test_endpoints = [ "http://httpbin.org/get?domain=__DOMAIN__&ip=__IP__&record_type=__RECORDTYPE__", "http://httpbingo.org/get?domain=__DOMAIN__&ip=__IP__&record_type=__RECORDTYPE__", ] domain = "test.example.com" ip = "111.111.111.111" last_exception = None for endpoint_id in test_endpoints: try: provider = CallbackProvider(endpoint_id, "", ssl="auto") mock_logger = self._setup_provider_with_mock_logger(provider) self._random_delay() # Add random delay before real request result = provider.set_record(domain, ip, "A") if result: self.assertTrue(result) self._assert_callback_result_logged(mock_logger, domain, ip) return # 成功则退出 else: # 如果结果为False，可能是5xx错误，尝试下一个端点 continue except Exception as e: last_exception = e # 网络问题时继续尝试下一个端点 error_msg = str(e).lower() network_keywords = [ "timeout", "connection", "resolution", "unreachable", "network", "ssl", "certificate", ] if any(keyword in error_msg for keyword in network_keywords): continue # 尝试下一个端点 else: # 其他异常重新抛出 raise # 如果所有端点都失败，跳过测试 error_info = " - Last error: {}".format(str(last_exception)) if last_exception else "" self.skipTest("All network endpoints unavailable for GET callback test{}".format(error_info)) def test_real_callback_post_method_with_json(self): """Test real callback using POST method with JSON data and verify logger calls""" # 尝试多个测试端点以提高可靠性 test_endpoints = ["http://httpbingo.org/post", "http://httpbin.org/post"] token = '{"domain": "__DOMAIN__", "ip": "__IP__", "record_type": "__RECORDTYPE__", "ttl": "__TTL__"}' domain = "test.example.com" ip = "203.0.113.2" last_exception = None for endpoint_id in test_endpoints: try: provider = CallbackProvider(endpoint_id, token) # Setup provider with mock logger mock_logger = self._setup_provider_with_mock_logger(provider) self._random_delay() # Add random delay before real request result = provider.set_record(domain, ip, "A", 300) if result: # httpbin/httpbingo returns JSON with our posted data, so it should be truthy self.assertTrue(result) # Verify that logger.info was called with response containing domain and IP self._assert_callback_result_logged(mock_logger, domain, ip) return # 成功则退出 else: # 如果结果为False，可能是5xx错误，尝试下一个端点 continue except Exception as e: last_exception = e # 网络问题时继续尝试下一个端点 error_msg = str(e).lower() network_keywords = [ "timeout", "connection", "resolution", "unreachable", "network", "ssl", "certificate", ] if any(keyword in error_msg for keyword in network_keywords): continue # 尝试下一个端点 else: # 其他异常重新抛出 raise # 如果所有端点都失败，跳过测试 error_info = " - Last error: {}".format(str(last_exception)) if last_exception else "" self.skipTest("All network endpoints unavailable for POST callback test{}".format(error_info)) def test_real_callback_error_handling(self): """Test real callback error handling with invalid URL""" # Use an invalid URL to test error handling id = "http://postman-echo.com/status/400" # This returns HTTP 400 provider = CallbackProvider(id, "") self._random_delay() # Add random delay before real request result = provider.set_record("test.example.com", "203.0.113.5") self.assertFalse(result) if __name__ == "__main__": unittest.main()