You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
87 lines
3.1 KiB
87 lines
3.1 KiB
# -*- coding: utf-8 -*-
|
|
"""Unit tests for system configuration service."""
|
|
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from src.config import Config
|
|
from src.core.config_manager import ConfigManager
|
|
from src.services.system_config_service import ConfigConflictError, SystemConfigService
|
|
|
|
|
|
class SystemConfigServiceTestCase(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
self.env_path = Path(self.temp_dir.name) / ".env"
|
|
self.env_path.write_text(
|
|
"\n".join(
|
|
[
|
|
"STOCK_LIST=600519,000001",
|
|
"GEMINI_API_KEY=secret-key-value",
|
|
"SCHEDULE_TIME=18:00",
|
|
"LOG_LEVEL=INFO",
|
|
]
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
os.environ["ENV_FILE"] = str(self.env_path)
|
|
Config.reset_instance()
|
|
|
|
self.manager = ConfigManager(env_path=self.env_path)
|
|
self.service = SystemConfigService(manager=self.manager)
|
|
|
|
def tearDown(self) -> None:
|
|
Config.reset_instance()
|
|
os.environ.pop("ENV_FILE", None)
|
|
self.temp_dir.cleanup()
|
|
|
|
def test_get_config_returns_raw_sensitive_values(self) -> None:
|
|
payload = self.service.get_config(include_schema=True)
|
|
items = {item["key"]: item for item in payload["items"]}
|
|
|
|
self.assertIn("GEMINI_API_KEY", items)
|
|
self.assertEqual(items["GEMINI_API_KEY"]["value"], "secret-key-value")
|
|
self.assertFalse(items["GEMINI_API_KEY"]["is_masked"])
|
|
self.assertTrue(items["GEMINI_API_KEY"]["raw_value_exists"])
|
|
|
|
def test_update_preserves_masked_secret(self) -> None:
|
|
old_version = self.manager.get_config_version()
|
|
response = self.service.update(
|
|
config_version=old_version,
|
|
items=[
|
|
{"key": "GEMINI_API_KEY", "value": "******"},
|
|
{"key": "STOCK_LIST", "value": "600519,300750"},
|
|
],
|
|
mask_token="******",
|
|
reload_now=False,
|
|
)
|
|
|
|
self.assertTrue(response["success"])
|
|
self.assertEqual(response["applied_count"], 1)
|
|
self.assertEqual(response["skipped_masked_count"], 1)
|
|
self.assertIn("STOCK_LIST", response["updated_keys"])
|
|
|
|
current_map = self.manager.read_config_map()
|
|
self.assertEqual(current_map["STOCK_LIST"], "600519,300750")
|
|
self.assertEqual(current_map["GEMINI_API_KEY"], "secret-key-value")
|
|
|
|
def test_validate_reports_invalid_time(self) -> None:
|
|
validation = self.service.validate(items=[{"key": "SCHEDULE_TIME", "value": "25:70"}])
|
|
self.assertFalse(validation["valid"])
|
|
self.assertTrue(any(issue["code"] == "invalid_format" for issue in validation["issues"]))
|
|
|
|
def test_update_raises_conflict_for_stale_version(self) -> None:
|
|
with self.assertRaises(ConfigConflictError):
|
|
self.service.update(
|
|
config_version="stale-version",
|
|
items=[{"key": "STOCK_LIST", "value": "600519"}],
|
|
reload_now=False,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|