test_config_file.py 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751
  1. # coding=utf-8
  2. # type: ignore[index,operator,assignment]
  3. """
  4. Unit tests for ddns.config.file module
  5. @author: GitHub Copilot
  6. """
  7. from __future__ import unicode_literals
  8. from __init__ import unittest
  9. import tempfile
  10. import shutil
  11. import os
  12. import json
  13. import io
  14. import sys
  15. from ddns.config.file import load_config, save_config
  16. # Python 2/3 compatibility
  17. if sys.version_info[0] >= 3:
  18. from io import StringIO
  19. unicode = str
  20. else:
  21. try:
  22. from StringIO import StringIO
  23. except ImportError:
  24. from io import StringIO
  25. FileNotFoundError = globals().get("FileNotFoundError", IOError)
  26. PermissionError = globals().get("PermissionError", IOError)
  27. class TestConfigFile(unittest.TestCase):
  28. """Test cases for configuration file loading and saving"""
  29. def setUp(self):
  30. """Set up test fixtures"""
  31. self.temp_dir = tempfile.mkdtemp()
  32. self.addCleanup(shutil.rmtree, self.temp_dir, ignore_errors=True)
  33. # Capture stdout and stderr output for testing
  34. # Use unicode-compatible StringIO for Python 2/3 compatibility
  35. self.stdout_capture = StringIO()
  36. self.stderr_capture = StringIO()
  37. self.original_stdout = __import__("sys").stdout
  38. self.original_stderr = __import__("sys").stderr
  39. def tearDown(self):
  40. """Clean up after tests"""
  41. __import__("sys").stdout = self.original_stdout
  42. __import__("sys").stderr = self.original_stderr
  43. def create_test_file(self, filename, content):
  44. # type: (str, str | dict) -> str
  45. """Helper method to create a test file with given content"""
  46. file_path = os.path.join(self.temp_dir, filename)
  47. # Use io.open with utf-8 encoding for Python 2 and 3 compatibility
  48. with io.open(file_path, "w", encoding="utf-8") as f:
  49. if isinstance(content, dict):
  50. # Dump JSON with ensure_ascii=False to preserve Unicode characters
  51. f.write(json.dumps(content, indent=2, ensure_ascii=False))
  52. else:
  53. # Write content directly
  54. f.write(content)
  55. return file_path
  56. def test_load_config_json_parsing(self):
  57. """Test loading valid JSON configuration"""
  58. json_content = '{"dns": "cloudflare", "id": "[email protected]", "token": "secret123", "ttl": 300}'
  59. file_path = self.create_test_file("test.json", json_content)
  60. config = load_config(file_path)
  61. expected = {"dns": "cloudflare", "id": "[email protected]", "token": "secret123", "ttl": 300}
  62. self.assertEqual(config, expected)
  63. # Verify JSON parsing was used (no specific output message for JSON success)
  64. def test_load_config_ast_parsing(self):
  65. """Test loading valid AST (Python dict) configuration"""
  66. ast_content = '{"dns": "dnspod", "id": "test123", "token": "abc456", "ttl": 600}'
  67. file_path = self.create_test_file("test.py", ast_content)
  68. config = load_config(file_path)
  69. expected = {"dns": "dnspod", "id": "test123", "token": "abc456", "ttl": 600}
  70. self.assertEqual(config, expected)
  71. # Verify JSON parsing was used (no specific output message for JSON success)
  72. def test_load_config_json_to_ast_fallback(self):
  73. """Test fallback from JSON to AST parsing"""
  74. # Patch the stdout in the file module directly
  75. import ddns.config.file
  76. original_stdout = ddns.config.file.stdout
  77. ddns.config.file.stdout = self.stdout_capture
  78. try:
  79. # Create content that's valid Python but invalid JSON (trailing comma)
  80. python_content = '{"dns": "alidns", "id": "test", "token": "xyz",}'
  81. file_path = self.create_test_file("test.conf", python_content)
  82. config = load_config(file_path)
  83. expected = {"dns": "alidns", "id": "test", "token": "xyz"}
  84. self.assertEqual(config, expected)
  85. # Verify fallback occurred - AST success message should be in stdout
  86. stdout_output = self.stdout_capture.getvalue()
  87. self.assertIn("Successfully loaded config file with AST parser", stdout_output)
  88. finally:
  89. # Restore stdout
  90. ddns.config.file.stdout = original_stdout
  91. def test_load_config_with_arrays(self):
  92. """Test loading configuration with arrays"""
  93. json_content = """{
  94. "dns": "dnspod",
  95. "ipv4": ["example.com", "test.com"],
  96. "ipv6": ["ipv6.example.com"],
  97. "proxy": ["http://proxy1.com", "http://proxy2.com"],
  98. "index4": ["default", "custom"],
  99. "index6": ["ipv6"]
  100. }"""
  101. file_path = self.create_test_file("test_arrays.json", json_content)
  102. config = load_config(file_path)
  103. expected = {
  104. "dns": "dnspod",
  105. "ipv4": ["example.com", "test.com"],
  106. "ipv6": ["ipv6.example.com"],
  107. "proxy": ["http://proxy1.com", "http://proxy2.com"],
  108. "index4": ["default", "custom"],
  109. "index6": ["ipv6"],
  110. }
  111. self.assertEqual(config, expected)
  112. def test_load_config_with_nested_objects_flattening(self):
  113. """Test configuration loading with nested object flattening"""
  114. json_content = """{
  115. "dns": "alidns",
  116. "log": {
  117. "level": "DEBUG",
  118. "file": "/var/log/ddns.log",
  119. "format": "%(asctime)s %(message)s"
  120. },
  121. "ssl": {
  122. "verify": true,
  123. "cert_path": "/path/to/cert.pem"
  124. }
  125. }"""
  126. file_path = self.create_test_file("test_log_fields.json", json_content)
  127. config = load_config(file_path)
  128. # Check that nested objects are flattened
  129. self.assertEqual(config["dns"], "alidns")
  130. self.assertEqual(config["log_level"], "DEBUG")
  131. self.assertEqual(config["log_file"], "/var/log/ddns.log")
  132. self.assertEqual(config["log_format"], "%(asctime)s %(message)s")
  133. self.assertEqual(config["ssl_verify"], True)
  134. self.assertEqual(config["ssl_cert_path"], "/path/to/cert.pem")
  135. # Original nested objects should be replaced with flattened keys
  136. self.assertNotIn("log", config)
  137. self.assertNotIn("ssl", config)
  138. self.assertIn("log_level", config)
  139. self.assertIn("ssl_verify", config)
  140. def test_load_config_boolean_and_null_values(self):
  141. """Test loading configuration with boolean and null values"""
  142. json_content = """{
  143. "cache": true,
  144. "debug": false,
  145. "ssl": true,
  146. "verify": false,
  147. "ttl": null,
  148. "line": null,
  149. "proxy": null
  150. }"""
  151. file_path = self.create_test_file("test_types.json", json_content)
  152. config = load_config(file_path)
  153. self.assertTrue(config["cache"])
  154. self.assertFalse(config["debug"])
  155. self.assertTrue(config["ssl"])
  156. self.assertFalse(config["verify"])
  157. self.assertIsNone(config["ttl"])
  158. self.assertIsNone(config["line"])
  159. self.assertIsNone(config["proxy"])
  160. def test_load_config_special_prefixes(self):
  161. """Test loading configuration with special prefix values"""
  162. json_content = """{
  163. "dns": "cloudflare",
  164. "index4": ["regex:192\\\\.168\\\\..*", "default"],
  165. "index6": ["cmd:curl -s ipv6.icanhazip.com", "backup"]
  166. }"""
  167. file_path = self.create_test_file("test_prefixes.json", json_content)
  168. config = load_config(file_path)
  169. self.assertEqual(config["dns"], "cloudflare")
  170. self.assertEqual(config["index4"], ["regex:192\\.168\\..*", "default"])
  171. self.assertEqual(config["index6"], ["cmd:curl -s ipv6.icanhazip.com", "backup"])
  172. def test_load_config_unicode_and_special_chars(self):
  173. """Test loading configuration with unicode and special characters"""
  174. json_content = """{
  175. "dns": "test",
  176. "id": "[email protected]",
  177. "token": "password123!@#$%^&*()",
  178. "description": "Test with unicode characters"
  179. }"""
  180. file_path = self.create_test_file("test_unicode.json", json_content)
  181. config = load_config(file_path)
  182. self.assertEqual(config["id"], "[email protected]")
  183. self.assertEqual(config["token"], "password123!@#$%^&*()")
  184. self.assertEqual(config["description"], "Test with unicode characters")
  185. def test_load_config_invalid_json_invalid_ast(self):
  186. """Test loading configuration that fails both JSON and AST parsing"""
  187. # Patch the stderr in the file module directly
  188. import ddns.config.file
  189. original_stderr = ddns.config.file.stderr
  190. ddns.config.file.stderr = self.stderr_capture
  191. try:
  192. invalid_content = '{"dns": "test", invalid syntax here}'
  193. file_path = self.create_test_file("test_invalid.conf", invalid_content)
  194. with self.assertRaises((ValueError, SyntaxError)):
  195. load_config(file_path)
  196. # Verify both parsers were attempted via stderr output
  197. stderr_output = self.stderr_capture.getvalue()
  198. self.assertIn("Both JSON and AST parsing failed", stderr_output)
  199. finally:
  200. # Restore stderr
  201. ddns.config.file.stderr = original_stderr
  202. def test_load_config_ast_non_dict(self):
  203. """Test AST parsing with non-dictionary content"""
  204. # Valid Python but not a dictionary
  205. non_dict_content = '["item1", "item2", "item3"]'
  206. file_path = self.create_test_file("test_list.py", non_dict_content)
  207. with self.assertRaises(AttributeError):
  208. load_config(file_path)
  209. # Should get AttributeError when trying to call .items() on a list
  210. def test_load_config_nonexistent_file(self):
  211. """Test loading configuration from non-existent file"""
  212. nonexistent_file = os.path.join(self.temp_dir, "nonexistent.json")
  213. with self.assertRaises(Exception):
  214. load_config(nonexistent_file)
  215. def test_load_config_permission_denied(self):
  216. """Test loading configuration from file with permission denied"""
  217. # Skip this test on Windows as file permissions work differently
  218. if os.name == "nt":
  219. self.skipTest("File permission tests not reliable on Windows")
  220. # Create a file and remove read permissions
  221. file_path = self.create_test_file("test_noperm.json", '{"dns": "test"}')
  222. try:
  223. os.chmod(file_path, 0o000) # Remove all permissions
  224. with self.assertRaises(Exception):
  225. load_config(file_path)
  226. finally:
  227. # Restore permissions for cleanup
  228. try:
  229. os.chmod(file_path, 0o777)
  230. except OSError:
  231. pass
  232. def test_load_config_empty_file(self):
  233. """Test loading configuration from empty file"""
  234. file_path = self.create_test_file("test_empty.json", "")
  235. with self.assertRaises(Exception):
  236. load_config(file_path)
  237. def test_load_config_whitespace_only(self):
  238. """Test loading configuration from file with only whitespace"""
  239. file_path = self.create_test_file("test_whitespace.json", " \n\t \n ")
  240. with self.assertRaises(Exception):
  241. load_config(file_path)
  242. def test_save_config_basic(self):
  243. """Test basic configuration saving"""
  244. config_data = {"dns": "cloudflare", "id": "[email protected]", "token": "secret123"}
  245. file_path = os.path.join(self.temp_dir, "save_test.json")
  246. result = save_config(file_path, config_data)
  247. self.assertTrue(result)
  248. self.assertTrue(os.path.exists(file_path))
  249. # Verify saved content can be loaded back and contains our data
  250. loaded_config = load_config(file_path)
  251. # save_config adds default values, so we only check our specific values
  252. self.assertEqual(loaded_config["dns"], config_data["dns"])
  253. self.assertEqual(loaded_config["id"], config_data["id"])
  254. self.assertEqual(loaded_config["token"], config_data["token"])
  255. # Check that schema and other defaults are added
  256. self.assertIn("$schema", loaded_config)
  257. self.assertIn("cache", loaded_config)
  258. def test_save_config_complex_data(self):
  259. """Test saving configuration with complex data types"""
  260. config_data = {
  261. "dns": "dnspod",
  262. "ipv4": ["item1", "item2"], # Changed "arrays" to "ipv4" which is a valid config field
  263. "log_level": "DEBUG", # Use log_level instead of nested
  264. "cache": True,
  265. "proxy": [],
  266. }
  267. file_path = os.path.join(self.temp_dir, "save_complex.json")
  268. result = save_config(file_path, config_data)
  269. self.assertTrue(result)
  270. # Verify saved content contains our specific values
  271. loaded_config = load_config(file_path)
  272. self.assertEqual(loaded_config["dns"], config_data["dns"])
  273. self.assertEqual(loaded_config["ipv4"], config_data["ipv4"])
  274. self.assertEqual(loaded_config["cache"], config_data["cache"])
  275. self.assertEqual(loaded_config["proxy"], config_data["proxy"])
  276. # Check that defaults are applied when values are not provided
  277. self.assertEqual(loaded_config["ttl"], 600) # Default value when not provided
  278. def test_save_config_invalid_path(self):
  279. """Test saving configuration to invalid path"""
  280. config_data = {"dns": "test"}
  281. invalid_path = "/invalid/path/that/does/not/exist/config.json"
  282. with self.assertRaises((IOError, FileNotFoundError)):
  283. save_config(invalid_path, config_data)
  284. def test_save_config_permission_denied(self):
  285. """Test saving configuration with permission denied"""
  286. # Skip this test on Windows as directory permissions work differently
  287. if os.name == "nt":
  288. self.skipTest("Directory permission tests not reliable on Windows")
  289. config_data = {"dns": "test"}
  290. # Create a directory and remove write permissions
  291. readonly_dir = os.path.join(self.temp_dir, "readonly")
  292. os.makedirs(readonly_dir)
  293. file_path = os.path.join(readonly_dir, "config.json")
  294. try:
  295. os.chmod(readonly_dir, 0o444) # Read-only
  296. with self.assertRaises((IOError, PermissionError)):
  297. save_config(file_path, config_data)
  298. finally:
  299. # Restore permissions for cleanup
  300. try:
  301. os.chmod(readonly_dir, 0o777)
  302. except OSError:
  303. pass
  304. def test_parser_priority_json_first(self):
  305. """Test that JSON parsing is always tried first regardless of file extension"""
  306. # Test with .py extension but valid JSON content
  307. json_content = '{"dns": "cloudflare", "id": "test123"}'
  308. py_file = self.create_test_file("config.py", json_content)
  309. config = load_config(py_file)
  310. expected = {"dns": "cloudflare", "id": "test123"}
  311. self.assertEqual(config, expected)
  312. # Verify JSON parsing was used (no specific output message for JSON success)
  313. def test_parser_priority_ast_fallback(self):
  314. """Test AST parsing fallback for any file extension"""
  315. # Patch the stdout in the file module directly
  316. import ddns.config.file
  317. original_stdout = ddns.config.file.stdout
  318. ddns.config.file.stdout = self.stdout_capture
  319. try:
  320. # Create content that's definitely invalid JSON but valid Python
  321. # Use a more obvious syntax that will definitely fail JSON parsing
  322. python_content = "{'dns': 'dnspod', 'id': 'test456'}" # Single quotes - invalid JSON
  323. json_file = self.create_test_file("config.json", python_content)
  324. config = load_config(json_file)
  325. expected = {"dns": "dnspod", "id": "test456"}
  326. self.assertEqual(config, expected)
  327. # Verify fallback occurred - check stdout for AST success message
  328. stdout_output = self.stdout_capture.getvalue()
  329. # The successful parsing itself proves AST fallback worked
  330. if stdout_output:
  331. # If we have output, verify the expected message is there
  332. self.assertIn("Successfully loaded config file with AST parser", stdout_output)
  333. # The successful parsing itself proves AST fallback worked
  334. finally:
  335. # Restore stdout
  336. ddns.config.file.stdout = original_stdout
  337. def test_load_config_large_file(self):
  338. """Test loading large configuration files"""
  339. # Create a large config with many keys
  340. large_config = {"dns": "cloudflare"}
  341. for i in range(1000):
  342. large_config["key_{}".format(i)] = "value_{}".format(i)
  343. config_file = self.create_test_file("large_config.json", large_config)
  344. loaded_config = load_config(config_file)
  345. self.assertEqual(len(loaded_config), 1001) # dns + 1000 keys
  346. self.assertEqual(loaded_config["dns"], "cloudflare")
  347. self.assertEqual(loaded_config["key_999"], "value_999")
  348. def test_load_config_numeric_keys_in_nested(self):
  349. """Test nested objects with numeric keys"""
  350. config_with_numeric = {"dns": "dnspod", "servers": {"1": "primary.dns.com", "2": "secondary.dns.com"}}
  351. config_file = self.create_test_file("numeric_keys.json", config_with_numeric)
  352. loaded_config = load_config(config_file)
  353. self.assertEqual(loaded_config["dns"], "dnspod")
  354. self.assertEqual(loaded_config["servers_1"], "primary.dns.com")
  355. self.assertEqual(loaded_config["servers_2"], "secondary.dns.com")
  356. def test_load_config_special_characters_in_keys(self):
  357. """Test configuration with special characters in keys"""
  358. special_config = {"dns": "cloudflare", "nested-key": {"sub@key": "value1", "sub.key": "value2"}}
  359. config_file = self.create_test_file("special_chars.json", special_config)
  360. loaded_config = load_config(config_file)
  361. self.assertEqual(loaded_config["dns"], "cloudflare")
  362. self.assertEqual(loaded_config["nested-key_sub@key"], "value1")
  363. self.assertEqual(loaded_config["nested-key_sub.key"], "value2")
  364. def test_load_config_file_encoding_utf8(self):
  365. """Test loading files with UTF-8 encoding and special characters"""
  366. unicode_config = {"dns": "cloudflare", "description": "测试配置文件", "symbols": "αβγδε", "emoji": "🌍🔧⚡"}
  367. config_file = self.create_test_file("unicode.json", unicode_config)
  368. loaded_config = load_config(config_file)
  369. self.assertEqual(loaded_config["dns"], "cloudflare")
  370. self.assertEqual(loaded_config["description"], "测试配置文件")
  371. self.assertEqual(loaded_config["symbols"], "αβγδε")
  372. self.assertEqual(loaded_config["emoji"], "🌍🔧⚡")
  373. def test_load_config_json_with_hash_comments(self):
  374. """测试加载带有 # 注释的JSON配置文件"""
  375. json_with_comments = """{
  376. # Configuration for DDNS
  377. "dns": "cloudflare", # DNS provider
  378. "id": "[email protected]",
  379. "token": "secret123", # API token
  380. "ttl": 300
  381. # End of config
  382. }"""
  383. file_path = self.create_test_file("test_hash_comments.json", json_with_comments)
  384. config = load_config(file_path)
  385. expected = {"dns": "cloudflare", "id": "[email protected]", "token": "secret123", "ttl": 300}
  386. self.assertEqual(config, expected)
  387. def test_load_config_json_with_double_slash_comments(self):
  388. """测试加载带有 // 注释的JSON配置文件"""
  389. json_with_comments = """{
  390. // Configuration for DDNS
  391. "$schema": "https://ddns.newfuture.cc/schema/v4.0.json", // Schema validation
  392. "debug": false, // false=disable, true=enable
  393. "dns": "dnspod_com", // DNS provider
  394. "id": "1008666",
  395. "token": "ae86$cbbcctv666666666666666", // API Token
  396. "ipv4": ["test.lorzl.ml"], // IPv4 domains
  397. "ipv6": ["test.lorzl.ml"], // IPv6 domains
  398. "proxy": null // Proxy settings
  399. }"""
  400. file_path = self.create_test_file("test_double_slash_comments.json", json_with_comments)
  401. config = load_config(file_path)
  402. expected = {
  403. "$schema": "https://ddns.newfuture.cc/schema/v4.0.json",
  404. "debug": False,
  405. "dns": "dnspod_com",
  406. "id": "1008666",
  407. "token": "ae86$cbbcctv666666666666666",
  408. "ipv4": ["test.lorzl.ml"],
  409. "ipv6": ["test.lorzl.ml"],
  410. "proxy": None,
  411. }
  412. self.assertEqual(config, expected)
  413. def test_save_config_pretty_format(self):
  414. """Test that saved JSON is properly formatted"""
  415. config_data = {"dns": "cloudflare", "log_level": "DEBUG", "log_file": "/var/log/ddns.log"}
  416. save_file = os.path.join(self.temp_dir, "formatted.json")
  417. result = save_config(save_file, config_data)
  418. self.assertTrue(result)
  419. self.assertTrue(os.path.exists(save_file))
  420. # Check that the file is properly indented
  421. with open(save_file, "r") as f:
  422. content = f.read()
  423. # Should have proper indentation (2 spaces for top level, 4 for nested)
  424. self.assertIn(' "dns":', content)
  425. self.assertIn(' "level":', content) # log.level should be nested with 4 spaces
  426. def test_save_config_readonly_file(self):
  427. """Test saving to a read-only file"""
  428. readonly_file = os.path.join(self.temp_dir, "readonly.json")
  429. # Create file and make it read-only
  430. with open(readonly_file, "w") as f:
  431. f.write("{}")
  432. try:
  433. os.chmod(readonly_file, 0o444) # Read-only
  434. config_data = {"dns": "test"}
  435. with self.assertRaises((IOError, PermissionError)):
  436. save_config(readonly_file, config_data)
  437. finally:
  438. # Clean up - make writable again
  439. try:
  440. os.chmod(readonly_file, 0o777)
  441. os.remove(readonly_file)
  442. except OSError:
  443. pass
  444. def test_load_config_mixed_types_comprehensive(self):
  445. """Test loading configuration with all supported data types"""
  446. mixed_config = {
  447. "dns": "cloudflare",
  448. "ttl": 300,
  449. "cache": True,
  450. "ssl": False,
  451. "timeout": None,
  452. "servers": ["8.8.8.8", "1.1.1.1"],
  453. "weights": [0.5, 0.3, 0.2],
  454. "metadata": {"version": "1.0", "author": "test", "enabled": True},
  455. }
  456. config_file = self.create_test_file("mixed_types.json", mixed_config)
  457. loaded_config = load_config(config_file)
  458. # Test all types are preserved
  459. self.assertEqual(loaded_config["dns"], "cloudflare")
  460. self.assertEqual(loaded_config["ttl"], 300)
  461. self.assertTrue(loaded_config["cache"])
  462. self.assertFalse(loaded_config["ssl"])
  463. self.assertIsNone(loaded_config["timeout"])
  464. self.assertEqual(loaded_config["servers"], ["8.8.8.8", "1.1.1.1"])
  465. self.assertEqual(loaded_config["weights"], [0.5, 0.3, 0.2])
  466. # Test flattened nested object
  467. self.assertEqual(loaded_config["metadata_version"], "1.0")
  468. self.assertEqual(loaded_config["metadata_author"], "test")
  469. self.assertTrue(loaded_config["metadata_enabled"])
  470. def test_load_config_v41_providers_format(self):
  471. """Test loading configuration with v4.1 providers format"""
  472. config_data = {
  473. "$schema": "https://ddns.newfuture.cc/schema/v4.1.json",
  474. "ssl": "auto",
  475. "cache": True,
  476. "log": {"level": "INFO", "file": "/var/log/ddns.log"},
  477. "providers": [
  478. {
  479. "provider": "cloudflare",
  480. "id": "[email protected]",
  481. "token": "token1",
  482. "ipv4": ["test1.example.com"],
  483. "ttl": 300,
  484. },
  485. {
  486. "provider": "dnspod",
  487. "id": "[email protected]",
  488. "token": "token2",
  489. "ipv4": ["test2.example.com"],
  490. "ttl": 600,
  491. },
  492. ],
  493. }
  494. config_file = self.create_test_file("v41_providers.json", config_data)
  495. loaded_configs = load_config(config_file)
  496. # Should return a list of configs
  497. self.assertIsInstance(loaded_configs, list)
  498. self.assertEqual(len(loaded_configs), 2)
  499. # Test first provider config
  500. config1 = loaded_configs[0]
  501. self.assertEqual(config1["dns"], "cloudflare") # name mapped to dns
  502. self.assertEqual(config1["id"], "[email protected]")
  503. self.assertEqual(config1["token"], "token1")
  504. self.assertEqual(config1["ipv4"], ["test1.example.com"])
  505. self.assertEqual(config1["ttl"], 300)
  506. # Test global configs are inherited
  507. self.assertEqual(config1["ssl"], "auto")
  508. self.assertTrue(config1["cache"])
  509. self.assertEqual(config1["log_level"], "INFO")
  510. self.assertEqual(config1["log_file"], "/var/log/ddns.log")
  511. # Test second provider config
  512. config2 = loaded_configs[1]
  513. self.assertEqual(config2["dns"], "dnspod") # name mapped to dns
  514. self.assertEqual(config2["id"], "[email protected]")
  515. self.assertEqual(config2["token"], "token2")
  516. self.assertEqual(config2["ipv4"], ["test2.example.com"])
  517. self.assertEqual(config2["ttl"], 600)
  518. # Test global configs are inherited in second config too
  519. self.assertEqual(config2["ssl"], "auto")
  520. self.assertTrue(config2["cache"])
  521. self.assertEqual(config2["log_level"], "INFO")
  522. self.assertEqual(config2["log_file"], "/var/log/ddns.log")
  523. def test_load_config_v41_providers_conflict_with_dns(self):
  524. """Test loading configuration where providers and dns fields conflict"""
  525. import ddns.config.file
  526. original_stderr = ddns.config.file.stderr
  527. ddns.config.file.stderr = self.stderr_capture
  528. try:
  529. config_data = {
  530. "dns": "cloudflare", # Should conflict with providers
  531. "providers": [{"provider": "dnspod", "token": "test_token"}],
  532. }
  533. config_file = self.create_test_file("conflict.json", config_data)
  534. with self.assertRaises(ValueError) as context:
  535. load_config(config_file)
  536. self.assertIn("providers and dns fields conflict", str(context.exception))
  537. # Verify error message in stderr
  538. stderr_output = self.stderr_capture.getvalue()
  539. self.assertIn("'providers' and 'dns' fields cannot be used simultaneously", stderr_output)
  540. finally:
  541. ddns.config.file.stderr = original_stderr
  542. def test_load_config_v41_providers_missing_name(self):
  543. """Test loading configuration where provider is missing name field"""
  544. import ddns.config.file
  545. original_stderr = ddns.config.file.stderr
  546. ddns.config.file.stderr = self.stderr_capture
  547. try:
  548. config_data = {
  549. "providers": [
  550. {
  551. "id": "[email protected]",
  552. "token": "test_token",
  553. # Missing "provider" field
  554. }
  555. ]
  556. }
  557. config_file = self.create_test_file("missing_name.json", config_data)
  558. with self.assertRaises(ValueError) as context:
  559. load_config(config_file)
  560. self.assertIn("provider missing provider field", str(context.exception))
  561. # Verify error message in stderr
  562. stderr_output = self.stderr_capture.getvalue()
  563. self.assertIn("Each provider must have a 'provider' field", stderr_output)
  564. finally:
  565. ddns.config.file.stderr = original_stderr
  566. def test_load_config_v41_providers_single_provider(self):
  567. """Test loading configuration with single provider in v4.1 format"""
  568. config_data = {
  569. "ssl": False,
  570. "providers": [{"provider": "debug", "token": "dummy_token", "ipv4": ["test.example.com"]}],
  571. }
  572. config_file = self.create_test_file("single_provider.json", config_data)
  573. loaded_configs = load_config(config_file)
  574. # Should return a list with one config
  575. self.assertIsInstance(loaded_configs, list)
  576. self.assertEqual(len(loaded_configs), 1)
  577. config = loaded_configs[0]
  578. self.assertEqual(config["dns"], "debug")
  579. self.assertEqual(config["token"], "dummy_token")
  580. self.assertEqual(config["ipv4"], ["test.example.com"])
  581. self.assertFalse(config["ssl"]) # Global config inherited
  582. def test_load_config_v41_providers_with_nested_objects(self):
  583. """Test loading v4.1 providers format with nested objects in providers"""
  584. config_data = {
  585. "cache": True,
  586. "providers": [
  587. {
  588. "provider": "cloudflare",
  589. "token": "test_token",
  590. "custom": {"setting1": "value1", "setting2": "value2"},
  591. }
  592. ],
  593. }
  594. config_file = self.create_test_file("providers_nested.json", config_data)
  595. loaded_configs = load_config(config_file)
  596. self.assertIsInstance(loaded_configs, list)
  597. self.assertEqual(len(loaded_configs), 1)
  598. config = loaded_configs[0]
  599. self.assertEqual(config["dns"], "cloudflare")
  600. self.assertEqual(config["token"], "test_token")
  601. self.assertTrue(config["cache"])
  602. # Test nested objects in provider config are flattened
  603. self.assertEqual(config["custom_setting1"], "value1")
  604. self.assertEqual(config["custom_setting2"], "value2")
  605. self.assertNotIn("custom", config)
  606. if __name__ == "__main__":
  607. unittest.main()