From 81840212e066393df4025c7e7acc123744ea79cd Mon Sep 17 00:00:00 2001 From: iclart Date: Tue, 27 Jan 2026 13:58:29 +0800 Subject: [PATCH] feat(provider.cloudns): add ClouDNS.net DNS provider support (#663) --- ddns/config/cli.py | 1 + ddns/provider/__init__.py | 3 + ddns/provider/cloudns.py | 156 +++++++++++++++++++++ tests/test_provider_cloudns.py | 242 +++++++++++++++++++++++++++++++++ 4 files changed, 402 insertions(+) create mode 100644 ddns/provider/cloudns.py create mode 100644 tests/test_provider_cloudns.py diff --git a/ddns/config/cli.py b/ddns/config/cli.py index 5ced083..00f3fc6 100644 --- a/ddns/config/cli.py +++ b/ddns/config/cli.py @@ -125,6 +125,7 @@ def _add_ddns_args(arg): # type: (ArgumentParser) -> None "aliesa", "callback", "cloudflare", + "cloudns", "debug", "dnscom", "dnspod_com", diff --git a/ddns/provider/__init__.py b/ddns/provider/__init__.py index f69df62..3e1c6d8 100644 --- a/ddns/provider/__init__.py +++ b/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 diff --git a/ddns/provider/cloudns.py b/ddns/provider/cloudns.py new file mode 100644 index 0000000..c4908aa --- /dev/null +++ b/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 diff --git a/tests/test_provider_cloudns.py b/tests/test_provider_cloudns.py new file mode 100644 index 0000000..96929d1 --- /dev/null +++ b/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()