DDNS/tests/test_provider_base.py

#!/usr/bin/env python
# coding=utf-8
"""
BaseProvider 单元测试
支持 Python 2.7 和 Python 3
"""

from base_test import BaseProviderTestCase, unittest
from ddns.provider._base import BaseProvider, encode_params


class _TestProvider(BaseProvider):
    """测试用的具体Provider实现"""

    endpoint = "https://api.example.com"

    def __init__(self, id="test_id", token="test_token_123456789", **options):
        super(_TestProvider, self).__init__(id, token, **options)
        self._test_zone_data = {"example.com": "zone123", "test.com": "zone456"}
        self._test_records = {}

    def _query_zone_id(self, domain):
        return self._test_zone_data.get(domain)

    def _query_record(self, zone_id, subdomain, main_domain, record_type, line=None, extra=None):
        key = "{}-{}-{}".format(zone_id, subdomain, record_type)
        return self._test_records.get(key)

    def _create_record(self, zone_id, subdomain, main_domain, value, record_type, ttl=None, line=None, extra=None):
        key = "{}-{}-{}".format(zone_id, subdomain, record_type)
        self._test_records[key] = {"id": "rec123", "name": subdomain, "value": value, "type": record_type}
        return True

    def _update_record(self, zone_id, old_record, value, record_type, ttl=None, line=None, extra=None):
        old_record["value"] = value
        return True


class TestBaseProvider(BaseProviderTestCase):
    """BaseProvider 测试类"""

    def setUp(self):
        """测试初始化"""
        super(TestBaseProvider, self).setUp()
        self.provider = _TestProvider()

    def test_init_success(self):
        """测试正常初始化"""
        provider = _TestProvider("test_id", "test_token")
        self.assertEqual(provider.id, "test_id")
        self.assertEqual(provider.token, "test_token")
        self.assertIsNotNone(provider.logger)
        self.assertEqual(provider._proxy, None)  # proxy 初始化为 None
        self.assertEqual(provider._zone_map, {})

    def test_validate_missing_id(self):
        """测试缺少id的验证"""
        with self.assertRaises(ValueError) as cm:
            _TestProvider("", "token")
        self.assertIn("id must be configured", str(cm.exception))

    def test_validate_missing_token(self):
        """测试缺少token的验证"""
        with self.assertRaises(ValueError) as cm:
            _TestProvider("id", "")
        self.assertIn("token must be configured", str(cm.exception))

    def test_init_with_endpoint_override(self):
        """测试使用endpoint参数覆盖默认API"""
        custom_endpoint = "https://custom.api.com"
        provider = _TestProvider("test_id", "test_token", endpoint=custom_endpoint)
        self.assertEqual(provider.endpoint, custom_endpoint)
        self.assertEqual(provider.id, "test_id")
        self.assertEqual(provider.token, "test_token")

    def test_init_without_endpoint_uses_default(self):
        """测试不提供endpoint时使用默认API"""
        provider = _TestProvider("test_id", "test_token")
        self.assertEqual(provider.endpoint, "https://api.example.com")  # 使用类级别的默认值
        self.assertEqual(provider.id, "test_id")
        self.assertEqual(provider.token, "test_token")

    def test_init_with_empty_endpoint_ignored(self):
        """测试空endpoint参数被忽略"""
        provider = _TestProvider("test_id", "test_token", endpoint="")
        self.assertEqual(provider.endpoint, "https://api.example.com")  # 使用类级别的默认值

        provider = _TestProvider("test_id", "test_token", endpoint=None)
        self.assertEqual(provider.endpoint, "https://api.example.com")  # 使用类级别的默认值

    def test_remark_exists_and_format(self):
        """测试remark存在且格式正确"""
        provider = _TestProvider("test_id", "test_token")
        self.assertTrue(hasattr(provider, "remark"))
        self.assertIsInstance(provider.remark, str)
        self.assertGreater(len(provider.remark), 0)
        # 检查是否包含基本的说明信息
        self.assertIn("DDNS", provider.remark)

    def test_endpoint_priority_over_class_api(self):
        """测试endpoint参数优先级高于类级别API"""

        # 创建一个有不同默认API的测试类
        class _CustomAPIProvider(_TestProvider):
            endpoint = "https://different.api.com"

        # 不使用endpoint - 应该使用类级别的API
        provider1 = _CustomAPIProvider("id", "token")
        self.assertEqual(provider1.endpoint, "https://different.api.com")

        # 使用endpoint - 应该覆盖类级别的API
        custom_endpoint = "https://override.api.com"
        provider2 = _CustomAPIProvider("id", "token", endpoint=custom_endpoint)
        self.assertEqual(provider2.endpoint, custom_endpoint)

    def test_get_zone_id_from_cache(self):
        """测试从缓存获取zone_id"""
        self.provider._zone_map["cached.com"] = "cached_zone"
        zone_id = self.provider.get_zone_id("cached.com")
        self.assertEqual(zone_id, "cached_zone")

    def test_get_zone_id_query_and_cache(self):
        """测试查询并缓存zone_id"""
        zone_id = self.provider.get_zone_id("example.com")
        self.assertEqual(zone_id, "zone123")
        self.assertEqual(self.provider._zone_map["example.com"], "zone123")

    def test_split_custom_domain_with_tilde(self):
        """测试用~分隔的自定义域名"""
        from ddns.provider._base import _split_custom_domain

        sub, main = _split_custom_domain("www~example.com")
        self.assertEqual(sub, "www")
        self.assertEqual(main, "example.com")

    def test_split_custom_domain_with_plus(self):
        """测试用+分隔的自定义域名"""
        from ddns.provider._base import _split_custom_domain

        sub, main = _split_custom_domain("api+test.com")
        self.assertEqual(sub, "api")
        self.assertEqual(main, "test.com")

    def test_split_custom_domain_no_separator(self):
        """测试没有分隔符的域名"""
        from ddns.provider._base import _split_custom_domain

        sub, main = _split_custom_domain("example.com")
        self.assertIsNone(sub)
        self.assertEqual(main, "example.com")

    def test_join_domain_normal(self):
        """测试正常合并域名"""
        from ddns.provider._base import join_domain

        domain = join_domain("www", "example.com")
        self.assertEqual(domain, "www.example.com")

    def test_join_domain_empty_sub(self):
        """测试空子域名合并"""
        from ddns.provider._base import join_domain

        domain = join_domain("", "example.com")
        self.assertEqual(domain, "example.com")

        domain = join_domain("@", "example.com")
        self.assertEqual(domain, "example.com")

    def test_encode_dict(self):
        """测试编码字典参数"""
        params = {"key1": "value1", "key2": "value2"}
        result = encode_params(params)
        # 由于字典顺序可能不同，我们检查包含关系
        self.assertIn("key1=value1", result)
        self.assertIn("key2=value2", result)

    def test_encode_none(self):
        """测试编码None参数"""
        result = encode_params(None)
        self.assertEqual(result, "")

    def test_mask_sensitive_data_empty(self):
        """测试空数据打码"""
        result = self.provider._mask_sensitive_data("")
        self.assertEqual(result, "")

        result = self.provider._mask_sensitive_data(None)
        self.assertEqual(result, None)

    def test_mask_sensitive_data_long_token(self):
        """测试长token的打码"""
        data = "token=test_token_123456789&other=value"
        result = self.provider._mask_sensitive_data(data)
        expected = "token=te***89&other=value"
        self.assertEqual(result, expected)

    def test_set_record_create(self):
        """测试创建记录"""
        result = self.provider.set_record("www~example.com", "1.2.3.4", "A")
        self.assertTrue(result)
        # 验证记录是否被创建
        record = self.provider._query_record("zone123", "www", "example.com", "A", None, {})
        self.assertIsNotNone(record)
        if record:  # Type narrowing for mypy
            self.assertEqual(record["value"], "1.2.3.4")

    def test_set_record_update_existing(self):
        """测试更新现有记录"""
        # 先创建一个记录
        self.provider.set_record("www~example.com", "1.2.3.4", "A")
        # 再更新它
        result = self.provider.set_record("www~example.com", "9.8.7.6", "A")
        self.assertTrue(result)
        record = self.provider._query_record("zone123", "www", "example.com", "A", None, {})
        if record:  # Type narrowing for mypy
            self.assertEqual(record["value"], "9.8.7.6")

    def test_set_record_invalid_domain(self):
        """测试无效域名"""
        result = self.provider.set_record("invalid.notfound", "1.2.3.4", "A")
        self.assertFalse(result)


if __name__ == "__main__":
    # 运行测试
    unittest.main(verbosity=2)