DDNS/tests/test_provider_callback.py

# coding=utf-8
"""
Unit tests for CallbackProvider

@author: GitHub Copilot
"""

import os
import sys
import json
import logging
import random
import platform
import shutil
import tempfile
from time import sleep
from base_test import BaseProviderTestCase, is_network_error, patch, unittest
from ddns.__main__ import update_ip
from ddns.cache import Cache
from ddns.config import Config
from ddns.provider.callback import CallbackProvider
from ddns.util.http import HttpResponse


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 TestCallbackProviderHTTPResponses(BaseProviderTestCase):
    """Use real response handling and isolated on-disk caches without network requests."""

    def setUp(self):
        super(TestCallbackProviderHTTPResponses, self).setUp()
        self.temp_dir = tempfile.mkdtemp(prefix="ddns-callback-test-")
        self.addCleanup(shutil.rmtree, self.temp_dir)
        self.cache_path = os.path.join(self.temp_dir, "records.cache")
        # Keep file timestamps aligned with the clock, including on coarse Windows clocks.
        self.cache_timestamp = 1700000000
        cache_clock = patch("ddns.cache.time", return_value=self.cache_timestamp)
        cache_clock.start()
        self.addCleanup(cache_clock.stop)
        self.domain = "www.example.com"
        self.address = "192.0.2.9"
        self.url = "https://example.invalid/update?domain=__DOMAIN__&ip=__IP__"
        self.post_body = '{"domain": "__DOMAIN__", "value": "__IP__"}'

    def _sync_once(self, status, body, token):
        config = Config(
            json_config={
                "dns": "callback",
                "id": self.url,
                "token": token,
                "ipv4": [self.domain],
                "cache": self.cache_path,
            }
        )
        provider = CallbackProvider(config.id, config.token)
        self.mock_logger(provider)
        cache = Cache.new(config.cache, config.md5(), provider.logger, config.cache_max_age)
        try:
            response = HttpResponse(status, "Test response", {}, body)
            with patch("ddns.__main__.get_ip", return_value=self.address), patch(
                "ddns.provider._base.request", return_value=response
            ) as mock_request:
                result = update_ip(provider, cache, ["public"], [self.domain], "A", config)
                return result, list(mock_request.call_args_list)
        finally:
            cache.close()
            os.utime(self.cache_path, (self.cache_timestamp, self.cache_timestamp))

    def _read_cache(self):
        with open(self.cache_path, "r") as cache_file:
            return json.load(cache_file)

    def _assert_failure_retries(self, status, token):
        result, calls = self._sync_once(status, "Nonempty error response", token)
        self.assertFalse(result)
        self.assertEqual([call[0][0] for call in calls], ["POST" if token else "GET"])
        self.assertEqual(self._read_cache(), {})

        result, calls = self._sync_once(204, "", token)
        self.assertTrue(result)
        self.assertEqual(len(calls), 1)
        self.assertEqual(self._read_cache(), {self.domain + ":A": self.address})

        result, calls = self._sync_once(429, "Must not be requested after success", token)
        self.assertTrue(result)
        self.assertEqual(calls, [])

    def test_get_404_is_not_cached(self):
        self._assert_failure_retries(404, "")

    def test_post_404_is_not_cached(self):
        self._assert_failure_retries(404, self.post_body)

    def test_get_429_is_not_cached(self):
        self._assert_failure_retries(429, "")

    def test_post_429_is_not_cached(self):
        self._assert_failure_retries(429, self.post_body)

    def test_get_400_is_not_cached(self):
        self._assert_failure_retries(400, "")

    def test_post_400_is_not_cached(self):
        self._assert_failure_retries(400, self.post_body)

    def test_empty_2xx_response_is_success(self):
        provider = CallbackProvider(self.url, "")
        self.mock_logger(provider)
        for status in (200, 201, 204, 299):
            with patch(
                "ddns.provider._base.request", return_value=HttpResponse(status, "Success", {}, "")
            ) as mock_request:
                self.assertTrue(provider.set_record(self.domain, self.address))
                mock_request.assert_called_once()

    def test_other_non_2xx_responses_fail(self):
        provider = CallbackProvider(self.url, "")
        self.mock_logger(provider)
        for status in (199, 300, 302, 304, 403, 409, 500):
            with patch(
                "ddns.provider._base.request", return_value=HttpResponse(status, "Failure", {}, "Error body")
            ) as mock_request:
                self.assertFalse(provider.set_record(self.domain, self.address))
                mock_request.assert_called_once()

    def test_http_200_business_error_body_is_not_interpreted(self):
        provider = CallbackProvider(self.url, "")
        self.mock_logger(provider)
        body = '{"success": false, "error": "Custom application response"}'
        with patch("ddns.provider._base.request", return_value=HttpResponse(200, "OK", {}, body)) as mock_request:
            self.assertTrue(provider.set_record(self.domain, self.address))
            mock_request.assert_called_once()


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__",
        ]
        random.shuffle(test_endpoints)

        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")
                self.configure_test_http(provider)
                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
                if is_network_error(e, include_ssl=True):
                    continue
                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"]
        random.shuffle(test_endpoints)

        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)
                self.configure_test_http(provider)
                # 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
                if is_network_error(e, include_ssl=True):
                    continue
                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.configure_test_http(provider)

        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()