feat(provider.cloudns): add ClouDNS.net DNS provider support (#663)

ddns/config/cli.py

@@ -125,6 +125,7 @@ def _add_ddns_args(arg):  # type: (ArgumentParser) -> None
             "aliesa",
             "callback",
             "cloudflare",
+            "cloudns",
             "debug",
             "dnscom",
             "dnspod_com",

ddns/provider/__init__.py

@@ -4,6 +4,7 @@ from .alidns import AlidnsProvider
 from .aliesa import AliesaProvider
 from .callback import CallbackProvider
 from .cloudflare import CloudflareProvider
+from .cloudns import CloudnsProvider
 from .debug import DebugProvider
 from .dnscom import DnscomProvider
 from .dnspod import DnspodProvider
@@ -50,6 +51,8 @@ def get_provider_class(provider_name):
         "edgeone_noacc": EdgeOneDnsProvider,  # 非加速域名
         # cloudflare
         "cloudflare": CloudflareProvider,
+        # cloudns
+        "cloudns": CloudnsProvider,
         # aliyun alidns
         "alidns": AlidnsProvider,
         "aliyun": AlidnsProvider,  # 兼容aliyun

ddns/provider/cloudns.py

@@ -0,0 +1,156 @@
+# coding=utf-8
+"""
+ClouDNS API
+@author: NewFuture (modifications for ClouDNS)
+"""
+
+from ._base import BaseProvider, TYPE_FORM
+
+
+class CloudnsProvider(BaseProvider):
+    """
+    ClouDNS API
+    ClouDNS 接口解析操作库
+    """
+
+    endpoint = "https://api.cloudns.net"
+    content_type = TYPE_FORM
+
+    def _request(self, action, **params):
+        """
+        发送请求数据，自动添加认证参数
+
+        Send request to ClouDNS API.
+        """
+        # Add authentication
+        params["auth-id"] = self.id
+        params["auth-password"] = self.token
+
+        # Filter None values
+        params = {k: v for k, v in params.items() if v is not None}
+
+        data = self._http("POST", action, body=params)
+
+        # CloudNS API returns different formats:
+        # - Success for modifications: {"status": "Success", ...}
+        # - Failure: {"status": "Failed", "statusDescription": "..."}
+        # - Query success: raw data (e.g., {"id1": {...}, "id2": {...}}) without status field
+
+        # Check for explicit error status
+        if data and isinstance(data, dict):
+            if data.get("status") == "Failed":
+                error_msg = data.get("statusDescription", "Unknown error")
+                self.logger.warning("ClouDNS API error: %s", error_msg)
+                return None
+            # If status is "Success" or no status field (raw data), return as-is
+            return data
+
+        return data
+
+    def _query_zone_id(self, domain):
+        """
+        ClouDNS uses domain name directly as zone identifier
+        """
+        # ClouDNS uses domain-name directly, no numeric zone ID
+        # Return the domain itself as the zone identifier
+        return domain
+
+    def _query_record(self, zone_id, subdomain, main_domain, record_type, line, extra):
+        """
+        Query DNS records
+        https://www.cloudns.net/wiki/article/57/
+        """
+        # For @ (root) records, ClouDNS uses empty string or "@"
+        host = subdomain if subdomain != "@" else ""
+
+        params = {
+            "domain-name": zone_id,
+            "host": host,
+            "type": record_type
+        }
+
+        data = self._request("/dns/records.json", **params)
+
+        if data and isinstance(data, dict):
+            # CloudNS returns records as a dict with record IDs as keys
+            # Format: {"id1": {record_data}, "id2": {record_data}}
+            # Or error format: {"status": "Failed", ...}
+
+            # Check if it's an error response
+            if "status" in data:
+                return None
+
+            # Iterate through records (values of the dict)
+            for record in data.values():
+                # Match host - empty string means root
+                record_host = record.get("host", "")
+                if (record_host == host or
+                    (record_host == "@" and subdomain == "@") or
+                    (record_host == "" and subdomain == "@")):
+                    if record.get("type") == record_type:
+                        return record
+
+        return None
+
+    def _create_record(self, zone_id, subdomain, main_domain, value, record_type, ttl, line, extra):
+        """
+        Create new DNS record
+        https://www.cloudns.net/wiki/article/58/
+        """
+        # Default TTL if not specified
+        if ttl is None:
+            ttl = 3600  # 1 hour default
+
+        # For @ (root) records
+        host = subdomain if subdomain != "@" else ""
+
+        params = {
+            "domain-name": zone_id,
+            "record-type": record_type,
+            "host": host,
+            "record": value,
+            "ttl": ttl
+        }
+
+        data = self._request("/dns/add-record.json", **params)
+
+        if data and data.get("status") == "Success":
+            self.logger.info("Record created successfully")
+            return True
+        else:
+            self.logger.error("Failed to create record: %s", data)
+            return False
+
+    def _update_record(self, zone_id, old_record, value, record_type, ttl, line, extra):
+        """
+        Update existing DNS record
+        https://www.cloudns.net/wiki/article/60/
+        """
+        # Default TTL if not specified
+        if ttl is None:
+            ttl = 3600
+
+        record_id = old_record.get("id")
+        if not record_id:
+            self.logger.error("Record ID not found in old_record")
+            return False
+
+        # Get original host
+        host = old_record.get("host", "")
+
+        params = {
+            "domain-name": zone_id,
+            "record-id": record_id,
+            "host": host,
+            "record": value,
+            "ttl": ttl
+        }
+
+        data = self._request("/dns/mod-record.json", **params)
+
+        if data and data.get("status") == "Success":
+            self.logger.info("Record updated successfully")
+            return True
+        else:
+            self.logger.error("Failed to update record: %s", data)
+            return False

tests/test_provider_cloudns.py

@@ -0,0 +1,242 @@
+# coding=utf-8
+"""
+Unit tests for CloudnsProvider
+
+@author: Claude
+"""
+
+from base_test import BaseProviderTestCase, patch, unittest
+
+from ddns.provider.cloudns import CloudnsProvider
+
+
+class TestCloudnsProvider(BaseProviderTestCase):
+    """Test cases for CloudnsProvider"""
+
+    def setUp(self):
+        """Set up test fixtures"""
+        super(TestCloudnsProvider, self).setUp()
+        self.id = "1234"
+        self.token = "test_password"
+
+    def test_class_constants(self):
+        """Test CloudnsProvider class constants"""
+        self.assertEqual(CloudnsProvider.endpoint, "https://api.cloudns.net")
+        self.assertEqual(CloudnsProvider.content_type, "application/x-www-form-urlencoded")
+        self.assertTrue(CloudnsProvider.decode_response)
+
+    def test_init(self):
+        """Test CloudnsProvider initialization"""
+        provider = CloudnsProvider(self.id, self.token)
+        self.assertEqual(provider.id, self.id)
+        self.assertEqual(provider.token, self.token)
+        self.assertEqual(provider.endpoint, "https://api.cloudns.net")
+
+    def test_query_zone_id(self):
+        """Test _query_zone_id returns domain itself"""
+        provider = CloudnsProvider(self.id, self.token)
+        result = provider._query_zone_id("example.com")
+        self.assertEqual(result, "example.com")
+
+    def test_query_record_success(self):
+        """Test _query_record method with successful response"""
+        provider = CloudnsProvider(self.id, self.token)
+
+        with patch.object(provider, "_http") as mock_http:
+            # CloudNS returns records as dict with record IDs as keys
+            mock_http.return_value = {
+                "12345": {"id": "12345", "host": "www", "type": "A", "record": "1.2.3.4", "ttl": "3600"}
+            }
+
+            result = provider._query_record("example.com", "www", "example.com", "A", None, {})
+
+            self.assertIsNotNone(result)
+            self.assertEqual(result["id"], "12345")
+            self.assertEqual(result["host"], "www")
+
+    def test_query_record_root(self):
+        """Test _query_record for root (@) records"""
+        provider = CloudnsProvider(self.id, self.token)
+
+        with patch.object(provider, "_http") as mock_http:
+            mock_http.return_value = {
+                "12345": {"id": "12345", "host": "@", "type": "A", "record": "1.2.3.4", "ttl": "3600"}
+            }
+
+            result = provider._query_record("example.com", "@", "example.com", "A", None, {})
+
+            self.assertIsNotNone(result)
+            self.assertEqual(result["id"], "12345")
+
+    def test_query_record_not_found(self):
+        """Test _query_record when no matching record is found"""
+        provider = CloudnsProvider(self.id, self.token)
+
+        with patch.object(provider, "_http") as mock_http:
+            # Empty dict means no records found
+            mock_http.return_value = {}
+
+            result = provider._query_record("example.com", "www", "example.com", "A", None, {})
+
+            self.assertIsNone(result)
+
+    def test_create_record_success(self):
+        """Test _create_record method with successful creation"""
+        provider = CloudnsProvider(self.id, self.token)
+
+        with patch.object(provider, "_http") as mock_http:
+            mock_http.return_value = {
+                "status": "Success",
+                "statusDescription": "Successfully added"
+            }
+
+            result = provider._create_record("example.com", "www", "example.com", "1.2.3.4", "A", 3600, None, {})
+
+            self.assertTrue(result)
+
+    def test_create_record_default_ttl(self):
+        """Test _create_record uses default TTL when None is provided"""
+        provider = CloudnsProvider(self.id, self.token)
+
+        with patch.object(provider, "_http") as mock_http:
+            mock_http.return_value = {
+                "status": "Success"
+            }
+
+            provider._create_record("example.com", "www", "example.com", "1.2.3.4", "A", None, None, {})
+
+            # Verify default TTL was used
+            call_args = mock_http.call_args
+            body = call_args[1]["body"]
+            self.assertEqual(body["ttl"], 3600)
+
+    def test_create_record_failure(self):
+        """Test _create_record method with failed creation"""
+        provider = CloudnsProvider(self.id, self.token)
+
+        with patch.object(provider, "_http") as mock_http:
+            mock_http.return_value = {
+                "status": "Failed",
+                "statusDescription": "Invalid authentication"
+            }
+
+            result = provider._create_record("example.com", "www", "example.com", "1.2.3.4", "A", 3600, None, {})
+
+            self.assertFalse(result)
+
+    def test_update_record_success(self):
+        """Test _update_record method with successful update"""
+        provider = CloudnsProvider(self.id, self.token)
+
+        old_record = {"id": "12345", "host": "www", "type": "A", "record": "5.6.7.8"}
+
+        with patch.object(provider, "_http") as mock_http:
+            mock_http.return_value = {
+                "status": "Success",
+                "statusDescription": "Successfully updated"
+            }
+
+            result = provider._update_record("example.com", old_record, "1.2.3.4", "A", 3600, None, {})
+
+            self.assertTrue(result)
+
+    def test_update_record_missing_id(self):
+        """Test _update_record when old_record has no id"""
+        provider = CloudnsProvider(self.id, self.token)
+
+        old_record = {"host": "www", "type": "A"}  # No id field
+
+        result = provider._update_record("example.com", old_record, "1.2.3.4", "A", 3600, None, {})
+
+        self.assertFalse(result)
+
+    def test_update_record_default_ttl(self):
+        """Test _update_record uses default TTL when None is provided"""
+        provider = CloudnsProvider(self.id, self.token)
+
+        old_record = {"id": "12345", "host": "www"}
+
+        with patch.object(provider, "_http") as mock_http:
+            mock_http.return_value = {
+                "status": "Success"
+            }
+
+            provider._update_record("example.com", old_record, "1.2.3.4", "A", None, None, {})
+
+            # Verify default TTL was used
+            call_args = mock_http.call_args
+            body = call_args[1]["body"]
+            self.assertEqual(body["ttl"], 3600)
+
+    def test_request_adds_auth(self):
+        """Test _request method adds authentication parameters"""
+        provider = CloudnsProvider(self.id, self.token)
+
+        with patch.object(provider, "_http") as mock_http:
+            mock_http.return_value = {"status": "Success"}
+
+            provider._request("/dns/records.json", **{"domain-name": "example.com"})
+
+            call_args = mock_http.call_args
+            body = call_args[1]["body"]
+            self.assertEqual(body["auth-id"], self.id)
+            self.assertEqual(body["auth-password"], self.token)
+
+
+class TestCloudnsProviderIntegration(BaseProviderTestCase):
+    """Integration test cases for CloudnsProvider - testing with minimal mocking"""
+
+    def setUp(self):
+        """Set up test fixtures"""
+        super(TestCloudnsProviderIntegration, self).setUp()
+        self.id = "1234"
+        self.token = "test_password"
+
+    def test_full_workflow_create_new_record(self):
+        """Test complete workflow for creating a new record"""
+        provider = CloudnsProvider(self.id, self.token)
+
+        with patch.object(provider, "_request") as mock_request:
+            # Simulate API responses
+            mock_request.side_effect = [
+                {},  # No existing record (empty dict)
+                {"status": "Success"}  # Record created
+            ]
+
+            result = provider.set_record("www.example.com", "1.2.3.4", "A", 3600)
+
+            self.assertTrue(result)
+
+    def test_full_workflow_update_existing_record(self):
+        """Test complete workflow for updating an existing record"""
+        provider = CloudnsProvider(self.id, self.token)
+
+        with patch.object(provider, "_request") as mock_request:
+            # Simulate API responses
+            existing_record = {"id": "12345", "host": "www", "type": "A", "record": "5.6.7.8"}
+            mock_request.side_effect = [
+                {"12345": existing_record},  # Existing record found
+                {"status": "Success"}  # Record updated
+            ]
+
+            result = provider.set_record("www.example.com", "1.2.3.4", "A", 3600)
+
+            self.assertTrue(result)
+
+    def test_full_workflow_root_record(self):
+        """Test workflow for root (@) record"""
+        provider = CloudnsProvider(self.id, self.token)
+
+        with patch.object(provider, "_request") as mock_request:
+            mock_request.side_effect = [
+                {},  # No existing record (empty dict)
+                {"status": "Success"}  # Record created
+            ]
+
+            result = provider.set_record("example.com", "1.2.3.4", "A", 3600)
+
+            self.assertTrue(result)
+
+
+if __name__ == "__main__":
+    unittest.main()