dekun
53863559f4
Retarget git remote, install path, and deploy docs from crypto_monitor to crypto_monitor_user. Co-authored-by: Cursor <cursoragent@cursor.com>
70 lines
2.7 KiB
Python
70 lines
2.7 KiB
Python
"""shared_env_lib:AI 字段与四文件同步."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
|
|
from lib.env.env_file_lib import apply_env_updates, read_env_lines
|
|
from lib.env.shared_env_lib import (
|
|
AI_ENV_KEYS,
|
|
apply_ai_env_to_all,
|
|
build_ai_env_payload,
|
|
validate_ai_env_updates,
|
|
)
|
|
|
|
|
|
class TestSharedEnvLib(unittest.TestCase):
|
|
def test_ai_keys_frozen(self) -> None:
|
|
self.assertIn("OPENAI_API_KEY", AI_ENV_KEYS)
|
|
self.assertIn("AI_PROVIDER", AI_ENV_KEYS)
|
|
|
|
def test_validate_rejects_unknown(self) -> None:
|
|
clean, errors = validate_ai_env_updates({"NOT_A_KEY": "x"})
|
|
self.assertEqual(clean, {})
|
|
self.assertTrue(any("未知" in e for e in errors))
|
|
|
|
def test_validate_skips_masked_secret(self) -> None:
|
|
clean, errors = validate_ai_env_updates({"OPENAI_API_KEY": "****abcd"})
|
|
self.assertEqual(errors, [])
|
|
self.assertNotIn("OPENAI_API_KEY", clean)
|
|
|
|
def test_apply_syncs_hub_and_instances(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
hub = os.path.join(tmp, "manual_trading_hub")
|
|
okx = os.path.join(tmp, "crypto_monitor_okx")
|
|
os.makedirs(hub)
|
|
os.makedirs(okx)
|
|
hub_env = os.path.join(hub, ".env")
|
|
okx_env = os.path.join(okx, ".env")
|
|
example = os.path.join(hub, ".env.example")
|
|
with open(example, "w", encoding="utf-8") as f:
|
|
f.write("AI_PROVIDER=openai\nOPENAI_API_KEY=\n")
|
|
with open(hub_env, "w", encoding="utf-8") as f:
|
|
f.write("AI_PROVIDER=openai\n")
|
|
with open(okx_env, "w", encoding="utf-8") as f:
|
|
f.write("AI_PROVIDER=ollama\n")
|
|
|
|
import lib.env.shared_env_lib as mod
|
|
|
|
orig_hub = mod.hub_env_path
|
|
orig_dirs = dict(mod.INSTANCE_ENV_DIRS)
|
|
try:
|
|
mod.hub_env_path = lambda: hub_env # type: ignore[method-assign]
|
|
mod.hub_example_path = lambda: example # type: ignore[method-assign]
|
|
mod.INSTANCE_ENV_DIRS = {"okx": __import__("pathlib").Path(okx)} # type: ignore[misc]
|
|
|
|
result = apply_ai_env_to_all({"AI_PROVIDER": "openai", "OPENAI_MODEL": "gpt-test"})
|
|
self.assertTrue(result["ok"])
|
|
self.assertEqual(read_env_lines(hub_env)[0], "AI_PROVIDER=openai")
|
|
okx_lines = read_env_lines(okx_env)
|
|
self.assertIn("AI_PROVIDER=openai", okx_lines)
|
|
self.assertIn("OPENAI_MODEL=gpt-test", okx_lines)
|
|
finally:
|
|
mod.hub_env_path = orig_hub # type: ignore[method-assign]
|
|
mod.INSTANCE_ENV_DIRS = orig_dirs # type: ignore[misc]
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|