"""Unit tests for scripts/executor_config.py (P5-M2). Run: python scripts/test_executor_config.py Covers happy path / boundary / abnormal / empty-value cases for the executor config loader and validator. exit 0 == PASS. NOTE: All strings in this file are ASCII only. """ import json import os import sys import tempfile import unittest from unittest import mock _SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__)) if _SCRIPTS_DIR not in sys.path: sys.path.insert(0, _SCRIPTS_DIR) import executor_config # noqa: E402 class ExecutorConfigTest(unittest.TestCase): """Test config load priority, resolution and validation.""" def setUp(self): self._keep = dict(os.environ) for key in ("EXECUTOR_CONFIG", "WEB_BASE_URL", "MOTORCAD_MODEL", "EXECUTOR_INSTANCES", "EXECUTOR_POLL_INTERVAL", "EXECUTOR_LOG_DIR", "EXECUTOR_LOG_LEVEL", "EXECUTOR_TOOL", "EXECUTOR_MOCK"): os.environ.pop(key, None) self._tmp = tempfile.mkdtemp(prefix="exec_cfg_test_") def tearDown(self): os.environ.clear() os.environ.update(self._keep) import shutil shutil.rmtree(self._tmp, ignore_errors=True) def _isolate(self): """Point _base_dir/repo_root at an empty temp dir so the real repo-root sidecar config cannot influence tests.""" return ( mock.patch.object(executor_config, "_base_dir", return_value=self._tmp), mock.patch.object(executor_config, "repo_root", return_value=self._tmp), ) def _write(self, name, obj): path = os.path.join(self._tmp, name) with open(path, "w", encoding="utf-8") as fh: json.dump(obj, fh) return path # ---------- happy path ---------- def test_defaults_when_no_file_and_no_env(self): with self._isolate()[0], self._isolate()[1]: cfg = executor_config.load_config() self.assertEqual(cfg["web_base_url"], "http://127.0.0.1:8000") self.assertEqual(cfg["instances"], 1) self.assertEqual(cfg["poll_interval"], 5) self.assertEqual(cfg["log_level"], "INFO") self.assertEqual(cfg["tool"], "motorcad") self.assertIs(cfg["enable_mock"], False) self.assertEqual(cfg["config_source"], "defaults") self.assertTrue(os.path.isabs(cfg["model_path"])) self.assertTrue(cfg["model_path"].startswith(self._tmp)) self.assertTrue(cfg["log_dir"].startswith(self._tmp)) def test_file_load_merges_and_resolves(self): path = self._write("executor_config.json", { "web_base_url": "http://192.168.1.10:9000", "model_path": "models/custom.mot", "instances": 3, "poll_interval": 2, "log_dir": "logs/exec", "log_level": "DEBUG", "tool": "motorcad", "enable_mock": True, }) cfg = executor_config.load_config(cli_path=path) self.assertEqual(cfg["web_base_url"], "http://192.168.1.10:9000") self.assertEqual(cfg["instances"], 3) self.assertEqual(cfg["poll_interval"], 2) self.assertEqual(cfg["log_level"], "DEBUG") self.assertIs(cfg["enable_mock"], True) self.assertEqual(cfg["config_source"], path) root = executor_config.repo_root() expected_model = os.path.normpath(os.path.join("models", "custom.mot")) self.assertTrue(cfg["model_path"].endswith(expected_model)) self.assertTrue(cfg["model_path"].startswith(root)) self.assertTrue(cfg["log_dir"].startswith(root)) def test_absolute_paths_kept(self): path = self._write("executor_config.json", { "model_path": "D:/models/abs.mot", "log_dir": "D:/logs", }) cfg = executor_config.load_config(cli_path=path) self.assertEqual(cfg["model_path"], os.path.normpath("D:/models/abs.mot")) self.assertEqual(cfg["log_dir"], os.path.normpath("D:/logs")) # ---------- env overrides ---------- def test_env_overrides_file_and_cli(self): path = self._write("executor_config.json", { "web_base_url": "http://file:8000", "instances": 2, "poll_interval": 7, }) os.environ["WEB_BASE_URL"] = "http://env:9999" os.environ["MOTORCAD_MODEL"] = "models/env.mot" os.environ["EXECUTOR_INSTANCES"] = "4" os.environ["EXECUTOR_MOCK"] = "true" cfg = executor_config.load_config(cli_path=path) self.assertEqual(cfg["web_base_url"], "http://env:9999") self.assertEqual(cfg["instances"], 4) self.assertEqual(cfg["poll_interval"], 7) # file value kept self.assertIs(cfg["enable_mock"], True) def test_env_numeric_invalid(self): os.environ["EXECUTOR_INSTANCES"] = "abc" with self.assertRaises(ValueError): executor_config.load_config() # ---------- find_config_path priority ---------- def test_find_config_path_cli_first(self): a = self._write("a.json", {"instances": 1}) b = self._write("b.json", {"instances": 1}) os.environ["EXECUTOR_CONFIG"] = b self.assertEqual(executor_config.find_config_path(cli_path=a), a) def test_find_config_path_env_fallback(self): b = self._write("b.json", {"instances": 1}) os.environ["EXECUTOR_CONFIG"] = b self.assertEqual(executor_config.find_config_path(), b) def test_find_config_path_none(self): with self._isolate()[0], self._isolate()[1]: self.assertIsNone( executor_config.find_config_path("no/such/file.json")) # ---------- boundary ---------- def test_minimal_instances_ok(self): cfg = executor_config.load_config() self.assertGreaterEqual(cfg["instances"], 1) def test_float_poll_interval_ok(self): path = self._write("executor_config.json", {"poll_interval": 0.5}) cfg = executor_config.load_config(cli_path=path) self.assertEqual(cfg["poll_interval"], 0.5) def test_empty_model_path_allowed(self): path = self._write("executor_config.json", {"model_path": ""}) cfg = executor_config.load_config(cli_path=path) self.assertIsNone(cfg["model_path"]) # ---------- abnormal ---------- def test_invalid_json_raises(self): path = os.path.join(self._tmp, "bad.json") with open(path, "w", encoding="utf-8") as fh: fh.write("{ not json !!!") with self.assertRaises(ValueError): executor_config.load_config(cli_path=path) def test_non_dict_json_raises(self): path = self._write("arr.json", [1, 2, 3]) with self.assertRaises(ValueError): executor_config.load_config(cli_path=path) def test_validate_bad_url(self): with self.assertRaises(ValueError): executor_config.validate_config({ "web_base_url": "ftp://x", "instances": 1, "poll_interval": 1, "log_level": "INFO", "tool": "motorcad", "enable_mock": False}) def test_validate_zero_instances(self): with self.assertRaises(ValueError): executor_config.validate_config({ "web_base_url": "http://x", "instances": 0, "poll_interval": 1, "log_level": "INFO", "tool": "motorcad", "enable_mock": False}) def test_validate_bad_level(self): with self.assertRaises(ValueError): executor_config.validate_config({ "web_base_url": "http://x", "instances": 1, "poll_interval": 1, "log_level": "VERBOSE", "tool": "motorcad", "enable_mock": False}) def test_validate_bad_mock_type(self): with self.assertRaises(ValueError): executor_config.validate_config({ "web_base_url": "http://x", "instances": 1, "poll_interval": 1, "log_level": "INFO", "tool": "motorcad", "enable_mock": "yes"}) # ---------- empty values ---------- def test_empty_file_object_uses_defaults(self): path = self._write("empty.json", {}) cfg = executor_config.load_config(cli_path=path) self.assertEqual(cfg["web_base_url"], "http://127.0.0.1:8000") self.assertEqual(cfg["instances"], 1) def test_null_file_field_ignored(self): path = self._write("null.json", {"web_base_url": None, "instances": 2}) cfg = executor_config.load_config(cli_path=path) self.assertEqual(cfg["web_base_url"], "http://127.0.0.1:8000") self.assertEqual(cfg["instances"], 2) def test_empty_env_ignored(self): os.environ["WEB_BASE_URL"] = "" os.environ["EXECUTOR_INSTANCES"] = "" cfg = executor_config.load_config() self.assertEqual(cfg["web_base_url"], "http://127.0.0.1:8000") self.assertEqual(cfg["instances"], 1) if __name__ == "__main__": unittest.main(verbosity=2)