874 lines
36 KiB
Python
874 lines
36 KiB
Python
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("vision-standard", appconfig.cmhub_config(config)["vision_alias"])
|
|
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_cmhub_vision_alias_defaults_for_legacy_config_and_preserves_saved_value(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": {
|
|
"backend": "cmhub",
|
|
"cmhub": {
|
|
"base_url": "https://cmhub.example.com",
|
|
"title_alias": "title-standard",
|
|
"image_alias": "image-hd",
|
|
},
|
|
}
|
|
},
|
|
fh,
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
legacy = appconfig.load_config(config_path)
|
|
self.assertEqual("vision-standard", appconfig.cmhub_config(legacy)["vision_alias"])
|
|
|
|
legacy["ai"]["cmhub"]["vision_alias"] = "vision-custom"
|
|
saved = appconfig.save_config(legacy, path=config_path)
|
|
self.assertEqual("vision-custom", appconfig.cmhub_config(saved)["vision_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.DEFAULT_CMHUB_BASE_URL,
|
|
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_default_gateway_url_and_first_use_state_preserve_existing_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({"ai": {"cmhub": {"base_url": ""}}}, fh)
|
|
|
|
migrated = appconfig.load_config(config_path)
|
|
|
|
self.assertEqual(
|
|
appconfig.DEFAULT_CMHUB_BASE_URL,
|
|
appconfig.cmhub_config(migrated)["base_url"],
|
|
)
|
|
self.assertEqual("", appconfig.first_use_guide_state(migrated))
|
|
|
|
migrated["ai"]["cmhub"]["base_url"] = "https://gateway.example.com/api/v1"
|
|
saved = appconfig.save_config(migrated, path=config_path)
|
|
self.assertEqual(
|
|
"https://gateway.example.com",
|
|
appconfig.cmhub_config(saved)["base_url"],
|
|
)
|
|
|
|
pending = appconfig.save_first_use_guide_state(
|
|
"pending",
|
|
path=config_path,
|
|
)
|
|
self.assertEqual("pending", appconfig.first_use_guide_state(pending))
|
|
self.assertEqual(
|
|
"https://gateway.example.com",
|
|
appconfig.cmhub_config(pending)["base_url"],
|
|
)
|
|
|
|
dismissed = appconfig.save_first_use_guide_state(
|
|
"dismissed",
|
|
path=config_path,
|
|
)
|
|
self.assertEqual("dismissed", appconfig.first_use_guide_state(dismissed))
|
|
with self.assertRaises(appconfig.ConfigError):
|
|
appconfig.save_first_use_guide_state("unknown", path=config_path)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_subscription_notice_id_persists_without_credentials(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
config_path = os.path.join(temp_dir, "config.json")
|
|
|
|
loaded = appconfig.load_config(config_path)
|
|
self.assertEqual("", appconfig.subscription_notice_id(loaded))
|
|
|
|
saved = appconfig.save_subscription_notice_id(
|
|
"subscription-notice-20260721",
|
|
path=config_path,
|
|
)
|
|
self.assertEqual(
|
|
"subscription-notice-20260721",
|
|
appconfig.subscription_notice_id(saved),
|
|
)
|
|
reloaded = appconfig.load_config(config_path)
|
|
self.assertEqual(
|
|
"subscription-notice-20260721",
|
|
appconfig.subscription_notice_id(reloaded),
|
|
)
|
|
|
|
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_rename_ai_model_preserves_fields_and_rejects_invalid_name(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
models_path = os.path.join(temp_dir, "ai_models.json")
|
|
appconfig.add_ai_model(
|
|
{
|
|
"name": "Text 2",
|
|
"category": "text",
|
|
"enabled": False,
|
|
"url": "https://example.invalid/v1/chat/completions",
|
|
"model": "demo-model",
|
|
"api_key": "sk-1234567890",
|
|
"api_type": "chat",
|
|
"connect_timeout_seconds": 12,
|
|
"timeout_seconds": 45,
|
|
"extra_body": {"temperature": 0},
|
|
},
|
|
path=models_path,
|
|
)
|
|
before = appconfig.get_model("Text 2", path=models_path)
|
|
|
|
renamed = appconfig.rename_ai_model(
|
|
"Text 2",
|
|
" 文本模型 ",
|
|
path=models_path,
|
|
)
|
|
|
|
self.assertEqual("文本模型", renamed["name"])
|
|
self.assertEqual(
|
|
{**before, "name": "文本模型"},
|
|
appconfig.get_model("文本模型", path=models_path),
|
|
)
|
|
with self.assertRaisesRegex(appconfig.ConfigError, "模型名称不能为空"):
|
|
appconfig.rename_ai_model("文本模型", " ", path=models_path)
|
|
with self.assertRaisesRegex(appconfig.ConfigError, "模型名称已存在"):
|
|
appconfig.rename_ai_model(
|
|
"文本模型",
|
|
"GPT-5.5 文本",
|
|
path=models_path,
|
|
)
|
|
|
|
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_image_model_config_check_keeps_legacy_models_but_blocks_them_for_generation(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
models_path = os.path.join(temp_dir, "ai_models.json")
|
|
config = appconfig.default_ai_models_config()
|
|
legacy_image = config["models"][1]
|
|
legacy_image.update(
|
|
{
|
|
"url": "https://legacy.example.com/v1/chat/completions",
|
|
"model": "legacy-image",
|
|
"api_key": "sk-legacy-secret",
|
|
"api_type": "auto",
|
|
}
|
|
)
|
|
appconfig.save_ai_models_config(config, path=models_path)
|
|
|
|
loaded = appconfig.get_model("Nano Banana 2", path=models_path)
|
|
self.assertEqual("auto", loaded["api_type"])
|
|
result = appconfig.check_image_model_config("Nano Banana 2", path=models_path)
|
|
self.assertFalse(result["ok"])
|
|
self.assertTrue(result["check_only"])
|
|
self.assertIn("OpenAI 图片编辑接口", result["error"])
|
|
self.assertNotIn("sk-legacy-secret", result["error"])
|
|
|
|
appconfig.update_ai_model(
|
|
"Nano Banana 2",
|
|
path=models_path,
|
|
api_type="images_edits",
|
|
)
|
|
self.assertTrue(
|
|
appconfig.check_image_model_config("Nano Banana 2", path=models_path)["ok"]
|
|
)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
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()
|