test: 建立单元测试基座

- 新增 unittest tests 基座,覆盖 appconfig 与 db 纯逻辑

- 增加 excel/prompts 未实现模块的契约占位测试

- 补强 config.json 对 *_token 与 *_password 的敏感字段拦截

- 更新任务看板、当前状态、API 合约和进度记录
This commit is contained in:
chengma
2026-06-27 09:17:20 +08:00
parent 0de7c84163
commit aba272dd21
10 changed files with 345 additions and 8 deletions
+1
View File
@@ -0,0 +1 @@
+19
View File
@@ -0,0 +1,19 @@
"""Shared unittest helpers."""
import os
import sys
import tempfile
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
class TempDirMixin:
def make_temp_dir(self):
return tempfile.TemporaryDirectory(prefix="cmshopee_test_", dir=Path(__file__).parent)
def assert_removed(self, path):
self.assertFalse(os.path.exists(path), path)
+118
View File
@@ -0,0 +1,118 @@
import os
import sys
import unittest
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("images", appconfig.image_dir(config))
self.assertEqual(240, appconfig.response_timeout(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))
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)
if __name__ == "__main__":
unittest.main()
+144
View File
@@ -0,0 +1,144 @@
import os
import sys
import unittest
sys.path.insert(0, os.path.dirname(__file__))
from _helpers import TempDirMixin
from app import db
class DbTests(TempDirMixin, unittest.TestCase):
def test_init_db_is_idempotent_and_sets_pragmas(self):
with self.make_temp_dir() as temp_dir:
db_path = os.path.join(temp_dir, "cmshopee.db")
db.init_db(db_path)
db.init_db(db_path)
conn = db.connect(db_path)
try:
self.assertEqual(1, conn.execute("PRAGMA foreign_keys").fetchone()[0])
self.assertEqual("wal", conn.execute("PRAGMA journal_mode").fetchone()[0])
self.assertEqual(5000, conn.execute("PRAGMA busy_timeout").fetchone()[0])
tables = {
row["name"]
for row in conn.execute(
"SELECT name FROM sqlite_master WHERE type = 'table'"
).fetchall()
}
self.assertTrue({"batches", "accounts", "tasks"}.issubset(tables))
finally:
conn.close()
self.assert_removed(temp_dir)
def test_account_batch_task_lifecycle(self):
with self.make_temp_dir() as temp_dir:
db_path = os.path.join(temp_dir, "cmshopee.db")
db.init_db(db_path)
batch_id = db.create_batch(["input.xlsx"], note="导入", path=db_path)
batch = db.get_batch(batch_id, path=db_path)
self.assertEqual("导入", batch.note)
self.assertEqual([os.path.abspath("input.xlsx")], batch.source_files)
account = db.add_account(
"shop",
"alias",
"shopee.tw",
9222,
note="备注",
path=db_path,
)
self.assertEqual("alias", account.alias)
self.assertEqual("alias_cdb6fdbe", account.slug)
db.update_account("alias", path=db_path, debug_port=9333)
self.assertEqual(9333, db.get_account_by_alias("alias", path=db_path).debug_port)
self.assertEqual(1, len(db.list_accounts(path=db_path)))
count = db.insert_tasks(
batch_id,
[
{
"source_file": "input.xlsx",
"source_file_abs": os.path.abspath("input.xlsx"),
"source_sheet": "Sheet1",
"source_row": 2,
"account_name": "shop",
"alias": "alias",
"item_id": "51100639510",
}
],
path=db_path,
)
self.assertEqual(1, count)
task = db.list_tasks(batch_id=batch_id, alias="alias", path=db_path)[0]
self.assertEqual("imported", task.stage)
self.assertEqual("pending", task.status)
db.mark_running(task.id, "collect", path=db_path)
self.assertEqual("running", db.list_tasks(path=db_path)[0].status)
db.mark_failed(task.id, "collect", "采集失败", path=db_path)
failed = db.list_tasks(path=db_path)[0]
self.assertEqual("imported", failed.stage)
self.assertEqual("failed", failed.status)
self.assertEqual(1, failed.collect_attempts)
db.set_collected(task.id, "旧标题", "old.jpg", path=db_path)
collected = db.list_tasks(path=db_path)[0]
self.assertEqual("collected", collected.stage)
self.assertEqual("success", collected.status)
self.assertEqual("旧标题", collected.old_title)
db.set_generated(task.id, "新标题", "new.jpg", path=db_path)
generated = db.list_tasks(path=db_path)[0]
self.assertEqual("generated", generated.stage)
self.assertEqual("新标题", generated.new_title)
db.set_applied(task.id, False, "按钮禁用", path=db_path)
apply_failed = db.list_tasks(path=db_path)[0]
self.assertEqual("generated", apply_failed.stage)
self.assertEqual("failed", apply_failed.status)
self.assertEqual(1, apply_failed.apply_attempts)
db.set_applied(task.id, True, path=db_path)
applied = db.list_tasks(path=db_path)[0]
self.assertEqual("applied", applied.stage)
self.assertEqual("success", applied.status)
self.assertEqual(1, applied.committed)
self.assertEqual(2, applied.apply_attempts)
self.assert_removed(temp_dir)
def test_duplicate_task_and_invalid_update_raise_clear_errors(self):
with self.make_temp_dir() as temp_dir:
db_path = os.path.join(temp_dir, "cmshopee.db")
db.init_db(db_path)
batch_id = db.create_batch(["input.xlsx"], path=db_path)
row = {
"source_file": "input.xlsx",
"source_file_abs": os.path.abspath("input.xlsx"),
"source_sheet": "Sheet1",
"source_row": 2,
"alias": "alias",
"item_id": "51100639510",
}
db.insert_tasks(batch_id, [row], path=db_path)
with self.assertRaises(db.DbError):
db.insert_tasks(batch_id, [row], path=db_path)
with self.assertRaises(db.DbError):
db.update_account("alias", path=db_path, unknown_field=True)
self.assert_removed(temp_dir)
if __name__ == "__main__":
unittest.main()
+43
View File
@@ -0,0 +1,43 @@
import importlib
import importlib.util
import os
import sys
import unittest
sys.path.insert(0, os.path.dirname(__file__))
from _helpers import REPO_ROOT
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
def import_or_skip(module_name):
if importlib.util.find_spec(module_name) is None:
raise unittest.SkipTest(f"{module_name} 尚未实现")
return importlib.import_module(module_name)
class PendingModuleContractTests(unittest.TestCase):
def test_excel_contract_when_module_exists(self):
excel = import_or_skip("app.excel")
for name in ("import_tasks", "match_summary", "write_back", "export_copy"):
self.assertTrue(callable(getattr(excel, name, None)), name)
def test_prompts_contract_when_module_exists(self):
prompts = import_or_skip("app.prompts")
for name in (
"load_title_prompt",
"save_title_prompt",
"list_cover_templates",
"load_cover_template",
"save_cover_template",
"rename_cover_template",
"delete_cover_template",
"render_prompt",
):
self.assertTrue(callable(getattr(prompts, name, None)), name)
if __name__ == "__main__":
unittest.main()