github/DDNS
4
0
Fork 0
mirror of https://github.com/NewFuture/DDNS.git synced 2026-10-19 16:15:45 +00:00
Code Issues Projects Releases Wiki Activity
Files
27f0f1c0ac79dc5ae1067c864b233ccaa9a4befe
DDNS/tests/test_ip.py
T
New FutureCopilot AppNewFuture
 27f0f1c0ac fix: correct DNS sync results and configuration retention (#739) 
* fix: correct DNS sync results and configuration retention

Keep address families independent, reject failed update responses, preserve complete IPv6 addresses, and retain Studio drafts and effective extra settings. Add offline regressions and wire Studio tests into documentation builds.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

* test(provider): make disk cache regressions deterministic

Align the cache fixture wall clock and file mtime so Windows Python 3.12 clock granularity cannot turn a freshly saved cache into a future-dated file. Keep real disk reads, all cache assertions, production expiry policy and CI matrices unchanged.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

---------

Co-authored-by: NewFuture <6290356+NewFuture@users.noreply.github.com>
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
2026-09-06 12:45:46 +08:00

437 lines
17 KiB
Python
Raw Blame History

# -*- coding:utf-8 -*-
"""
Tests for ddns.ip module including integration tests
"""
from __init__ import unittest, patch
from ddns import ip
from ddns.__main__ import get_ip
from ddns.util.http import HttpResponse
class TestIpModule(unittest.TestCase):
"""测试IP获取模块"""
def setUp(self):
"""设置测试环境"""
self.original_ssl_verify = ip.ssl_verify
def tearDown(self):
"""清理测试环境"""
ip.ssl_verify = self.original_ssl_verify
@patch("ddns.ip.request")
def test_url_v4_success(self, mock_request):
"""测试自定义URL获取IPv4 - 成功"""
# 模拟成功响应
mock_request.return_value = HttpResponse(200, "OK", {}, "1.2.3.4")
result = ip.public_v4("https://test.example.com/ip")
self.assertEqual(result, "1.2.3.4")
mock_request.assert_called_once_with("GET", "https://test.example.com/ip", verify=ip.ssl_verify, retries=2)
@patch("ddns.ip.request")
def test_url_v6_success(self, mock_request):
"""测试自定义URL获取IPv6 - 成功"""
# 模拟成功响应
mock_request.return_value = HttpResponse(200, "OK", {}, "2001:db8::1")
result = ip.public_v6("https://test.example.com/ipv6")
self.assertEqual(result, "2001:db8::1")
mock_request.assert_called_once_with("GET", "https://test.example.com/ipv6", verify=ip.ssl_verify, retries=2)
@patch("ddns.ip.request")
def test_url_v4_request_failure(self, mock_request):
"""测试自定义URL获取IPv4 - 请求失败"""
# 模拟请求异常
mock_request.side_effect = Exception("Network error")
result = ip.public_v4("https://test.example.com/ip")
self.assertIsNone(result)
mock_request.assert_called_once_with("GET", "https://test.example.com/ip", verify=ip.ssl_verify, retries=2)
@patch("ddns.ip.request")
def test_url_v4_invalid_response(self, mock_request):
"""测试自定义URL获取IPv4 - 无效响应"""
# 模拟无效响应
mock_request.return_value = HttpResponse(200, "OK", {}, "invalid response")
result = ip.public_v4("https://test.example.com/ip")
self.assertIsNone(result)
mock_request.assert_called_once_with("GET", "https://test.example.com/ip", verify=ip.ssl_verify, retries=2)
@patch("ddns.ip.request")
def test_public_v4_multiple_apis_first_success(self, mock_request):
"""测试公网IPv4获取 - 多个API第一个成功"""
# 模拟第一个API成功
mock_request.return_value = HttpResponse(200, "OK", {}, "1.2.3.4")
result = ip.public_v4()
self.assertEqual(result, "1.2.3.4")
# 应该只调用第一个API
mock_request.assert_called_once_with("GET", ip.PUBLIC_IPV4_APIS[0], verify=ip.ssl_verify, retries=2)
@patch("ddns.ip.request")
def test_public_v4_multiple_apis_fallback_success(self, mock_request):
"""测试公网IPv4获取 - 多个API第一个失败第二个成功"""
def mock_request_side_effect(method, url, **kwargs):
if url == ip.PUBLIC_IPV4_APIS[0]:
raise Exception("First API failed")
return HttpResponse(200, "OK", {}, "1.2.3.4")
mock_request.side_effect = mock_request_side_effect
result = ip.public_v4()
self.assertEqual(result, "1.2.3.4")
# 应该调用前两个API
self.assertEqual(mock_request.call_count, 2)
mock_request.assert_any_call("GET", ip.PUBLIC_IPV4_APIS[0], verify=ip.ssl_verify, retries=2)
mock_request.assert_any_call("GET", ip.PUBLIC_IPV4_APIS[1], verify=ip.ssl_verify, retries=2)
@patch("ddns.ip.request")
def test_public_v4_multiple_apis_all_fail(self, mock_request):
"""测试公网IPv4获取 - 多个API全部失败"""
# 模拟所有API都失败
mock_request.side_effect = Exception("All APIs failed")
result = ip.public_v4()
self.assertIsNone(result)
# 应该调用所有API
self.assertEqual(mock_request.call_count, len(ip.PUBLIC_IPV4_APIS))
@patch("ddns.ip.request")
def test_public_v6_multiple_apis_first_success(self, mock_request):
"""测试公网IPv6获取 - 多个API第一个成功"""
# 模拟第一个API成功
mock_request.return_value = HttpResponse(200, "OK", {}, "2001:db8::1")
result = ip.public_v6()
self.assertEqual(result, "2001:db8::1")
# 应该只调用第一个API
mock_request.assert_called_once_with("GET", ip.PUBLIC_IPV6_APIS[0], verify=ip.ssl_verify, retries=2)
@patch("ddns.ip.request")
def test_public_v6_multiple_apis_fallback_success(self, mock_request):
"""测试公网IPv6获取 - 多个API第一个失败第二个成功"""
def mock_request_side_effect(method, url, **kwargs):
if url == ip.PUBLIC_IPV6_APIS[0]:
raise Exception("First API failed")
return HttpResponse(200, "OK", {}, "2001:db8::1")
mock_request.side_effect = mock_request_side_effect
result = ip.public_v6()
self.assertEqual(result, "2001:db8::1")
# 应该调用前两个API
self.assertEqual(mock_request.call_count, 2)
mock_request.assert_any_call("GET", ip.PUBLIC_IPV6_APIS[0], verify=ip.ssl_verify, retries=2)
mock_request.assert_any_call("GET", ip.PUBLIC_IPV6_APIS[1], verify=ip.ssl_verify, retries=2)
def test_public_ipv4_apis_list_exists(self):
"""测试IPv4 API列表存在并包含所需的API"""
expected_apis = [
"https://api.ipify.org",
"https://ipv4.ddnsip.cn",
"https://ipinfo.io/ip",
"https://api-ipv4.ip.sb/ip",
"http://checkip.amazonaws.com",
]
self.assertEqual(ip.PUBLIC_IPV4_APIS, expected_apis)
def test_public_ipv6_apis_list_exists(self):
"""测试IPv6 API列表存在并包含所需的API"""
expected_apis = [
"https://api6.ipify.org/",
"https://ipv6.ddnsip.cn",
"https://api-ipv6.ip.sb/ip",
"http://ipv6.icanhazip.com",
]
self.assertEqual(ip.PUBLIC_IPV6_APIS, expected_apis)
@patch("ddns.ip.request")
def test_get_ip_public_mode_fallback(self, mock_request):
"""测试通过get_ip使用public模式的自动fallback功能"""
# 模拟第一个API失败，第二个成功
def mock_request_side_effect(method, url, **kwargs):
if url == ip.PUBLIC_IPV4_APIS[0]:
raise Exception("First API failed")
elif url == ip.PUBLIC_IPV4_APIS[1]:
return HttpResponse(200, "OK", {}, "1.2.3.4")
else:
raise Exception("Unexpected URL")
mock_request.side_effect = mock_request_side_effect
# 使用"public"规则获取IPv4地址
result = get_ip("4", ["public"])
self.assertEqual(result, "1.2.3.4")
# 应该调用了前两个API
self.assertEqual(mock_request.call_count, 2)
@patch("ddns.ip.request")
def test_get_ip_url_mode_backward_compatibility(self, mock_request):
"""测试通过get_ip使用url:模式的向后兼容性"""
# 模拟成功响应
mock_request.return_value = HttpResponse(200, "OK", {}, "1.2.3.4")
# 使用"url:"规则获取IPv4地址
result = get_ip("4", ["url:https://custom.api.com/ip"])
self.assertEqual(result, "1.2.3.4")
# 应该只调用指定的API
mock_request.assert_called_once()
args, kwargs = mock_request.call_args
self.assertEqual(args[1], "https://custom.api.com/ip")
@patch("ddns.ip.request")
def test_get_ip_multiple_url_rules_fallback(self, mock_request):
"""测试get_ip在多个URL规则之间回退"""
def mock_request_side_effect(method, url, **kwargs):
if url == "https://primary.example.com/ip":
body = "invalid response"
elif url == "https://backup.example.com/ip":
body = "1.2.3.4"
else:
raise Exception("Unexpected URL")
return HttpResponse(200, "OK", {}, body)
mock_request.side_effect = mock_request_side_effect
result = get_ip("4", ["url:https://primary.example.com/ip", "url:https://backup.example.com/ip"])
self.assertEqual(result, "1.2.3.4")
self.assertEqual(mock_request.call_count, 2)
@patch("ddns.ip.request")
def test_get_ip_multiple_rules_fallback(self, mock_request):
"""测试get_ip在规则返回空结果时继续尝试下一条"""
def mock_request_side_effect(method, url, **kwargs):
if url in ip.PUBLIC_IPV4_APIS:
body = "invalid response"
elif url == "https://backup.api.com/ip":
body = "1.2.3.4"
else:
raise Exception("Unexpected URL")
return HttpResponse(200, "OK", {}, body)
mock_request.side_effect = mock_request_side_effect
result = get_ip("4", ["public", "url:https://backup.api.com/ip"])
self.assertEqual(result, "1.2.3.4")
self.assertEqual(mock_request.call_count, len(ip.PUBLIC_IPV4_APIS) + 1)
@patch("ddns.ip.os_name", "posix")
@patch("ddns.ip.try_run")
def test_regex_v4_uses_ip_address(self, mock_try_run):
"""测试Unix IPv4正则通过参数列表执行ip命令"""
mock_try_run.return_value = "2: eth0\n inet 192.0.2.10/24 scope global eth0\n"
result = ip.regex_v4(r"192\.0\.2\..*")
self.assertEqual(result, "192.0.2.10")
mock_try_run.assert_called_once_with(["ip", "address"])
@patch("ddns.ip.os_name", "posix")
@patch("ddns.ip.try_run")
def test_regex_v6_uses_ip_address(self, mock_try_run):
"""测试Unix IPv6正则通过参数列表执行ip命令"""
mock_try_run.return_value = "2: eth0\n inet6 2409:8a00::1/64 scope global\n"
result = ip.regex_v6(r"2409:.*")
self.assertEqual(result, "2409:8a00::1")
mock_try_run.assert_called_once_with(["ip", "address"])
@patch("ddns.ip.os_name", "posix")
@patch("ddns.ip.try_run")
def test_regex_falls_back_to_ifconfig_when_ip_fails(self, mock_try_run):
"""测试ip命令失败时直接执行ifconfig"""
mock_try_run.side_effect = [None, "inet addr:198.51.100.7 Mask:255.255.255.0\n"]
result = ip.regex_v4(r"198\.51\.100\..*")
self.assertEqual(result, "198.51.100.7")
self.assertEqual(mock_try_run.call_args_list[0][0][0], ["ip", "address"])
self.assertEqual(mock_try_run.call_args_list[1][0][0], ["ifconfig"])
@patch("ddns.ip.os_name", "posix")
@patch("ddns.ip.try_run")
def test_regex_does_not_fallback_after_successful_ip_command(self, mock_try_run):
"""测试ip命令成功但不匹配时不执行ifconfig"""
mock_try_run.return_value = "2: eth0\n inet 192.0.2.10/24 scope global eth0\n"
result = ip.regex_v4(r"172\.16\..*")
self.assertIsNone(result)
mock_try_run.assert_called_once_with(["ip", "address"])
@patch("ddns.ip.os_name", "posix")
@patch("ddns.ip.try_run")
def test_regex_returns_none_when_network_commands_fail(self, mock_try_run):
"""测试ip和ifconfig均失败时返回空结果"""
mock_try_run.side_effect = [None, None]
result = ip.regex_v4(r".*")
self.assertIsNone(result)
self.assertEqual(mock_try_run.call_count, 2)
@patch("ddns.ip.os_name", "nt")
@patch("ddns.ip.try_run")
def test_regex_v4_uses_ipconfig_on_windows(self, mock_try_run):
"""测试Windows IPv4正则直接执行ipconfig"""
mock_try_run.return_value = "IPv4 Address. . . . . . . . . . . : 203.0.113.9\r\n"
result = ip.regex_v4(r"203\.0\.113\..*")
self.assertEqual(result, "203.0.113.9")
mock_try_run.assert_called_once_with(["ipconfig"])
@patch("ddns.ip.public_v4")
@patch("ddns.ip.os_name", "posix")
@patch("ddns.ip.try_run")
def test_get_ip_regex_rule_fallback(self, mock_try_run, mock_public_v4):
"""测试regex规则返回空结果时继续回退"""
mock_try_run.return_value = "inet 192.168.1.10/24\ninet 10.0.0.2/24\n"
mock_public_v4.return_value = "1.2.3.4"
result = get_ip("4", ["regex:172\\.16\\..*", "public"])
self.assertEqual(result, "1.2.3.4")
mock_public_v4.assert_called_once_with()
@patch("ddns.ip.public_v4")
@patch("ddns.__main__.check_output")
def test_get_ip_cmd_rule_fallback_on_empty_output(self, mock_check_output, mock_public_v4):
"""测试cmd规则输出为空时继续回退"""
mock_check_output.return_value = b""
mock_public_v4.return_value = "1.2.3.4"
result = get_ip("4", ["cmd:test-ip", "public"])
self.assertEqual(result, "1.2.3.4")
mock_check_output.assert_called_once_with("test-ip")
mock_public_v4.assert_called_once_with()
@patch("ddns.ip.public_v4")
@patch("ddns.__main__.check_output")
def test_get_ip_shell_rule_fallback_on_empty_output(self, mock_check_output, mock_public_v4):
"""测试shell规则输出为空时继续回退"""
mock_check_output.return_value = b""
mock_public_v4.return_value = "1.2.3.4"
result = get_ip("4", ["shell:test-ip", "public"])
self.assertEqual(result, "1.2.3.4")
mock_check_output.assert_called_once_with("test-ip", shell=True)
mock_public_v4.assert_called_once_with()
@patch("ddns.ip.request")
def test_get_ip_ipv6_rule_fallback(self, mock_request):
"""测试IPv6规则链在空结果时继续回退"""
def mock_request_side_effect(method, url, **kwargs):
if url == "https://primary.example.com/ipv6":
body = "invalid response"
elif url == "https://backup.example.com/ipv6":
body = "2001:db8::1"
else:
raise Exception("Unexpected URL")
return HttpResponse(200, "OK", {}, body)
mock_request.side_effect = mock_request_side_effect
result = get_ip("6", ["url:https://primary.example.com/ipv6", "url:https://backup.example.com/ipv6"])
self.assertEqual(result, "2001:db8::1")
self.assertEqual(mock_request.call_count, 2)
@patch("ddns.ip.request")
def test_error_response_address_is_not_used(self, mock_request):
"""Do not treat an address in an HTTP failure body as a discovery result."""
for status in (302, 403, 404, 429, 500, 503):
mock_request.return_value = HttpResponse(status, "request failed", {}, "upstream 198.51.100.20 failed")
self.assertIsNone(ip.public_v4("https://test.example.com/ip"), str(status))
@patch("ddns.ip.request")
def test_public_apis_fall_back_after_http_error(self, mock_request):
"""HTTP errors must advance to the next IPv4 or IPv6 discovery endpoint."""
for public, failed_address, expected_address in (
(ip.public_v4, "198.51.100.20", "192.0.2.10"),
(ip.public_v6, "2001:db8::bad", "2001:db8::10"),
):
mock_request.reset_mock()
mock_request.side_effect = [
HttpResponse(503, "Unavailable", {}, "upstream {} unavailable".format(failed_address)),
HttpResponse(200, "OK", {}, expected_address),
]
self.assertEqual(public(), expected_address)
self.assertEqual(mock_request.call_count, 2)
@patch("ddns.ip.request")
def test_ipv6_literals_are_not_truncated(self, mock_request):
"""Keep compressed, uncompressed, and embedded-IPv4 literals complete."""
for address in (
"::",
"::1",
"2001:db8::1",
"2001:db8:0:1:2:3:4:5",
"2001:db8::192.0.2.10",
"::ffff:192.0.2.10",
"::192.0.2.10",
"2001:db8:0:1:2:3:192.0.2.10",
"2001:db8:0:1::192.0.2.10",
"2001:db8:0:1:2::192.0.2.10",
"2001:db8::1:2:192.0.2.10",
"2001:DB8::192.0.2.10",
):
mock_request.return_value = HttpResponse(200, "OK", {}, "Address: {}.".format(address))
self.assertEqual(ip.public_v6("https://test.example.com/ip"), address)
@patch("ddns.ip.request")
def test_ipv6_compact_labels_preserve_complete_addresses(self, mock_request):
"""A text-label colon is a separator, not part of the IPv6 literal."""
for label in ("IP:", "address:"):
for address in ("2001:db8::1", "::1", "::ffff:192.0.2.10", "2001:db8::192.0.2.10"):
mock_request.return_value = HttpResponse(200, "OK", {}, label + address)
self.assertEqual(ip.public_v6("https://test.example.com/ip"), address, label + address)
@patch("ddns.ip.request")
def test_invalid_ipv6_literal_does_not_yield_a_valid_fragment(self, mock_request):
"""A malformed address must not be replaced with one of its valid-looking fragments."""
for address in (
"2001:db8::192.0.2.999",
"2001:db8:0:0:0:0:0:0:1",
"2001:db8:::1",
":2001:db8::1",
"2001:db8::1:",
":::1",
):
mock_request.return_value = HttpResponse(200, "OK", {}, address)
self.assertIsNone(ip.public_v6("https://test.example.com/ip"), address)
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink