test_config_file.py 30 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758
  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. import os
  281. # Skip this test on Windows as path creation behavior is different
  282. if os.name == "nt":
  283. self.skipTest("Path creation behavior differs on Windows")
  284. config_data = {"dns": "test"}
  285. invalid_path = "/invalid/path/that/does/not/exist/config.json"
  286. with self.assertRaises((IOError, OSError, FileNotFoundError)):
  287. save_config(invalid_path, config_data)
  288. def test_save_config_permission_denied(self):
  289. """Test saving configuration with permission denied"""
  290. # Skip this test on Windows as directory permissions work differently
  291. if os.name == "nt":
  292. self.skipTest("Directory permission tests not reliable on Windows")
  293. config_data = {"dns": "test"}
  294. # Create a directory and remove write permissions
  295. readonly_dir = os.path.join(self.temp_dir, "readonly")
  296. os.makedirs(readonly_dir)
  297. file_path = os.path.join(readonly_dir, "config.json")
  298. try:
  299. os.chmod(readonly_dir, 0o444) # Read-only
  300. with self.assertRaises((IOError, PermissionError)):
  301. save_config(file_path, config_data)
  302. finally:
  303. # Restore permissions for cleanup
  304. try:
  305. os.chmod(readonly_dir, 0o777)
  306. except OSError:
  307. pass
  308. def test_parser_priority_json_first(self):
  309. """Test that JSON parsing is always tried first regardless of file extension"""
  310. # Test with .py extension but valid JSON content
  311. json_content = '{"dns": "cloudflare", "id": "test123"}'
  312. py_file = self.create_test_file("config.py", json_content)
  313. config = load_config(py_file)
  314. expected = {"dns": "cloudflare", "id": "test123"}
  315. self.assertEqual(config, expected)
  316. # Verify JSON parsing was used (no specific output message for JSON success)
  317. def test_parser_priority_ast_fallback(self):
  318. """Test AST parsing fallback for any file extension"""
  319. # Patch the stdout in the file module directly
  320. import ddns.config.file
  321. original_stdout = ddns.config.file.stdout
  322. ddns.config.file.stdout = self.stdout_capture
  323. try:
  324. # Create content that's definitely invalid JSON but valid Python
  325. # Use a more obvious syntax that will definitely fail JSON parsing
  326. python_content = "{'dns': 'dnspod', 'id': 'test456'}" # Single quotes - invalid JSON
  327. json_file = self.create_test_file("config.json", python_content)
  328. config = load_config(json_file)
  329. expected = {"dns": "dnspod", "id": "test456"}
  330. self.assertEqual(config, expected)
  331. # Verify fallback occurred - check stdout for AST success message
  332. stdout_output = self.stdout_capture.getvalue()
  333. # The successful parsing itself proves AST fallback worked
  334. if stdout_output:
  335. # If we have output, verify the expected message is there
  336. self.assertIn("Successfully loaded config file with AST parser", stdout_output)
  337. # The successful parsing itself proves AST fallback worked
  338. finally:
  339. # Restore stdout
  340. ddns.config.file.stdout = original_stdout
  341. def test_load_config_large_file(self):
  342. """Test loading large configuration files"""
  343. # Create a large config with many keys
  344. large_config = {"dns": "cloudflare"}
  345. for i in range(1000):
  346. large_config["key_{}".format(i)] = "value_{}".format(i)
  347. config_file = self.create_test_file("large_config.json", large_config)
  348. loaded_config = load_config(config_file)
  349. self.assertEqual(len(loaded_config), 1001) # dns + 1000 keys
  350. self.assertEqual(loaded_config["dns"], "cloudflare")
  351. self.assertEqual(loaded_config["key_999"], "value_999")
  352. def test_load_config_numeric_keys_in_nested(self):
  353. """Test nested objects with numeric keys"""
  354. config_with_numeric = {"dns": "dnspod", "servers": {"1": "primary.dns.com", "2": "secondary.dns.com"}}
  355. config_file = self.create_test_file("numeric_keys.json", config_with_numeric)
  356. loaded_config = load_config(config_file)
  357. self.assertEqual(loaded_config["dns"], "dnspod")
  358. self.assertEqual(loaded_config["servers_1"], "primary.dns.com")
  359. self.assertEqual(loaded_config["servers_2"], "secondary.dns.com")
  360. def test_load_config_special_characters_in_keys(self):
  361. """Test configuration with special characters in keys"""
  362. special_config = {"dns": "cloudflare", "nested-key": {"sub@key": "value1", "sub.key": "value2"}}
  363. config_file = self.create_test_file("special_chars.json", special_config)
  364. loaded_config = load_config(config_file)
  365. self.assertEqual(loaded_config["dns"], "cloudflare")
  366. self.assertEqual(loaded_config["nested-key_sub@key"], "value1")
  367. self.assertEqual(loaded_config["nested-key_sub.key"], "value2")
  368. def test_load_config_file_encoding_utf8(self):
  369. """Test loading files with UTF-8 encoding and special characters"""
  370. unicode_config = {"dns": "cloudflare", "description": "测试配置文件", "symbols": "αβγδε", "emoji": "🌍🔧⚡"}
  371. config_file = self.create_test_file("unicode.json", unicode_config)
  372. loaded_config = load_config(config_file)
  373. self.assertEqual(loaded_config["dns"], "cloudflare")
  374. self.assertEqual(loaded_config["description"], "测试配置文件")
  375. self.assertEqual(loaded_config["symbols"], "αβγδε")
  376. self.assertEqual(loaded_config["emoji"], "🌍🔧⚡")
  377. def test_load_config_json_with_hash_comments(self):
  378. """测试加载带有 # 注释的JSON配置文件"""
  379. json_with_comments = """{
  380. # Configuration for DDNS
  381. "dns": "cloudflare", # DNS provider
  382. "id": "[email protected]",
  383. "token": "secret123", # API token
  384. "ttl": 300
  385. # End of config
  386. }"""
  387. file_path = self.create_test_file("test_hash_comments.json", json_with_comments)
  388. config = load_config(file_path)
  389. expected = {"dns": "cloudflare", "id": "[email protected]", "token": "secret123", "ttl": 300}
  390. self.assertEqual(config, expected)
  391. def test_load_config_json_with_double_slash_comments(self):
  392. """测试加载带有 // 注释的JSON配置文件"""
  393. json_with_comments = """{
  394. // Configuration for DDNS
  395. "$schema": "https://ddns.newfuture.cc/schema/v4.0.json", // Schema validation
  396. "debug": false, // false=disable, true=enable
  397. "dns": "dnspod_com", // DNS provider
  398. "id": "1008666",
  399. "token": "ae86$cbbcctv666666666666666", // API Token
  400. "ipv4": ["test.lorzl.ml"], // IPv4 domains
  401. "ipv6": ["test.lorzl.ml"], // IPv6 domains
  402. "proxy": null // Proxy settings
  403. }"""
  404. file_path = self.create_test_file("test_double_slash_comments.json", json_with_comments)
  405. config = load_config(file_path)
  406. expected = {
  407. "$schema": "https://ddns.newfuture.cc/schema/v4.0.json",
  408. "debug": False,
  409. "dns": "dnspod_com",
  410. "id": "1008666",
  411. "token": "ae86$cbbcctv666666666666666",
  412. "ipv4": ["test.lorzl.ml"],
  413. "ipv6": ["test.lorzl.ml"],
  414. "proxy": None,
  415. }
  416. self.assertEqual(config, expected)
  417. def test_save_config_pretty_format(self):
  418. """Test that saved JSON is properly formatted"""
  419. config_data = {"dns": "cloudflare", "log_level": "DEBUG", "log_file": "/var/log/ddns.log"}
  420. save_file = os.path.join(self.temp_dir, "formatted.json")
  421. result = save_config(save_file, config_data)
  422. self.assertTrue(result)
  423. self.assertTrue(os.path.exists(save_file))
  424. # Check that the file is properly indented
  425. with open(save_file, "r") as f:
  426. content = f.read()
  427. # Should have proper indentation (2 spaces for top level, 4 for nested)
  428. self.assertIn(' "dns":', content)
  429. self.assertIn(' "level":', content) # log.level should be nested with 4 spaces
  430. def test_save_config_readonly_file(self):
  431. """Test saving to a read-only file"""
  432. readonly_file = os.path.join(self.temp_dir, "readonly.json")
  433. # Create file and make it read-only
  434. with open(readonly_file, "w") as f:
  435. f.write("{}")
  436. try:
  437. os.chmod(readonly_file, 0o444) # Read-only
  438. config_data = {"dns": "test"}
  439. with self.assertRaises((IOError, PermissionError)):
  440. save_config(readonly_file, config_data)
  441. finally:
  442. # Clean up - make writable again
  443. try:
  444. os.chmod(readonly_file, 0o777)
  445. os.remove(readonly_file)
  446. except OSError:
  447. pass
  448. def test_load_config_mixed_types_comprehensive(self):
  449. """Test loading configuration with all supported data types"""
  450. mixed_config = {
  451. "dns": "cloudflare",
  452. "ttl": 300,
  453. "cache": True,
  454. "ssl": False,
  455. "timeout": None,
  456. "servers": ["8.8.8.8", "1.1.1.1"],
  457. "weights": [0.5, 0.3, 0.2],
  458. "metadata": {"version": "1.0", "author": "test", "enabled": True},
  459. }
  460. config_file = self.create_test_file("mixed_types.json", mixed_config)
  461. loaded_config = load_config(config_file)
  462. # Test all types are preserved
  463. self.assertEqual(loaded_config["dns"], "cloudflare")
  464. self.assertEqual(loaded_config["ttl"], 300)
  465. self.assertTrue(loaded_config["cache"])
  466. self.assertFalse(loaded_config["ssl"])
  467. self.assertIsNone(loaded_config["timeout"])
  468. self.assertEqual(loaded_config["servers"], ["8.8.8.8", "1.1.1.1"])
  469. self.assertEqual(loaded_config["weights"], [0.5, 0.3, 0.2])
  470. # Test flattened nested object
  471. self.assertEqual(loaded_config["metadata_version"], "1.0")
  472. self.assertEqual(loaded_config["metadata_author"], "test")
  473. self.assertTrue(loaded_config["metadata_enabled"])
  474. def test_load_config_v41_providers_format(self):
  475. """Test loading configuration with v4.1 providers format"""
  476. config_data = {
  477. "$schema": "https://ddns.newfuture.cc/schema/v4.1.json",
  478. "ssl": "auto",
  479. "cache": True,
  480. "log": {"level": "INFO", "file": "/var/log/ddns.log"},
  481. "providers": [
  482. {
  483. "provider": "cloudflare",
  484. "id": "[email protected]",
  485. "token": "token1",
  486. "ipv4": ["test1.example.com"],
  487. "ttl": 300,
  488. },
  489. {
  490. "provider": "dnspod",
  491. "id": "[email protected]",
  492. "token": "token2",
  493. "ipv4": ["test2.example.com"],
  494. "ttl": 600,
  495. },
  496. ],
  497. }
  498. config_file = self.create_test_file("v41_providers.json", config_data)
  499. loaded_configs = load_config(config_file)
  500. # Should return a list of configs
  501. self.assertIsInstance(loaded_configs, list)
  502. self.assertEqual(len(loaded_configs), 2)
  503. # Test first provider config
  504. config1 = loaded_configs[0]
  505. self.assertEqual(config1["dns"], "cloudflare") # name mapped to dns
  506. self.assertEqual(config1["id"], "[email protected]")
  507. self.assertEqual(config1["token"], "token1")
  508. self.assertEqual(config1["ipv4"], ["test1.example.com"])
  509. self.assertEqual(config1["ttl"], 300)
  510. # Test global configs are inherited
  511. self.assertEqual(config1["ssl"], "auto")
  512. self.assertTrue(config1["cache"])
  513. self.assertEqual(config1["log_level"], "INFO")
  514. self.assertEqual(config1["log_file"], "/var/log/ddns.log")
  515. # Test second provider config
  516. config2 = loaded_configs[1]
  517. self.assertEqual(config2["dns"], "dnspod") # name mapped to dns
  518. self.assertEqual(config2["id"], "[email protected]")
  519. self.assertEqual(config2["token"], "token2")
  520. self.assertEqual(config2["ipv4"], ["test2.example.com"])
  521. self.assertEqual(config2["ttl"], 600)
  522. # Test global configs are inherited in second config too
  523. self.assertEqual(config2["ssl"], "auto")
  524. self.assertTrue(config2["cache"])
  525. self.assertEqual(config2["log_level"], "INFO")
  526. self.assertEqual(config2["log_file"], "/var/log/ddns.log")
  527. def test_load_config_v41_providers_conflict_with_dns(self):
  528. """Test loading configuration where providers and dns fields conflict"""
  529. import ddns.config.file
  530. original_stderr = ddns.config.file.stderr
  531. ddns.config.file.stderr = self.stderr_capture
  532. try:
  533. config_data = {
  534. "dns": "cloudflare", # Should conflict with providers
  535. "providers": [{"provider": "dnspod", "token": "test_token"}],
  536. }
  537. config_file = self.create_test_file("conflict.json", config_data)
  538. with self.assertRaises(ValueError) as context:
  539. load_config(config_file)
  540. self.assertIn("providers and dns fields conflict", str(context.exception))
  541. # Verify error message in stderr
  542. stderr_output = self.stderr_capture.getvalue()
  543. self.assertIn("'providers' and 'dns' fields cannot be used simultaneously", stderr_output)
  544. finally:
  545. ddns.config.file.stderr = original_stderr
  546. def test_load_config_v41_providers_missing_name(self):
  547. """Test loading configuration where provider is missing name field"""
  548. import ddns.config.file
  549. original_stderr = ddns.config.file.stderr
  550. ddns.config.file.stderr = self.stderr_capture
  551. try:
  552. config_data = {
  553. "providers": [
  554. {
  555. "id": "[email protected]",
  556. "token": "test_token",
  557. # Missing "provider" field
  558. }
  559. ]
  560. }
  561. config_file = self.create_test_file("missing_name.json", config_data)
  562. with self.assertRaises(ValueError) as context:
  563. load_config(config_file)
  564. self.assertIn("provider missing provider field", str(context.exception))
  565. # Verify error message in stderr
  566. stderr_output = self.stderr_capture.getvalue()
  567. self.assertIn("Each provider must have a 'provider' field", stderr_output)
  568. finally:
  569. ddns.config.file.stderr = original_stderr
  570. def test_load_config_v41_providers_single_provider(self):
  571. """Test loading configuration with single provider in v4.1 format"""
  572. config_data = {
  573. "ssl": False,
  574. "providers": [{"provider": "debug", "token": "dummy_token", "ipv4": ["test.example.com"]}],
  575. }
  576. config_file = self.create_test_file("single_provider.json", config_data)
  577. loaded_configs = load_config(config_file)
  578. # Should return a list with one config
  579. self.assertIsInstance(loaded_configs, list)
  580. self.assertEqual(len(loaded_configs), 1)
  581. config = loaded_configs[0]
  582. self.assertEqual(config["dns"], "debug")
  583. self.assertEqual(config["token"], "dummy_token")
  584. self.assertEqual(config["ipv4"], ["test.example.com"])
  585. self.assertFalse(config["ssl"]) # Global config inherited
  586. def test_load_config_v41_providers_with_nested_objects(self):
  587. """Test loading v4.1 providers format with nested objects in providers"""
  588. config_data = {
  589. "cache": True,
  590. "providers": [
  591. {
  592. "provider": "cloudflare",
  593. "token": "test_token",
  594. "custom": {"setting1": "value1", "setting2": "value2"},
  595. }
  596. ],
  597. }
  598. config_file = self.create_test_file("providers_nested.json", config_data)
  599. loaded_configs = load_config(config_file)
  600. self.assertIsInstance(loaded_configs, list)
  601. self.assertEqual(len(loaded_configs), 1)
  602. config = loaded_configs[0]
  603. self.assertEqual(config["dns"], "cloudflare")
  604. self.assertEqual(config["token"], "test_token")
  605. self.assertTrue(config["cache"])
  606. # Test nested objects in provider config are flattened
  607. self.assertEqual(config["custom_setting1"], "value1")
  608. self.assertEqual(config["custom_setting2"], "value2")
  609. self.assertNotIn("custom", config)
  610. if __name__ == "__main__":
  611. unittest.main()