|
2 | 2 | # -*- coding: utf-8 -*- |
3 | 3 | # Filename: tests/test_config.py |
4 | 4 | # Description: Unit tests for config.py — defaults, deep merge, load/save round trip |
5 | | -# and the 0600 permissions that protect the Axle token. |
| 5 | +# the 0600 permissions that protect the Axle token, and the two |
| 6 | +# v0.2 fixes: whitespace trimming and a named ConfigError. |
6 | 7 | # Author: CliveS & Claude Opus 5 |
7 | 8 | # Date: 20-09-2026 |
8 | | -# Version: 1.0 |
| 9 | +# Version: 1.1 |
9 | 10 |
|
10 | 11 | import json |
11 | 12 | import os |
@@ -84,6 +85,112 @@ def test_a_partial_file_never_loses_a_required_key(self): |
84 | 85 | self.assertIn(key, cfg[section], f"{section}.{key} lost by the merge") |
85 | 86 |
|
86 | 87 |
|
| 88 | +class TestTrimsPastedFields(unittest.TestCase): |
| 89 | + """A token copied out of a browser routinely carries a trailing newline. |
| 90 | + Trimming happens on LOAD, not at the check — trimming only at the check would |
| 91 | + let setup report ready while the daemon still sent the untrimmed value.""" |
| 92 | + |
| 93 | + def _loaded(self, user): |
| 94 | + with tempfile.TemporaryDirectory() as d: |
| 95 | + p = os.path.join(d, "config.json") |
| 96 | + with open(p, "w", encoding="utf-8") as fh: |
| 97 | + json.dump(user, fh) |
| 98 | + return cfgmod.load_config(p)[0] |
| 99 | + |
| 100 | + def test_the_token_and_ip_are_trimmed(self): |
| 101 | + cfg = self._loaded({"inverter": {"ip": " 192.168.1.50\n"}, |
| 102 | + "vpp": {"axle_token": "\ttok "}}) |
| 103 | + self.assertEqual(cfg["inverter"]["ip"], "192.168.1.50") |
| 104 | + self.assertEqual(cfg["vpp"]["axle_token"], "tok") |
| 105 | + |
| 106 | + def test_pushover_keys_are_trimmed_too(self): |
| 107 | + cfg = self._loaded({"notify": {"pushover_token": " a\n", "pushover_user": "b "}}) |
| 108 | + self.assertEqual(cfg["notify"]["pushover_token"], "a") |
| 109 | + self.assertEqual(cfg["notify"]["pushover_user"], "b") |
| 110 | + |
| 111 | + def test_a_whitespace_only_token_becomes_empty_so_the_check_fails(self): |
| 112 | + cfg = self._loaded({"inverter": {"ip": "192.168.1.50"}, |
| 113 | + "vpp": {"axle_token": " \n"}}) |
| 114 | + self.assertEqual(cfg["vpp"]["axle_token"], "") |
| 115 | + self.assertFalse(cfgmod.is_configured(cfg)) |
| 116 | + |
| 117 | + def test_non_string_values_are_left_alone(self): |
| 118 | + cfg = self._loaded({"inverter": {"ip": 123}}) |
| 119 | + self.assertEqual(cfg["inverter"]["ip"], 123) |
| 120 | + |
| 121 | + def test_other_fields_are_not_trimmed(self): |
| 122 | + # Only the hand-pasted ones. A deliberate trailing space elsewhere is the |
| 123 | + # user's business. |
| 124 | + cfg = self._loaded({"web": {"bind": " 0.0.0.0 "}}) |
| 125 | + self.assertEqual(cfg["web"]["bind"], " 0.0.0.0 ") |
| 126 | + |
| 127 | + |
| 128 | +class TestConfigError(unittest.TestCase): |
| 129 | + """A config that exists but cannot be used must name itself and refuse to |
| 130 | + start. It must NEVER fall back to DEFAULTS — that would silently drive the |
| 131 | + inverter to a 4 kW target and a 10% reserve nobody chose.""" |
| 132 | + |
| 133 | + def _load(self, text, mode="w"): |
| 134 | + d = tempfile.mkdtemp() |
| 135 | + p = os.path.join(d, "config.json") |
| 136 | + with open(p, mode) as fh: |
| 137 | + fh.write(text) |
| 138 | + return p |
| 139 | + |
| 140 | + def test_malformed_json_names_the_file_and_the_place(self): |
| 141 | + p = self._load('{"vpp": {"axle_token": "abc",}') |
| 142 | + with self.assertRaises(cfgmod.ConfigError) as cm: |
| 143 | + cfgmod.load_config(p) |
| 144 | + msg = str(cm.exception) |
| 145 | + self.assertIn(p, msg) |
| 146 | + self.assertIn("line 1", msg) |
| 147 | + self.assertIn("--setup", msg) |
| 148 | + |
| 149 | + def test_the_original_error_is_kept_as_the_cause(self): |
| 150 | + p = self._load("{not json") |
| 151 | + with self.assertRaises(cfgmod.ConfigError) as cm: |
| 152 | + cfgmod.load_config(p) |
| 153 | + self.assertIsInstance(cm.exception.__cause__, json.JSONDecodeError) |
| 154 | + |
| 155 | + def test_valid_json_that_is_not_an_object_is_refused(self): |
| 156 | + for label, text in (("a list", "[1, 2]"), ("a string", '"hello"'), |
| 157 | + ("null", "null"), ("a number", "42")): |
| 158 | + with self.subTest(label): |
| 159 | + p = self._load(text) |
| 160 | + with self.assertRaises(cfgmod.ConfigError) as cm: |
| 161 | + cfgmod.load_config(p) |
| 162 | + self.assertIn("JSON object", str(cm.exception)) |
| 163 | + |
| 164 | + def test_a_file_that_is_not_utf8_is_refused(self): |
| 165 | + p = self._load(b"\xff\xfe{}", mode="wb") |
| 166 | + with self.assertRaises(cfgmod.ConfigError) as cm: |
| 167 | + cfgmod.load_config(p) |
| 168 | + self.assertIn("UTF-8", str(cm.exception)) |
| 169 | + |
| 170 | + def test_an_unreadable_file_is_refused_with_its_reason(self): |
| 171 | + p = self._load("{}") |
| 172 | + os.chmod(p, 0o000) |
| 173 | + self.addCleanup(os.chmod, p, 0o600) |
| 174 | + if os.geteuid() == 0: |
| 175 | + self.skipTest("running as root — permissions are not enforced") |
| 176 | + with self.assertRaises(cfgmod.ConfigError) as cm: |
| 177 | + cfgmod.load_config(p) |
| 178 | + self.assertIn(p, str(cm.exception)) |
| 179 | + |
| 180 | + def test_it_never_silently_falls_back_to_defaults(self): |
| 181 | + # THE point. A corrupt file returning (DEFAULTS, True) would start the |
| 182 | + # daemon on an export target and reserve floor the user never chose. |
| 183 | + p = self._load("{ broken") |
| 184 | + with self.assertRaises(cfgmod.ConfigError): |
| 185 | + cfgmod.load_config(p) |
| 186 | + |
| 187 | + def test_a_missing_file_is_still_not_an_error(self): |
| 188 | + with tempfile.TemporaryDirectory() as d: |
| 189 | + cfg, existed = cfgmod.load_config(os.path.join(d, "absent.json")) |
| 190 | + self.assertFalse(existed) |
| 191 | + self.assertEqual(cfg["vpp"]["export_target_kw"], 4.0) |
| 192 | + |
| 193 | + |
87 | 194 | class TestSaveConfig(unittest.TestCase): |
88 | 195 | def test_round_trip(self): |
89 | 196 | with tempfile.TemporaryDirectory() as d: |
@@ -132,10 +239,25 @@ def test_needs_both(self): |
132 | 239 | self.assertEqual(cfgmod.is_configured(self._cfg(ip, token)), want) |
133 | 240 |
|
134 | 241 | def test_whitespace_is_not_a_token(self): |
135 | | - # Documents CURRENT behaviour: a config whose token is a stray space passes |
136 | | - # is_configured() and then fails at Axle with a 401. Flagged, not silently |
137 | | - # "fixed" here — changing it is a behaviour change, not a test. |
138 | | - self.assertTrue(cfgmod.is_configured(self._cfg("192.168.1.50", " "))) |
| 242 | + # Was a real flaw until v0.2: a token of stray whitespace passed the check |
| 243 | + # and then got a 401 from Axle, so setup said ready and nothing worked. |
| 244 | + for label, ip, token in ( |
| 245 | + ("space token", "192.168.1.50", " "), |
| 246 | + ("newline token", "192.168.1.50", "\n"), |
| 247 | + ("tab ip", "\t", "tok"), |
| 248 | + ): |
| 249 | + with self.subTest(label): |
| 250 | + self.assertFalse(cfgmod.is_configured(self._cfg(ip, token))) |
| 251 | + |
| 252 | + def test_a_non_dict_section_is_not_configured(self): |
| 253 | + # A hand-edited file can replace a whole section with a scalar. Indexing |
| 254 | + # it would raise AttributeError out of a boolean check. |
| 255 | + self.assertFalse(cfgmod.is_configured({"inverter": "", "vpp": {"axle_token": "t"}})) |
| 256 | + self.assertFalse(cfgmod.is_configured({})) |
| 257 | + |
| 258 | + def test_a_non_string_value_does_not_explode(self): |
| 259 | + self.assertTrue(cfgmod.is_configured( |
| 260 | + {"inverter": {"ip": 19216815}, "vpp": {"axle_token": "tok"}})) |
139 | 261 |
|
140 | 262 |
|
141 | 263 | if __name__ == "__main__": |
|
0 commit comments