import json import tempfile import unittest from pathlib import Path ROOT_DIR = Path(__file__).resolve().parents[1] ROOT_CONFIG_FILE = ROOT_DIR / "config.json" class ConfigLoadingTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls._created_root_config = False if not ROOT_CONFIG_FILE.exists(): ROOT_CONFIG_FILE.write_text(json.dumps({"auth-key": "test-auth"}), encoding="utf-8") cls._created_root_config = True from services import config as config_module cls.config_module = config_module @classmethod def tearDownClass(cls) -> None: if cls._created_root_config and ROOT_CONFIG_FILE.exists(): ROOT_CONFIG_FILE.unlink() def test_load_settings_ignores_directory_config_path(self) -> None: with tempfile.TemporaryDirectory() as tmp_dir: base_dir = Path(tmp_dir) data_dir = base_dir / "data" config_dir = base_dir / "config.json" os_auth_key = "env-auth" config_dir.mkdir() module = self.config_module old_base_dir = module.BASE_DIR old_data_dir = module.DATA_DIR old_config_file = module.CONFIG_FILE old_env_auth_key = module.os.environ.get("CHATGPT2API_AUTH_KEY") try: module.BASE_DIR = base_dir module.DATA_DIR = data_dir module.CONFIG_FILE = config_dir module.os.environ["CHATGPT2API_AUTH_KEY"] = os_auth_key settings = module._load_settings() self.assertEqual(settings.auth_key, os_auth_key) self.assertEqual(settings.refresh_account_interval_minute, 5) finally: module.BASE_DIR = old_base_dir module.DATA_DIR = old_data_dir module.CONFIG_FILE = old_config_file if old_env_auth_key is None: module.os.environ.pop("CHATGPT2API_AUTH_KEY", None) else: module.os.environ["CHATGPT2API_AUTH_KEY"] = old_env_auth_key def test_image_poll_max_attempts_defaults_and_clamps(self) -> None: module = self.config_module with tempfile.TemporaryDirectory() as tmp_dir: config_file = Path(tmp_dir) / "config.json" config_file.write_text(json.dumps({"auth-key": "test-auth"}), encoding="utf-8") store = module.ConfigStore(config_file) self.assertEqual(store.image_poll_max_attempts, 2) store.data["image_poll_max_attempts"] = 0 self.assertEqual(store.image_poll_max_attempts, 1) store.data["image_poll_max_attempts"] = 99 self.assertEqual(store.image_poll_max_attempts, 10) store.data["image_poll_max_attempts"] = "invalid" self.assertEqual(store.image_poll_max_attempts, 2) def test_shared_announcements_migrates_local_when_shared_missing(self) -> None: module = self.config_module with tempfile.TemporaryDirectory() as tmp_dir: config_file = Path(tmp_dir) / "config.json" config_file.write_text( json.dumps( { "auth-key": "test-auth", "announcements": [ { "id": "local-notice", "type": "notice", "target": "normal", "content": "local notice", "enabled": True, "created_at": "2026-06-11T00:00:00Z", "updated_at": "2026-06-11T00:00:00Z", } ], } ), encoding="utf-8", ) store = module.ConfigStore(config_file) storage = _NamedStateStorage({}) store._shared_announcements_storage = lambda: storage # type: ignore[method-assign] items = store.get_announcements(include_disabled=True) self.assertEqual([item["content"] for item in items], ["local notice"]) self.assertEqual(storage.saved_key, module.SHARED_ANNOUNCEMENTS_KEY) self.assertEqual(storage.state["items"][0]["content"], "local notice") def test_shared_announcements_empty_state_wins_over_local(self) -> None: module = self.config_module with tempfile.TemporaryDirectory() as tmp_dir: config_file = Path(tmp_dir) / "config.json" config_file.write_text( json.dumps( { "auth-key": "test-auth", "announcements": [ { "id": "old-notice", "type": "notice", "target": "normal", "content": "old local notice", "enabled": True, } ], } ), encoding="utf-8", ) store = module.ConfigStore(config_file) storage = _NamedStateStorage({"items": []}) store._shared_announcements_storage = lambda: storage # type: ignore[method-assign] self.assertEqual(store.get_announcements(include_disabled=True), []) self.assertIsNone(storage.saved_key) def test_shared_announcements_skip_migration_does_not_seed_local(self) -> None: module = self.config_module previous_skip = module.os.environ.get("CHATGPT2API_SKIP_USER_STORAGE_MIGRATION") module.os.environ["CHATGPT2API_SKIP_USER_STORAGE_MIGRATION"] = "true" try: with tempfile.TemporaryDirectory() as tmp_dir: config_file = Path(tmp_dir) / "config.json" config_file.write_text( json.dumps( { "auth-key": "test-auth", "announcements": [ { "id": "local-notice", "type": "notice", "target": "normal", "content": "local notice", "enabled": True, } ], } ), encoding="utf-8", ) store = module.ConfigStore(config_file) storage = _NamedStateStorage({}) store._shared_announcements_storage = lambda: storage # type: ignore[method-assign] self.assertEqual([item["content"] for item in store.get_announcements(include_disabled=True)], ["local notice"]) self.assertIsNone(storage.saved_key) finally: if previous_skip is None: module.os.environ.pop("CHATGPT2API_SKIP_USER_STORAGE_MIGRATION", None) else: module.os.environ["CHATGPT2API_SKIP_USER_STORAGE_MIGRATION"] = previous_skip def test_update_announcements_saves_shared_state(self) -> None: module = self.config_module with tempfile.TemporaryDirectory() as tmp_dir: config_file = Path(tmp_dir) / "config.json" config_file.write_text(json.dumps({"auth-key": "test-auth"}), encoding="utf-8") store = module.ConfigStore(config_file) storage = _NamedStateStorage({"items": []}) store._shared_announcements_storage = lambda: storage # type: ignore[method-assign] result = store.update( { "announcements": [ { "id": "shared-timeline", "type": "timeline", "target": "all", "content": "shared timeline", "enabled": True, } ] } ) self.assertEqual(storage.saved_key, module.SHARED_ANNOUNCEMENTS_KEY) self.assertEqual(storage.state["items"][0]["content"], "shared timeline") self.assertEqual(result["announcements"][0]["content"], "shared timeline") def test_shared_credit_settings_migrates_local_when_shared_missing(self) -> None: module = self.config_module with tempfile.TemporaryDirectory() as tmp_dir: config_file = Path(tmp_dir) / "config.json" config_file.write_text( json.dumps( { "auth-key": "test-auth", "normal_user_default_credits": 0, "normal_user_daily_credits": 30, "shop_purchase_url": "https://shop.example", "global_system_prompt": "be safe", } ), encoding="utf-8", ) store = module.ConfigStore(config_file) storage = _NamedStateStorage({}) store._shared_state_storage = lambda: storage # type: ignore[method-assign] store._shared_settings_cache = None self.assertEqual(store.normal_user_daily_credits, 30) self.assertEqual(storage.saved_key, module.SHARED_SETTINGS_KEY) self.assertEqual(storage.state["normal_user_daily_credits"], 30) self.assertNotIn("shop_purchase_url", storage.state) self.assertEqual(store.shop_purchase_url, "https://shop.example") def test_shared_credit_settings_override_local_credit_values_only(self) -> None: module = self.config_module with tempfile.TemporaryDirectory() as tmp_dir: config_file = Path(tmp_dir) / "config.json" config_file.write_text( json.dumps( { "auth-key": "test-auth", "normal_user_default_credits": 0, "normal_user_daily_credits": 30, } ), encoding="utf-8", ) store = module.ConfigStore(config_file) storage = _NamedStateStorage( { "normal_user_default_credits": 5, "normal_user_daily_credits": 88, "shop_purchase_url": "https://shared.example", "log_levels": ["debug"], } ) store._shared_state_storage = lambda: storage # type: ignore[method-assign] store._shared_settings_cache = None data = store.get() self.assertEqual(data["normal_user_default_credits"], 5) self.assertEqual(data["normal_user_daily_credits"], 88) self.assertEqual(data["shop_purchase_url"], "") self.assertEqual(data["log_levels"], []) def test_update_settings_saves_shared_credit_state_only(self) -> None: module = self.config_module with tempfile.TemporaryDirectory() as tmp_dir: config_file = Path(tmp_dir) / "config.json" config_file.write_text(json.dumps({"auth-key": "test-auth"}), encoding="utf-8") store = module.ConfigStore(config_file) storage = _NamedStateStorage({}) store._shared_state_storage = lambda: storage # type: ignore[method-assign] store._shared_settings_cache = None result = store.update( { "normal_user_default_credits": 3, "normal_user_daily_credits": 66, "shop_purchase_url": "https://buy.example", "log_levels": ["info", "warning"], } ) self.assertEqual(storage.saved_key, module.SHARED_SETTINGS_KEY) self.assertEqual(storage.state["normal_user_default_credits"], 3) self.assertEqual(storage.state["normal_user_daily_credits"], 66) self.assertNotIn("shop_purchase_url", storage.state) self.assertNotIn("log_levels", storage.state) self.assertEqual(result["shop_purchase_url"], "https://buy.example") def test_prompt_optimizer_system_prompt_has_default_and_is_shared(self) -> None: module = self.config_module with tempfile.TemporaryDirectory() as tmp_dir: config_file = Path(tmp_dir) / "config.json" config_file.write_text(json.dumps({"auth-key": "test-auth"}), encoding="utf-8") store = module.ConfigStore(config_file) storage = _NamedStateStorage({}) store._shared_state_storage = lambda: storage # type: ignore[method-assign] store._shared_settings_cache = None self.assertEqual(store.prompt_optimizer_system_prompt, module.DEFAULT_PROMPT_OPTIMIZER_SYSTEM_PROMPT) result = store.update({"prompt_optimizer_system_prompt": "Return only a polished prompt."}) self.assertEqual(storage.state["prompt_optimizer_system_prompt"], "Return only a polished prompt.") self.assertEqual(result["prompt_optimizer_system_prompt"], "Return only a polished prompt.") def test_reverse_prompt_settings_have_defaults_and_are_shared(self) -> None: module = self.config_module with tempfile.TemporaryDirectory() as tmp_dir: config_file = Path(tmp_dir) / "config.json" config_file.write_text(json.dumps({"auth-key": "test-auth"}), encoding="utf-8") store = module.ConfigStore(config_file) storage = _NamedStateStorage({}) store._shared_state_storage = lambda: storage # type: ignore[method-assign] store._shared_settings_cache = None self.assertEqual(store.reverse_prompt_system_prompt, module.DEFAULT_REVERSE_PROMPT_SYSTEM_PROMPT) self.assertEqual(store.reverse_prompt_provider, "internal") self.assertEqual(store.reverse_prompt_base_url, "") self.assertEqual(store.reverse_prompt_api_key, "") self.assertEqual(store.reverse_prompt_model, "gpt-5-5") result = store.update({ "reverse_prompt_system_prompt": "Describe only what is visible.", "reverse_prompt_provider": "external", "reverse_prompt_base_url": "https://vision.example/v1", "reverse_prompt_api_key": "secret", "reverse_prompt_model": "vision-model", }) self.assertEqual(storage.state["reverse_prompt_system_prompt"], "Describe only what is visible.") self.assertEqual(storage.state["reverse_prompt_provider"], "external") self.assertEqual(storage.state["reverse_prompt_base_url"], "https://vision.example/v1") self.assertEqual(storage.state["reverse_prompt_api_key"], "secret") self.assertEqual(storage.state["reverse_prompt_model"], "vision-model") self.assertEqual(result["reverse_prompt_system_prompt"], "Describe only what is visible.") self.assertEqual(result["reverse_prompt_provider"], "external") self.assertEqual(result["reverse_prompt_model"], "vision-model") def test_product_visual_agent_settings_have_defaults_and_are_shared(self) -> None: module = self.config_module with tempfile.TemporaryDirectory() as tmp_dir: config_file = Path(tmp_dir) / "config.json" config_file.write_text(json.dumps({"auth-key": "test-auth"}), encoding="utf-8") store = module.ConfigStore(config_file) storage = _NamedStateStorage({}) store._shared_state_storage = lambda: storage # type: ignore[method-assign] store._shared_settings_cache = None self.assertEqual(store.product_visual_agent_provider, "internal") self.assertEqual(store.product_visual_agent_base_url, "") self.assertEqual(store.product_visual_agent_api_key, "") self.assertEqual(store.product_visual_agent_model, "gpt-5-5") result = store.update({ "product_visual_agent_provider": "external", "product_visual_agent_base_url": "https://agent.example/v1", "product_visual_agent_api_key": "agent-secret", "product_visual_agent_model": "vision-agent", }) self.assertEqual(storage.state["product_visual_agent_provider"], "external") self.assertEqual(storage.state["product_visual_agent_base_url"], "https://agent.example/v1") self.assertEqual(storage.state["product_visual_agent_api_key"], "agent-secret") self.assertEqual(storage.state["product_visual_agent_model"], "vision-agent") self.assertEqual(result["product_visual_agent_provider"], "external") self.assertEqual(result["product_visual_agent_model"], "vision-agent") def test_product_visual_agent_defaults_migrate_from_prompt_optimizer_settings(self) -> None: module = self.config_module with tempfile.TemporaryDirectory() as tmp_dir: config_file = Path(tmp_dir) / "config.json" config_file.write_text(json.dumps({"auth-key": "test-auth"}), encoding="utf-8") store = module.ConfigStore(config_file) storage = _NamedStateStorage({ "prompt_optimizer_provider": "external", "prompt_optimizer_base_url": "https://existing.example/v1", "prompt_optimizer_api_key": "existing-secret", "prompt_optimizer_model": "existing-vision-model", }) store._shared_state_storage = lambda: storage # type: ignore[method-assign] store._shared_settings_cache = None self.assertEqual(store.product_visual_agent_provider, "external") self.assertEqual(store.product_visual_agent_base_url, "https://existing.example/v1") self.assertEqual(store.product_visual_agent_api_key, "existing-secret") self.assertEqual(store.product_visual_agent_model, "existing-vision-model") def test_free_account_cleanup_settings_are_normalized(self) -> None: module = self.config_module with tempfile.TemporaryDirectory() as tmp_dir: config_file = Path(tmp_dir) / "config.json" config_file.write_text( json.dumps( { "auth-key": "test-auth", "free_account_cleanup": { "enabled": "true", "interval_minutes": "0", "failure_threshold": "2", "register_precheck_enabled": "false", "action": "remove-forever", }, } ), encoding="utf-8", ) store = module.ConfigStore(config_file) cleanup = store.get()["free_account_cleanup"] self.assertTrue(cleanup["enabled"]) self.assertEqual(cleanup["interval_minutes"], 1) self.assertEqual(cleanup["failure_threshold"], 2) self.assertFalse(cleanup["register_precheck_enabled"]) self.assertEqual(cleanup["action"], "mark_abnormal") def test_checkin_content_defaults_partial_update_and_blank_hides(self) -> None: module = self.config_module with tempfile.TemporaryDirectory() as tmp_dir: config_file = Path(tmp_dir) / "config.json" config_file.write_text(json.dumps({"auth-key": "test-auth"}), encoding="utf-8") store = module.ConfigStore(config_file) # 未配置时回落到内置默认文案，且 get() 一定带出该键。 defaults = store.get()["checkin_content"] self.assertEqual(defaults, module.DEFAULT_CHECKIN_CONTENT) # 只提交一个字段：其余字段沿用旧值，而不是被打回默认。 store.update({"checkin_content": {"notice": "改过的顶部说明"}}) self.assertEqual(store.checkin_content["notice"], "改过的顶部说明") self.assertEqual(store.checkin_content["dialog_text"], module.DEFAULT_CHECKIN_CONTENT["dialog_text"]) # 空串 = 显式隐藏该段。 store.update({"checkin_content": {"notice": "", "qrcode_url": ""}}) self.assertEqual(store.checkin_content["notice"], "") self.assertEqual(store.checkin_content["qrcode_url"], "") # 多行文案允许用数组提交，并按行拼接。 store.update({"checkin_content": {"dialog_footer": ["第一行", "第二行", ""]}}) self.assertEqual(store.checkin_content["dialog_footer"], "第一行\n第二行") self.assertEqual(store.checkin_content["notice"], "") # 无关更新不应覆盖已保存的文案；且要真落盘。 store.update({"thread_pool_tokens": 40}) self.assertEqual(store.checkin_content["notice"], "") persisted = json.loads(config_file.read_text(encoding="utf-8")) self.assertEqual(persisted["checkin_content"]["notice"], "") def test_checkin_credit_valid_days_defaults_to_permanent(self) -> None: module = self.config_module with tempfile.TemporaryDirectory() as tmp_dir: config_file = Path(tmp_dir) / "config.json" config_file.write_text(json.dumps({"auth-key": "test-auth"}), encoding="utf-8") store = module.ConfigStore(config_file) # 默认 0 = 签到积分为永久积分，且 get() 一定带出该键。 self.assertEqual(store.checkin_credit_valid_days, 0) self.assertEqual(store.get()["checkin_credit_valid_days"], 0) store.update({"checkin_credit_valid_days": 7}) self.assertEqual(store.checkin_credit_valid_days, 7) persisted = json.loads(config_file.read_text(encoding="utf-8")) self.assertEqual(persisted["checkin_credit_valid_days"], 7) # 负数收敛到 0（永久）；超大值封顶 3650 天。 store.update({"checkin_credit_valid_days": -3}) self.assertEqual(store.checkin_credit_valid_days, 0) store.update({"checkin_credit_valid_days": 999999}) self.assertEqual(store.checkin_credit_valid_days, 3650) self.assertEqual(store.get()["checkin_credit_valid_days"], 3650) def test_user_navigation_is_normalized_and_separated_by_audience(self) -> None: module = self.config_module with tempfile.TemporaryDirectory() as tmp_dir: config_file = Path(tmp_dir) / "config.json" config_file.write_text( json.dumps( { "auth-key": "test-auth", "normal_user_navigation": [ { "id": "tools", "title": " 工具导航 ", "url": "", "new_window": False, "enabled": True, "children": [ { "id": "images", "title": "在线生图", "url": "https://example.com/images", "new_window": True, "append_api_credentials": True, "api_key_param": "token", "base_url_param": "api_base", "enabled": True, }, { "id": "hidden-child", "title": "暂不展示", "url": "/hidden", "enabled": False, }, ], }, { "id": "disabled-root", "title": "停用入口", "url": "/disabled", "enabled": False, }, { "id": "unsafe", "title": "不安全链接", "url": "javascript:alert(1)", "enabled": True, }, ], "internal_user_navigation": [ { "id": "internal-tools", "title": "内部工具", "url": "/internal-tools", "enabled": True, } ], } ), encoding="utf-8", ) store = module.ConfigStore(config_file) editable = store.get_normal_user_navigation(include_disabled=True) visible = store.get_normal_user_navigation() self.assertEqual([item["id"] for item in editable], ["tools", "disabled-root"]) self.assertEqual(editable[0]["title"], "工具导航") self.assertEqual(len(editable[0]["children"]), 2) self.assertEqual([item["id"] for item in visible], ["tools"]) self.assertEqual([item["id"] for item in visible[0]["children"]], ["images"]) self.assertTrue(visible[0]["children"][0]["new_window"]) self.assertTrue(visible[0]["children"][0]["append_api_credentials"]) self.assertEqual(visible[0]["children"][0]["api_key_param"], "token") self.assertEqual(visible[0]["children"][0]["base_url_param"], "api_base") self.assertEqual( [item["id"] for item in store.get_internal_user_navigation()], ["internal-tools"], ) self.assertEqual(store.get_internal_user_navigation()[0]["url"], "/internal-tools") def test_chat_navigation_is_migrated_once_after_image(self) -> None: module = self.config_module with tempfile.TemporaryDirectory() as tmp_dir: config_file = Path(tmp_dir) / "config.json" config_file.write_text( json.dumps({ "auth-key": "test-auth", "normal_user_navigation": [ {"id": "image", "title": "画图", "url": "/image", "enabled": True}, {"id": "checkin", "title": "签到", "url": "/checkin", "enabled": True}, ], }), encoding="utf-8", ) store = module.ConfigStore(config_file) self.assertEqual( [item["url"] for item in store.get_normal_user_navigation()], ["/image", "/chat", "/checkin"], ) self.assertNotIn(module.CHAT_NAVIGATION_MIGRATION_KEY, store.get()) store.update({ "normal_user_navigation": [ item for item in store.get_normal_user_navigation(include_disabled=True) if item["url"] != "/chat" ] }) reloaded = module.ConfigStore(config_file) self.assertEqual( [item["url"] for item in reloaded.get_normal_user_navigation()], ["/image", "/checkin"], ) class _NamedStateStorage: def __init__(self, state: dict[str, object]): self.state = dict(state) self.saved_key: str | None = None def load_named_state(self, key: str) -> dict[str, object]: return dict(self.state) def save_named_state(self, key: str, state: dict[str, object]) -> None: self.saved_key = key self.state = dict(state) if __name__ == "__main__": unittest.main()