77 lines
2.8 KiB
Python
77 lines
2.8 KiB
Python
"""Config 配置模型与环境变量加载测试。"""
|
|
import _bootstrap # noqa: F401
|
|
import os
|
|
import unittest
|
|
|
|
from fault_log_analyzer.config import Config, _env, _env_bool, _env_float, _env_int
|
|
|
|
|
|
class TestEnvHelpers(unittest.TestCase):
|
|
def test_env_default(self):
|
|
self.assertEqual(_env("HMS_TEST_UNSET_XYZ", "dflt"), "dflt")
|
|
|
|
def test_env_int_parse_and_fallback(self):
|
|
os.environ["HMS_TEST_INT"] = "42"
|
|
self.assertEqual(_env_int("HMS_TEST_INT", 1), 42)
|
|
os.environ["HMS_TEST_INT"] = "abc"
|
|
self.assertEqual(_env_int("HMS_TEST_INT", 7), 7)
|
|
del os.environ["HMS_TEST_INT"]
|
|
|
|
def test_env_float_parse_and_fallback(self):
|
|
os.environ["HMS_TEST_FLOAT"] = "0.75"
|
|
self.assertAlmostEqual(_env_float("HMS_TEST_FLOAT", 1.0), 0.75)
|
|
os.environ["HMS_TEST_FLOAT"] = "x"
|
|
self.assertEqual(_env_float("HMS_TEST_FLOAT", 2.5), 2.5)
|
|
del os.environ["HMS_TEST_FLOAT"]
|
|
|
|
def test_env_bool_variants(self):
|
|
for raw in ("1", "true", "TRUE", "yes", "on"):
|
|
os.environ["HMS_TEST_BOOL"] = raw
|
|
self.assertTrue(_env_bool("HMS_TEST_BOOL", False))
|
|
for raw in ("0", "false", "no", "off"):
|
|
os.environ["HMS_TEST_BOOL"] = raw
|
|
self.assertFalse(_env_bool("HMS_TEST_BOOL", True))
|
|
del os.environ["HMS_TEST_BOOL"]
|
|
|
|
|
|
class TestConfig(unittest.TestCase):
|
|
def test_defaults(self):
|
|
cfg = Config()
|
|
self.assertEqual(cfg.kafka_bootstrap_servers, "localhost:9092")
|
|
self.assertEqual(cfg.kafka_logs_raw_topic, "logs.raw")
|
|
self.assertEqual(cfg.kafka_logs_fault_topic, "logs.fault")
|
|
self.assertEqual(cfg.cluster_eps, 0.75)
|
|
self.assertEqual(cfg.cluster_min_samples, 5)
|
|
self.assertEqual(cfg.api_port, 8080)
|
|
self.assertEqual(cfg.storage, "memory")
|
|
|
|
def test_from_env_defaults(self):
|
|
cfg = Config.from_env()
|
|
self.assertEqual(cfg.api_host, "127.0.0.1")
|
|
self.assertEqual(cfg.root_cause_window_minutes, 5)
|
|
|
|
def test_from_env_override(self):
|
|
os.environ["KAFKA_BOOTSTRAP_SERVERS"] = "kafka:29092"
|
|
os.environ["CLUSTER_EPS"] = "0.6"
|
|
os.environ["API_PORT"] = "9090"
|
|
os.environ["STORAGE"] = "mysql"
|
|
try:
|
|
cfg = Config.from_env()
|
|
self.assertEqual(cfg.kafka_bootstrap_servers, "kafka:29092")
|
|
self.assertAlmostEqual(cfg.cluster_eps, 0.6)
|
|
self.assertEqual(cfg.api_port, 9090)
|
|
self.assertEqual(cfg.storage, "mysql")
|
|
finally:
|
|
for k in ("KAFKA_BOOTSTRAP_SERVERS", "CLUSTER_EPS", "API_PORT", "STORAGE"):
|
|
os.environ.pop(k, None)
|
|
|
|
def test_to_dict_roundtrip(self):
|
|
cfg = Config(api_port=9001)
|
|
data = cfg.to_dict()
|
|
self.assertEqual(data["api_port"], 9001)
|
|
self.assertIn("cluster_eps", data)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|