DDNS/tests/test_util_http_retry.py

# coding=utf-8
"""
测试 HTTP RetryHandler 重试功能
Test HTTP RetryHandler retry functionality
"""

from __future__ import unicode_literals
import socket
import ssl
from threading import Thread
from __init__ import unittest, patch, MagicMock
import logging

# Python 2/3 compatibility
try:
    from http.server import BaseHTTPRequestHandler, HTTPServer
    from io import StringIO
    from urllib.error import URLError
except ImportError:
    from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
    from StringIO import StringIO  # type: ignore[no-redef]
    from urllib2 import URLError  # type: ignore[no-redef]

from ddns.util.http import RetryHandler, request


class _RetryStatusHandler(BaseHTTPRequestHandler):
    """返回可重试状态码的本地HTTP处理器"""

    request_count = 0

    def do_GET(self):
        _RetryStatusHandler.request_count += 1
        self.send_response(502)
        self.send_header("Content-Length", "0")
        self.end_headers()

    def log_message(self, format, *args):
        pass


class TestRetryHandler(unittest.TestCase):
    """测试 RetryHandler 类"""

    def setUp(self):
        """设置测试，创建retries=2的RetryHandler（会重试2次，总共最多3次请求）"""
        self.retry_handler = RetryHandler(retries=2)

    def test_init_default(self):
        """测试默认初始化"""
        handler = RetryHandler()
        self.assertEqual(handler.retries, 3)
        self.assertEqual(handler.RETRY_CODES, (408, 429, 500, 502, 503, 504))

    def test_init_custom(self):
        """测试自定义初始化"""
        handler = RetryHandler(retries=5)
        self.assertEqual(handler.retries, 5)
        self.assertEqual(handler.RETRY_CODES, (408, 429, 500, 502, 503, 504))

    def test_retry_codes_default(self):
        """测试默认重试状态码"""
        expected_codes = (408, 429, 500, 502, 503, 504)
        self.assertEqual(RetryHandler.RETRY_CODES, expected_codes)

    @patch("time.sleep")
    def test_network_error_retry(self, mock_sleep):
        """测试网络错误重试"""
        # 设置父级opener
        mock_parent = MagicMock()
        self.retry_handler.parent = mock_parent

        # 模拟request对象
        mock_req = MagicMock()
        mock_req.timeout = 30

        # 模拟响应
        mock_response = MagicMock()
        mock_response.getcode.return_value = 200
        mock_response.read.return_value = b"success"

        # 第一次失败，第二次成功 (retries=2，允许重试2次，总共最多3次请求)
        mock_parent.open.side_effect = [socket.timeout, mock_response]

        # 执行测试
        result = self.retry_handler._open(mock_req)

        # 验证重试次数
        self.assertEqual(mock_parent.open.call_count, 2)
        self.assertEqual(result, mock_response)

        # 验证sleep调用次数 (第一次失败后会sleep)
        self.assertEqual(mock_sleep.call_count, 1)

    @patch("time.sleep")
    def test_http_error_retry(self, mock_sleep):
        """测试HTTP错误重试"""
        # 设置父级opener
        mock_parent = MagicMock()
        self.retry_handler.parent = mock_parent

        # 模拟request对象
        mock_req = MagicMock()
        mock_req.timeout = 30

        # 模拟HTTP 500错误响应
        mock_error_response = MagicMock()
        mock_error_response.getcode.return_value = 500

        # 模拟成功响应
        mock_success_response = MagicMock()
        mock_success_response.getcode.return_value = 200

        # 第一次返回500错误，第二次成功 (retries=2，允许重试2次，总共最多3次请求)
        mock_parent.open.side_effect = [mock_error_response, mock_success_response]

        # 执行测试
        result = self.retry_handler._open(mock_req)

        # 验证重试次数
        self.assertEqual(mock_parent.open.call_count, 2)
        self.assertEqual(result, mock_success_response)

        # 验证sleep调用 (第一次失败后会sleep)
        mock_sleep.assert_called_once()

    def test_non_retryable_error_immediate_failure(self):
        """测试非重试错误立即失败"""
        # 设置父级opener
        mock_parent = MagicMock()
        self.retry_handler.parent = mock_parent

        # 模拟request对象
        mock_req = MagicMock()
        mock_req.timeout = 30

        # 模拟不可重试的异常
        mock_parent.open.side_effect = ValueError("Non-retryable error")

        # 执行测试，期望异常被抛出
        with self.assertRaises(ValueError):
            self.retry_handler._open(mock_req)

        # 验证只调用一次，没有重试
        self.assertEqual(mock_parent.open.call_count, 1)

    @patch("time.sleep")
    def test_max_retries_exceeded(self, mock_sleep):
        """测试超过最大重试次数"""
        # 设置父级opener
        mock_parent = MagicMock()
        self.retry_handler.parent = mock_parent

        # 模拟request对象
        mock_req = MagicMock()
        mock_req.timeout = 30

        # 所有请求都失败 (retries=2，会重试2次，总共3次请求: attempts 1, 2, 3)
        # 修复后的RetryHandler会正确抛出最后的异常
        mock_parent.open.side_effect = [socket.gaierror, socket.timeout, socket.timeout]

        # 执行测试，期望最后的异常被抛出
        with self.assertRaises(socket.timeout):
            self.retry_handler._open(mock_req)

        # 验证重试次数 (retries=2，总共3次请求)
        self.assertEqual(mock_parent.open.call_count, 3)

        # 验证sleep调用次数 (前两次失败后会sleep)
        self.assertEqual(mock_sleep.call_count, 2)

    def test_zero_retries_init(self):
        """测试0次重试初始化"""
        handler = RetryHandler(retries=0)
        self.assertEqual(handler.retries, 0)
        self.assertFalse(hasattr(handler, "default_open"))

    @patch("time.sleep")
    def test_zero_retries_behavior(self, mock_sleep):
        """测试0次重试时的各种情况：网络错误、HTTP错误、成功请求"""
        # 测试1: 网络错误立即失败
        handler = RetryHandler(retries=0)
        mock_parent = MagicMock()
        handler.parent = mock_parent
        mock_req = MagicMock()

        mock_parent.open.side_effect = socket.timeout("Connection timeout")
        with self.assertRaises(socket.timeout):
            handler._open(mock_req)
        self.assertEqual(mock_parent.open.call_count, 1)
        mock_sleep.assert_not_called()

        # 测试2: HTTP错误立即返回 (新handler实例)
        handler2 = RetryHandler(retries=0)
        mock_parent2 = MagicMock()
        handler2.parent = mock_parent2
        mock_req2 = MagicMock()

        mock_error_response = MagicMock()
        mock_error_response.getcode.return_value = 500
        mock_parent2.open.return_value = mock_error_response
        result = handler2._open(mock_req2)
        self.assertEqual(mock_parent2.open.call_count, 1)
        self.assertEqual(result, mock_error_response)

        # 测试3: 成功请求 (新handler实例)
        handler3 = RetryHandler(retries=0)
        mock_parent3 = MagicMock()
        handler3.parent = mock_parent3
        mock_req3 = MagicMock()

        mock_success_response = MagicMock()
        mock_success_response.getcode.return_value = 200
        mock_parent3.open.return_value = mock_success_response
        result = handler3._open(mock_req3)
        self.assertEqual(mock_parent3.open.call_count, 1)
        self.assertEqual(result, mock_success_response)


class TestRequestFunction(unittest.TestCase):
    """测试新的 request 函数"""

    @patch("ddns.util.http.build_opener")
    def test_request_with_retry(self, mock_build_opener):
        """测试带重试功能的request函数"""
        # Mock response
        mock_response = MagicMock()
        mock_response.getcode.return_value = 200
        mock_response.info.return_value = {}
        mock_response.read.return_value = b'{"success": true}'
        mock_response.msg = "OK"

        # Mock opener
        mock_opener = MagicMock()
        mock_opener.open.return_value = mock_response
        mock_build_opener.return_value = mock_opener

        # 测试调用
        result = request("GET", "http://example.com", retries=2)

        # 验证返回结果
        self.assertEqual(result.status, 200)
        self.assertEqual(result.body, '{"success": true}')

        # 验证build_opener被调用，并且包含RetryHandler
        mock_build_opener.assert_called_once()
        args = mock_build_opener.call_args[0]
        # 在Python 2中，检查实际的类名需要使用__class__.__name__
        handler_types = [getattr(handler, "__class__", type(handler)).__name__ for handler in args]
        self.assertIn("RetryHandler", handler_types)

    @patch("ddns.util.http.build_opener")
    def test_request_with_proxy(self, mock_build_opener):
        """测试request函数的代理功能"""
        mock_response = MagicMock()
        mock_response.getcode.return_value = 200
        mock_response.info.return_value = {}
        mock_response.read.return_value = b"test"
        mock_response.msg = "OK"

        mock_opener = MagicMock()
        mock_opener.open.return_value = mock_response
        mock_build_opener.return_value = mock_opener

        # 测试代理字符串
        proxy_string = "http://proxy:8080"
        result = request("GET", "http://example.com", proxies=[proxy_string])

        self.assertEqual(result.status, 200)
        mock_build_opener.assert_called_once()

    @patch("ddns.util.http.build_opener")
    def test_request_timeout_defaults_and_override(self, mock_build_opener):
        """Use method defaults unless the caller supplies a timeout."""
        mock_response = MagicMock()
        mock_response.getcode.return_value = 200
        mock_response.info.return_value = {}
        mock_response.read.return_value = b"test"
        mock_response.msg = "OK"
        mock_opener = MagicMock()
        mock_opener.open.return_value = mock_response
        mock_build_opener.return_value = mock_opener

        request("GET", "http://example.com")
        self.assertEqual(mock_opener.open.call_args[1]["timeout"], 60)

        mock_opener.open.reset_mock()
        request("POST", "http://example.com")
        self.assertEqual(mock_opener.open.call_args[1]["timeout"], 120)

        mock_opener.open.reset_mock()
        request("GET", "http://example.com", timeout=5)
        self.assertEqual(mock_opener.open.call_args[1]["timeout"], 5)

    @patch("time.sleep")
    def test_retry_handler_backoff_delays(self, mock_sleep):
        """测试 RetryHandler 的指数退避延迟"""
        # 直接测试 RetryHandler 而不是通过 request() 函数
        retry_handler = RetryHandler(retries=3)

        # 设置父级opener
        mock_parent = MagicMock()
        retry_handler.parent = mock_parent

        # 模拟request对象
        mock_req = MagicMock()
        mock_req.timeout = 30

        # 创建模拟的响应对象
        mock_response_1 = MagicMock()
        mock_response_1.getcode.return_value = 500

        mock_response_2 = MagicMock()
        mock_response_2.getcode.return_value = 500

        mock_response_3 = MagicMock()
        mock_response_3.getcode.return_value = 200

        # 设置parent.open的返回值序列：前两次500错误，第三次200成功
        mock_parent.open.side_effect = [mock_response_1, mock_response_2, mock_response_3]

        # 执行测试
        result = retry_handler._open(mock_req)

        # 验证返回成功响应
        self.assertEqual(result, mock_response_3)

        # 验证 time.sleep 被调用的次数和参数
        # 基于RetryHandler实现：attempt从1开始，延迟是2^attempt
        # 第一次失败(attempt=1)后sleep(2^1=2)，第二次失败(attempt=2)后sleep(2^2=4)
        expected_delays = [2, 4]  # 对应2^1, 2^2
        actual_delays = [call[0][0] for call in mock_sleep.call_args_list]
        self.assertEqual(actual_delays, expected_delays)

    @patch("ddns.util.http.build_opener")
    def test_default_retry_counts(self, mock_build_opener):
        """测试默认重试次数"""
        mock_response = MagicMock()
        mock_response.getcode.return_value = 200
        mock_response.info.return_value = {}
        mock_response.read.return_value = b"test"
        mock_response.msg = "OK"

        mock_opener = MagicMock()
        mock_opener.open.return_value = mock_response
        mock_build_opener.return_value = mock_opener

        # 测试默认重试次数
        request("GET", "http://example.com")

        # 验证RetryHandler被创建
        mock_build_opener.assert_called_once()
        args = mock_build_opener.call_args[0]
        handler_types = [getattr(handler, "__class__", type(handler)).__name__ for handler in args]
        self.assertIn("RetryHandler", handler_types)

        # 测试自定义重试次数
        mock_build_opener.reset_mock()
        request("GET", "http://example.com", retries=5)

        # 验证RetryHandler被创建
        mock_build_opener.assert_called_once()
        args = mock_build_opener.call_args[0]
        handler_types = [getattr(handler, "__class__", type(handler)).__name__ for handler in args]
        self.assertIn("RetryHandler", handler_types)

    @patch("ddns.util.http.build_opener")
    def test_request_with_zero_retries(self, mock_build_opener):
        """测试request函数设置0次重试"""
        mock_response = MagicMock()
        mock_response.getcode.return_value = 200
        mock_response.info.return_value = {}
        mock_response.read.return_value = b"test"
        mock_response.msg = "OK"

        mock_opener = MagicMock()
        mock_opener.open.return_value = mock_response
        mock_build_opener.return_value = mock_opener

        # 测试0次重试
        result = request("GET", "http://example.com", retries=0)

        # 验证请求成功
        self.assertEqual(result.status, 200)
        self.assertEqual(result.body, "test")

        # 验证RetryHandler被创建（即使是0次重试也会创建，但不会有default_open）
        mock_build_opener.assert_called_once()
        args = mock_build_opener.call_args[0]
        handler_types = [getattr(handler, "__class__", type(handler)).__name__ for handler in args]
        self.assertIn("RetryHandler", handler_types)


class TestHttpRetryLogging(unittest.TestCase):
    """测试HTTP重试日志，不依赖外部网络"""

    @patch("ddns.util.http.time.sleep")
    def test_http_502_retry_auto(self, mock_sleep):
        """测试HTTP 502状态码的重试机制和日志"""
        # 创建日志捕获器
        log_capture = StringIO()
        handler = logging.StreamHandler(log_capture)
        handler.setLevel(logging.WARNING)

        # 获取根logger并设置 - 这样可以捕获所有子logger的日志
        root_logger = logging.getLogger()
        original_level = root_logger.level
        original_handlers = root_logger.handlers[:]

        try:
            # 设置日志级别和处理器
            root_logger.setLevel(logging.WARNING)
            root_logger.handlers = [handler]

            _RetryStatusHandler.request_count = 0
            server = HTTPServer(("127.0.0.1", 0), _RetryStatusHandler)
            server_thread = Thread(target=server.serve_forever)
            server_thread.daemon = True
            server_thread.start()

            try:
                response = request(
                    "GET", "http://127.0.0.1:{}/".format(server.server_port), proxies=["DIRECT"], retries=1
                )
            finally:
                server.shutdown()
                server.server_close()
                server_thread.join()

            # 验证最终返回502错误
            self.assertEqual(response.status, 502)
            self.assertEqual(_RetryStatusHandler.request_count, 2)
            mock_sleep.assert_called_once_with(2)

            # 检查日志输出
            log_output = log_capture.getvalue()

            # 验证日志中包含重试信息（匹配实际的日志格式）
            self.assertIn(" retrying in 2 seconds", log_output)  # 日志中应该包含重试信息
            retry_count = log_output.count(" error, retrying in ")
            self.assertEqual(retry_count, 1, "应该有一次重试日志")
        finally:
            # 恢复原始日志设置
            root_logger.setLevel(original_level)
            root_logger.handlers = original_handlers

    @patch("ddns.util.http.time.sleep")
    def test_ssl_certificate_error_no_retry(self, mock_sleep):
        """测试直接和封装后的SSL证书错误都不触发重试"""
        errors = [ssl.SSLError("CERTIFICATE_VERIFY_FAILED"), URLError(ssl.SSLError("CERTIFICATE_VERIFY_FAILED"))]

        for error in errors:
            retry_handler = RetryHandler(retries=1)
            mock_parent = MagicMock()
            mock_parent.open.side_effect = error
            retry_handler.parent = mock_parent
            mock_req = MagicMock()
            mock_req.timeout = 60

            with self.assertRaises(type(error)):
                retry_handler._open(mock_req)

            self.assertEqual(mock_parent.open.call_count, 1)

        mock_sleep.assert_not_called()


if __name__ == "__main__":
    unittest.main()