mirror of
https://github.com/NewFuture/DDNS.git
synced 2026-10-19 16:15:45 +00:00
CopilotandNewFuture
030c5cb32b
* Initial plan * docs(agents): ignore VitePress and ESA files in update-agents workflow Co-authored-by: NewFuture <6290356+NewFuture@users.noreply.github.com> * style: fix linting issues in update_agents_structure.py Co-authored-by: NewFuture <6290356+NewFuture@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: NewFuture <6290356+NewFuture@users.noreply.github.com>
230 lines
8.6 KiB
Python
230 lines
8.6 KiB
Python
# 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"], 60)
|
|
|
|
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"], 60)
|
|
|
|
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()
|