import json import os import sys import unittest from unittest import mock sys.path.insert(0, os.path.dirname(__file__)) from _helpers import TempDirMixin from app import appconfig class AppConfigTests(TempDirMixin, unittest.TestCase): def test_config_load_update_and_response_timeout(self): with self.make_temp_dir() as temp_dir: config_path = os.path.join(temp_dir, "config.json") config = appconfig.load_config(config_path) self.assertTrue(os.path.exists(config_path)) self.assertEqual("", config["chrome_path"]) self.assertEqual(os.path.join(temp_dir, "images"), appconfig.image_dir(config)) self.assertEqual(os.path.join(temp_dir, "chrome_user_data_dir"), appconfig.user_data_root(config)) self.assertEqual(os.path.join(temp_dir, "cmshopee.db"), appconfig.db_path(config)) self.assertEqual(temp_dir, appconfig.data_dir(config)) self.assertEqual(config_path, config["config_path"]) self.assertEqual(os.path.join(temp_dir, "config", "ai_models.json"), config["ai_models_path"]) self.assertEqual(os.path.join(temp_dir, "config", "cmhub.json"), config["cmhub_config_path"]) self.assertEqual(240, appconfig.response_timeout(config)) self.assertFalse(appconfig.ai_config(config)["generate_cover"]) self.assertEqual("title", appconfig.ai_generate_mode(config)) self.assertEqual("title", appconfig.shopee_update_config(config)["update_mode"]) self.assertNotIn("allow_cover_update", appconfig.shopee_update_config(config)) self.assertEqual(1, appconfig.shopee_update_config(config)["max_parallel_accounts"]) self.assertEqual( { "platform": "Shopee", "country": "中国台湾", "language": "繁体中文", "ratio": "1:1", }, appconfig.product_suite_last_settings(config), ) self.assertEqual( "", appconfig.product_suite_last_account_alias(config), ) updated = appconfig.update_config( {"ai": {"resolution": "2k"}}, path=config_path, ) self.assertEqual(360, appconfig.response_timeout(updated)) self.assertEqual((9222, 9260), appconfig.debug_port_range(updated)) legacy = appconfig.save_config( { "ai": {"generate_cover": True}, "shopee_update": {"allow_cover_update": True}, }, path=config_path, ) self.assertEqual("title_cover", appconfig.ai_generate_mode(legacy)) self.assertTrue(appconfig.ai_config(legacy)["generate_cover"]) self.assertEqual("title_cover", appconfig.shopee_update_config(legacy)["update_mode"]) self.assertNotIn("allow_cover_update", appconfig.shopee_update_config(legacy)) cover_only = appconfig.save_config( { "ai": {"generate_mode": "cover"}, "shopee_update": {"update_mode": "cover"}, }, path=config_path, ) self.assertEqual("cover", appconfig.ai_generate_mode(cover_only)) self.assertTrue(appconfig.ai_config(cover_only)["generate_cover"]) self.assertEqual("cover", appconfig.shopee_update_config(cover_only)["update_mode"]) self.assertNotIn("allow_cover_update", appconfig.shopee_update_config(cover_only)) self.assert_removed(temp_dir) def test_product_suite_last_settings_are_normalized_and_preserve_other_config(self): with self.make_temp_dir() as temp_dir: config_path = os.path.join(temp_dir, "config.json") with open(config_path, "w", encoding="utf-8") as fh: json.dump( { "chrome_path": "custom-chrome.exe", "custom_section": {"keep": True}, "product_suite": { "last_account_alias": ["invalid"], "last_settings": { "platform": "未知平台", "country": "新加坡", "language": "英文", "ratio": "2:3", "unexpected": "ignore", } }, }, fh, ensure_ascii=False, ) loaded = appconfig.load_config(config_path) self.assertEqual( { "platform": "Shopee", "country": "新加坡", "language": "英文", "ratio": "1:1", }, appconfig.product_suite_last_settings(loaded), ) self.assertEqual( "", appconfig.product_suite_last_account_alias(loaded), ) updated = appconfig.update_config( { "product_suite": { "last_account_alias": " alias-b ", "last_settings": { "platform": "Amazon", "country": "中国台湾", "language": "繁体中文", "ratio": "16:9", } } }, path=config_path, ) self.assertEqual("custom-chrome.exe", updated["chrome_path"]) self.assertEqual({"keep": True}, updated["custom_section"]) self.assertEqual( "alias-b", appconfig.product_suite_last_account_alias(updated), ) self.assertEqual("16:9", appconfig.product_suite_last_settings(updated)["ratio"]) with open(config_path, "r", encoding="utf-8") as fh: persisted = json.load(fh) self.assertEqual( {"platform", "country", "language", "ratio"}, set(persisted["product_suite"]["last_settings"]), ) self.assertEqual( "alias-b", persisted["product_suite"]["last_account_alias"], ) self.assert_removed(temp_dir) def test_shopee_update_legacy_parallel_config_is_migrated(self): with self.make_temp_dir() as temp_dir: config_path = os.path.join(temp_dir, "config.json") with open(config_path, "w", encoding="utf-8") as fh: json.dump( { "shopee_update": { "allow_real_submit": False, "allow_cover_update": True, "close_success_tab": False, "parallel_accounts": False, "max_parallel_accounts": 2, } }, fh, ) loaded = appconfig.load_config(config_path) update_cfg = appconfig.shopee_update_config(loaded) self.assertEqual("title_cover", update_cfg["update_mode"]) self.assertEqual(1, update_cfg["max_parallel_accounts"]) for key in ( "allow_real_submit", "allow_cover_update", "close_success_tab", "parallel_accounts", ): self.assertNotIn(key, update_cfg) saved = appconfig.save_config(loaded, path=config_path) with open(config_path, "r", encoding="utf-8") as fh: persisted = json.load(fh) persisted_update = persisted["shopee_update"] self.assertEqual(1, saved["shopee_update"]["max_parallel_accounts"]) for key in ( "allow_real_submit", "allow_cover_update", "close_success_tab", "parallel_accounts", ): self.assertNotIn(key, persisted_update) with open(config_path, "w", encoding="utf-8") as fh: json.dump( { "shopee_update": { "parallel_accounts": True, "max_parallel_accounts": 16, } }, fh, ) loaded_parallel = appconfig.load_config(config_path) self.assertEqual( 5, appconfig.shopee_update_config(loaded_parallel)["max_parallel_accounts"], ) self.assert_removed(temp_dir) def test_config_save_does_not_persist_runtime_paths(self): with self.make_temp_dir() as temp_dir: config_path = os.path.join(temp_dir, "config.json") saved = appconfig.save_config( { "data_dir": os.path.join(temp_dir, "data"), "config_path": "runtime-only", "ai_models_path": "runtime-only", "cmhub_config_path": "runtime-only", "image_dir": "images", }, path=config_path, ) self.assertEqual(temp_dir, saved["data_dir"]) with open(config_path, "r", encoding="utf-8") as fh: persisted = json.load(fh) self.assertNotIn("data_dir", persisted) self.assertNotIn("config_path", persisted) self.assertNotIn("ai_models_path", persisted) self.assertNotIn("cmhub_config_path", persisted) self.assert_removed(temp_dir) def test_ai_concurrency_and_retry_are_clamped(self): with self.make_temp_dir() as temp_dir: config_path = os.path.join(temp_dir, "config.json") with open(config_path, "w", encoding="utf-8") as fh: json.dump( { "ai": { "title_concurrency": 0, "image_concurrency": 64, "retry": 20, } }, fh, ) loaded = appconfig.load_config(config_path) ai = appconfig.ai_config(loaded) self.assertEqual(1, ai["title_concurrency"]) self.assertEqual(5, ai["image_concurrency"]) self.assertEqual(10, ai["retry"]) saved = appconfig.save_config( { "ai": { "title_concurrency": 3, "image_concurrency": 4, "retry": 2, } }, path=config_path, ) ai = appconfig.ai_config(saved) self.assertEqual(3, ai["title_concurrency"]) self.assertEqual(4, ai["image_concurrency"]) self.assertEqual(2, ai["retry"]) self.assert_removed(temp_dir) def test_cmhub_download_with_curl_mode_is_normalized(self): with self.make_temp_dir() as temp_dir: config_path = os.path.join(temp_dir, "config.json") config = appconfig.load_config(config_path) self.assertEqual("auto", appconfig.cmhub_config(config)["download_with_curl"]) saved = appconfig.save_config( { "ai": { "cmhub": { "download_with_curl": True, } } }, path=config_path, ) self.assertEqual("true", appconfig.cmhub_config(saved)["download_with_curl"]) saved = appconfig.save_config( { "ai": { "cmhub": { "download_with_curl": "invalid", } } }, path=config_path, ) self.assertEqual("auto", appconfig.cmhub_config(saved)["download_with_curl"]) self.assert_removed(temp_dir) def test_data_paths_resolve_under_default_data_dir(self): with self.make_temp_dir() as temp_dir: data_root = os.path.join(temp_dir, "data") cfg = { "data_dir": data_root, "user_data_root": "chrome_user_data_dir", "image_dir": "images", "db_path": "cmshopee.db", } self.assertEqual(os.path.join(data_root, "chrome_user_data_dir"), appconfig.user_data_root(cfg)) self.assertEqual(os.path.join(data_root, "images"), appconfig.image_dir(cfg)) self.assertEqual(os.path.join(data_root, "cmshopee.db"), appconfig.db_path(cfg)) self.assertEqual(os.path.join(data_root, "title_prompt.txt"), appconfig.title_prompt_path(cfg)) self.assertEqual(os.path.join(data_root, "prompts", "title"), appconfig.title_templates_dir(cfg)) self.assertEqual(os.path.join(data_root, "prompts", "cover"), appconfig.cover_prompts_dir(cfg)) self.assertEqual( os.path.join(data_root, "prompts", "image_studio"), appconfig.image_studio_prompts_dir(cfg), ) self.assertEqual(os.path.join(data_root, "logs"), appconfig.diagnostic_log_dir(cfg)) portable = dict(cfg) portable["db_path"] = os.path.join("data", "cmshopee.db") self.assertEqual(os.path.join(data_root, "cmshopee.db"), appconfig.db_path(portable)) self.assert_removed(temp_dir) def test_prepare_data_dir_migrates_legacy_layout(self): with self.make_temp_dir() as temp_dir: with open(os.path.join(temp_dir, "config.json"), "w", encoding="utf-8") as fh: fh.write("{}") with open(os.path.join(temp_dir, "cmshopee.db"), "w", encoding="utf-8") as fh: fh.write("db") os.makedirs(os.path.join(temp_dir, "images"), exist_ok=True) with open(os.path.join(temp_dir, "images", "cover.jpg"), "w", encoding="utf-8") as fh: fh.write("image") os.makedirs(os.path.join(temp_dir, "config"), exist_ok=True) with open(os.path.join(temp_dir, "config", "cmhub.json"), "w", encoding="utf-8") as fh: fh.write("{}") data_root = os.path.join(temp_dir, "data") prepared = appconfig.prepare_data_dir(base_dir=temp_dir, data_dir_path=data_root) self.assertEqual(data_root, prepared) self.assertTrue(os.path.exists(os.path.join(data_root, "config.json"))) self.assertTrue(os.path.exists(os.path.join(data_root, "cmshopee.db"))) self.assertTrue(os.path.exists(os.path.join(data_root, "images", "cover.jpg"))) self.assertTrue(os.path.exists(os.path.join(data_root, "config", "cmhub.json"))) self.assertFalse(os.path.exists(os.path.join(temp_dir, "config.json"))) self.assertFalse(os.path.exists(os.path.join(temp_dir, "images"))) self.assert_removed(temp_dir) def test_prepare_data_dir_blocks_conflicting_migration(self): with self.make_temp_dir() as temp_dir: with open(os.path.join(temp_dir, "config.json"), "w", encoding="utf-8") as fh: fh.write("{}") data_root = os.path.join(temp_dir, "data") os.makedirs(data_root, exist_ok=True) with open(os.path.join(data_root, "config.json"), "w", encoding="utf-8") as fh: fh.write("{}") with self.assertRaises(appconfig.DataMigrationConflictError) as ctx: appconfig.prepare_data_dir(base_dir=temp_dir, data_dir_path=data_root) self.assertIn("无法自动迁移", str(ctx.exception)) self.assertTrue(os.path.exists(os.path.join(temp_dir, "config.json"))) self.assert_removed(temp_dir) def test_prepare_data_dir_reports_unwritable_path(self): with self.make_temp_dir() as temp_dir: data_root = os.path.join(temp_dir, "data") with open(data_root, "w", encoding="utf-8") as fh: fh.write("not a directory") with self.assertRaises(appconfig.DataDirectoryWriteError) as ctx: appconfig.prepare_data_dir( base_dir=temp_dir, data_dir_path=data_root, migrate=False, ) self.assertIn("数据目录不可写", str(ctx.exception)) self.assert_removed(temp_dir) def test_cmhub_defaults_old_config_and_key_helper(self): with self.make_temp_dir() as temp_dir: config_path = os.path.join(temp_dir, "config.json") cmhub_path = os.path.join(temp_dir, "config", "cmhub.json") config = appconfig.load_config(config_path) ai = appconfig.ai_config(config) self.assertEqual("cmhub", ai["backend"]) self.assertEqual("cmhub", appconfig.ai_backend(config)) self.assertEqual("", appconfig.cmhub_config(config)["base_url"]) self.assertEqual( appconfig.CMHUB_CONNECT_TIMEOUT_DEFAULT, appconfig.cmhub_config(config)["connect_timeout"], ) self.assertFalse(os.path.exists(cmhub_path)) self.assertEqual({"api_key": ""}, appconfig.load_cmhub_config(cmhub_path)) saved = appconfig.save_cmhub_config( {"api_key": "sk-cmhub-123456"}, path=cmhub_path, ) self.assertEqual("sk-cmhub-123456", saved["api_key"]) self.assertEqual("sk-cmhub-123456", appconfig.get_cmhub_api_key(cmhub_path)) self.assertEqual("sk-c***3456", appconfig.get_cmhub_api_key(cmhub_path, masked=True)) with open(config_path, "w", encoding="utf-8") as fh: json.dump({"ai": {"resolution": "512"}}, fh) migrated = appconfig.load_config(config_path) self.assertEqual("cmhub", appconfig.ai_config(migrated)["backend"]) self.assertEqual(180, appconfig.response_timeout(migrated)) with open(config_path, "w", encoding="utf-8") as fh: json.dump({"ai": {"cmhub": {"connect_timeout": 10}}}, fh) migrated = appconfig.load_config(config_path) self.assertEqual( appconfig.CMHUB_CONNECT_TIMEOUT_DEFAULT, appconfig.cmhub_config(migrated)["connect_timeout"], ) saved = appconfig.save_config( {"ai": {"cmhub": {"connect_timeout": 10}}}, path=config_path, ) self.assertEqual(10, appconfig.cmhub_config(saved)["connect_timeout"]) direct_cfg = appconfig.default_config() direct_cfg["ai"]["backend"] = "direct" self.assertEqual("direct", appconfig.ai_backend(direct_cfg)) self.assert_removed(temp_dir) def test_cmhub_base_url_normalizes_to_gateway_root(self): cases = { "https://cmhub.example.com/": "https://cmhub.example.com", "https://cmhub.example.com/api": "https://cmhub.example.com", "https://cmhub.example.com/api/v1/": "https://cmhub.example.com", "https://cmhub.example.com/some/path?x=1": "https://cmhub.example.com", "http://localhost:8000/api/v1": "http://localhost:8000", "localhost:8000/api/v1": "localhost:8000", } for raw, expected in cases.items(): with self.subTest(raw=raw): self.assertEqual(expected, appconfig.normalize_cmhub_base_url(raw)) self.assertEqual( expected + "/api/v1/models", appconfig.cmhub_request_url(raw, "/api/v1/models"), ) with self.make_temp_dir() as temp_dir: config_path = os.path.join(temp_dir, "config.json") config = appconfig.default_config() config["ai"]["cmhub"]["base_url"] = "https://cmhub.example.com/api/v1/" saved = appconfig.save_config(config, path=config_path) self.assertEqual("https://cmhub.example.com", saved["ai"]["cmhub"]["base_url"]) loaded = appconfig.load_config(config_path) self.assertEqual("https://cmhub.example.com", loaded["ai"]["cmhub"]["base_url"]) self.assert_removed(temp_dir) def test_config_rejects_sensitive_fields(self): with self.make_temp_dir() as temp_dir: config_path = os.path.join(temp_dir, "config.json") with self.assertRaises(appconfig.ConfigError): appconfig.save_config({"api_key": "secret"}, path=config_path) with self.assertRaises(appconfig.ConfigError): appconfig.save_config( {"ai": {"provider_token": "secret"}}, path=config_path, ) self.assert_removed(temp_dir) def test_ai_models_crud_filter_mask_and_get_model(self): with self.make_temp_dir() as temp_dir: models_path = os.path.join(temp_dir, "ai_models.json") models = appconfig.list_ai_models(path=models_path) self.assertEqual({"text", "image"}, {model["category"] for model in models}) self.assertTrue(all("api_key_set" in model for model in models)) appconfig.add_ai_model( { "name": "Text 2", "category": "text", "enabled": True, "url": "https://example.invalid/v1/chat/completions", "model": "demo-model", "api_key": "sk-1234567890", "api_type": "chat", "connect_timeout_seconds": 1, "extra_body": {"temperature": 0}, }, path=models_path, ) text_models = appconfig.list_ai_models("text", path=models_path) self.assertEqual(2, len(text_models)) self.assertEqual("sk-1***7890", text_models[-1]["api_key"]) self.assertTrue(text_models[-1]["api_key_set"]) private_model = appconfig.get_model("Text 2", path=models_path) self.assertEqual("sk-1234567890", private_model["api_key"]) self.assertEqual({"temperature": 0}, private_model["extra_body"]) appconfig.update_ai_model( "Text 2", path=models_path, name="Text 3", enabled=False, ) self.assertFalse(appconfig.get_model("Text 3", path=models_path)["enabled"]) self.assert_removed(temp_dir) def test_ai_model_constraints_and_connection_validation(self): with self.make_temp_dir() as temp_dir: models_path = os.path.join(temp_dir, "ai_models.json") appconfig.list_ai_models(path=models_path) with self.assertRaises(appconfig.ConfigError): appconfig.add_ai_model( { "name": "GPT-5.5 文本", "category": "text", "enabled": True, "api_type": "chat", "connect_timeout_seconds": 30, }, path=models_path, ) with self.assertRaises(appconfig.ConfigError): appconfig.delete_ai_model("Nano Banana 2", path=models_path) result = appconfig.test_ai_model("GPT-5.5 文本", path=models_path) self.assertFalse(result["ok"]) self.assertIn("url", result["error"]) self.assertIn("model", result["error"]) self.assertIn("api_key", result["error"]) self.assert_removed(temp_dir) def test_model_request_url_accepts_base_and_full_endpoint(self): self.assertEqual( "https://api.example.com/v1/chat/completions", appconfig.model_request_url( {"url": "https://api.example.com/v1", "api_type": "chat"} ), ) self.assertEqual( "https://openrouter.ai/api/v1/chat/completions", appconfig.model_request_url( {"url": "https://openrouter.ai/api/v1/", "api_type": "auto"} ), ) self.assertEqual( "https://api.example.com/v1/chat/completions?region=tw", appconfig.model_request_url( {"url": "https://api.example.com/v1?region=tw", "api_type": "chat"} ), ) self.assertEqual( "https://api.example.com/v1/chat/completions", appconfig.model_request_url( { "url": "https://api.example.com/v1/chat/completions", "api_type": "chat", } ), ) self.assertEqual( "https://api.example.com/v1/images/edits", appconfig.model_request_url( {"url": "https://api.example.com/v1", "api_type": "images_edits"} ), ) self.assertEqual( "https://api.example.com/custom/generate", appconfig.model_request_url( {"url": "https://api.example.com/custom/generate", "api_type": "chat"} ), ) def test_ai_model_test_uses_resolved_base_url(self): with self.make_temp_dir() as temp_dir: models_path = os.path.join(temp_dir, "ai_models.json") appconfig.save_ai_models_config( { "models": [ { "name": "Text", "category": "text", "enabled": True, "url": "https://api.example.com/v1", "model": "text-model", "api_key": "sk-text-secret", "api_type": "chat", "connect_timeout_seconds": 1, "extra_body": {}, }, { "name": "Image", "category": "image", "enabled": True, "url": "https://api.example.com/v1/chat/completions", "model": "image-model", "api_key": "sk-image-secret", "api_type": "auto", "connect_timeout_seconds": 1, "extra_body": {}, }, ] }, path=models_path, ) calls = [] class Response: status = 200 def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def read(self, size=-1): return b"{}" def fake_urlopen(request, timeout=None): calls.append((request, timeout)) return Response() with mock.patch("app.appconfig.urllib.request.urlopen", side_effect=fake_urlopen): result = appconfig.test_ai_model("Text", path=models_path) self.assertTrue(result["ok"]) self.assertEqual(200, result["status"]) self.assertEqual( "https://api.example.com/v1/chat/completions", calls[0][0].full_url, ) self.assertEqual(1, calls[0][1]) self.assert_removed(temp_dir) def test_sanitize_for_log_masks_secret_fields(self): payload = { "name": "demo", "email": "owner@example.com", "api_key": "sk-1234567890", "nested": { "support_email": "helpdesk@example.com", "password": "account-secret", "items": [ {"provider_token": "token-secret"}, {"value": "safe"}, {"api_key": {"value": "nested-secret"}}, ], }, } sanitized = appconfig.sanitize_for_log(payload) self.assertEqual("demo", sanitized["name"]) self.assertEqual("o***r@example.com", sanitized["email"]) self.assertEqual("h***k@example.com", sanitized["nested"]["support_email"]) self.assertEqual("sk-1***7890", sanitized["api_key"]) self.assertEqual("acco***cret", sanitized["nested"]["password"]) self.assertEqual("toke***cret", sanitized["nested"]["items"][0]["provider_token"]) self.assertEqual("safe", sanitized["nested"]["items"][1]["value"]) self.assertEqual("***", sanitized["nested"]["items"][2]["api_key"]) self.assertEqual("sk-1234567890", payload["api_key"]) if __name__ == "__main__": unittest.main()