jiangchang-platform-kit/tests/test_config.py

# -*- coding: utf-8 -*-
"""config 三级优先级与 ensure_env_file 测试。"""
from __future__ import annotations

import os
import tempfile
import unittest

from jiangchang_skill_core import config


class TestConfig(unittest.TestCase):
    def setUp(self) -> None:
        config.reset_cache()
        self._saved_env = dict(os.environ)

    def tearDown(self) -> None:
        os.environ.clear()
        os.environ.update(self._saved_env)
        config.reset_cache()

    def test_ensure_env_file_copies_once(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            example = os.path.join(tmp, ".env.example")
            with open(example, "w", encoding="utf-8") as f:
                f.write("FOO=from_example\nBAR=2\n")

            os.environ["JIANGCHANG_DATA_ROOT"] = tmp
            os.environ["JIANGCHANG_USER_ID"] = "testuser"

            dest = config.ensure_env_file("demo-skill", example)
            self.assertTrue(os.path.isfile(dest))
            with open(dest, encoding="utf-8") as f:
                content = f.read()
            self.assertIn("FOO=from_example", content)

            with open(dest, "w", encoding="utf-8") as f:
                f.write("FOO=user_edited\n")

            dest2 = config.ensure_env_file("demo-skill", example)
            self.assertEqual(dest, dest2)
            with open(dest2, encoding="utf-8") as f:
                self.assertIn("user_edited", f.read())

    def test_three_level_priority(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            example = os.path.join(tmp, ".env.example")
            with open(example, "w", encoding="utf-8") as f:
                f.write("KEY_A=example\nKEY_B=example\nKEY_C=example\n")

            os.environ["JIANGCHANG_DATA_ROOT"] = tmp
            os.environ["JIANGCHANG_USER_ID"] = "u1"
            dest = config.ensure_env_file("sk", example)

            with open(dest, "w", encoding="utf-8") as f:
                f.write("KEY_B=user_env\n")

            config.reset_cache()
            self.assertEqual(config.get("KEY_A"), "example")
            self.assertEqual(config.get("KEY_B"), "user_env")
            self.assertEqual(config.get("KEY_C"), "example")
            self.assertIsNone(config.get("MISSING"))

            os.environ["KEY_C"] = "process_env"
            self.assertEqual(config.get("KEY_C"), "process_env")

    def test_get_bool_float_int(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            example = os.path.join(tmp, ".env.example")
            with open(example, "w", encoding="utf-8") as f:
                f.write("FLAG=1\nRATE=2.5\nCOUNT=42\nBAD=x\n")

            os.environ["JIANGCHANG_DATA_ROOT"] = tmp
            os.environ["JIANGCHANG_USER_ID"] = "u1"
            config.ensure_env_file("sk", example)

            self.assertTrue(config.get_bool("FLAG"))
            self.assertFalse(config.get_bool("MISSING", default=False))
            self.assertAlmostEqual(config.get_float("RATE", 0.0), 2.5)
            self.assertEqual(config.get_int("COUNT", 0), 42)
            self.assertEqual(config.get_float("BAD", 9.9), 9.9)
            self.assertEqual(config.get_int("BAD", 7), 7)

    def test_parse_empty_value_with_inline_comment(self) -> None:
        """Regression: ``KEY=   # comment`` must be empty, not ``# comment``."""
        with tempfile.TemporaryDirectory() as tmp:
            example = os.path.join(tmp, ".env.example")
            with open(example, "w", encoding="utf-8") as f:
                f.write(
                    "DEFAULT_ACCOUNT_ID=                      # 可填账号管理中的编号\n"
                    "DEFAULT_LOGIN_ID=# 只填登录名\n"
                    "HAS_VALUE=real  # trailing note\n"
                    "URL=https://example.com/path#frag\n"
                    "QUOTED=\"keep # hash\"\n"
                    "EMPTY=\n"
                )

            os.environ["JIANGCHANG_DATA_ROOT"] = tmp
            os.environ["JIANGCHANG_USER_ID"] = "u1"
            # Clear process overrides for these keys if any
            for k in (
                "DEFAULT_ACCOUNT_ID",
                "DEFAULT_LOGIN_ID",
                "HAS_VALUE",
                "URL",
                "QUOTED",
                "EMPTY",
            ):
                os.environ.pop(k, None)

            dest = config.ensure_env_file("sk-comment", example)
            config.reset_cache()

            parsed = config._parse_env_file(dest)
            self.assertEqual(parsed.get("DEFAULT_ACCOUNT_ID"), "")
            self.assertEqual(parsed.get("DEFAULT_LOGIN_ID"), "")
            self.assertEqual(parsed.get("HAS_VALUE"), "real")
            self.assertEqual(parsed.get("URL"), "https://example.com/path#frag")
            self.assertEqual(parsed.get("QUOTED"), "keep # hash")
            self.assertEqual(parsed.get("EMPTY"), "")

            # Empty example/user values must not surface as fake non-empty defaults.
            self.assertIsNone(config.get("DEFAULT_ACCOUNT_ID"))
            self.assertIsNone(config.get("DEFAULT_LOGIN_ID"))
            self.assertEqual(config.get("HAS_VALUE"), "real")

    def test_merge_missing_env_keys_appends_only_new(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            example = os.path.join(tmp, ".env.example")
            dest = os.path.join(tmp, ".env")
            with open(example, "w", encoding="utf-8") as f:
                f.write("EXISTING=1\nNEW_KEY=abc\n")
            with open(dest, "w", encoding="utf-8") as f:
                f.write("EXISTING=keep\n")

            added = config.merge_missing_env_keys(
                example,
                dest,
                comment_skill="demo-skill",
            )
            self.assertEqual(added, ["NEW_KEY"])
            with open(dest, encoding="utf-8") as f:
                content = f.read()
            self.assertIn("EXISTING=keep", content)
            self.assertIn("NEW_KEY=abc", content)
            self.assertIn("Added by demo-skill", content)


if __name__ == "__main__":
    unittest.main()