study / test /test_config.py
xiaolongmr
签到积分账本与有效期 + 内容审核反绕过加固 + 签到文案配置化
44a4285
Raw History Blame Contribute Delete
28.6 kB
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()