From aba272dd211ee503fe37e706371ae61033bd3d11 Mon Sep 17 00:00:00 2001 From: chengma Date: Sat, 27 Jun 2026 09:17:20 +0800 Subject: [PATCH] =?UTF-8?q?test:=20=E5=BB=BA=E7=AB=8B=E5=8D=95=E5=85=83?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=E5=9F=BA=E5=BA=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 unittest tests 基座，覆盖 appconfig 与 db 纯逻辑 - 增加 excel/prompts 未实现模块的契约占位测试 - 补强 config.json 对 *_token 与 *_password 的敏感字段拦截 - 更新任务看板、当前状态、API 合约和进度记录 --- app/appconfig.py | 4 +- docs/06-tasks.md | 2 +- docs/api.md | 2 +- docs/current-state.md | 11 +-- progress.md | 9 +++ tests/__init__.py | 1 + tests/_helpers.py | 19 +++++ tests/test_appconfig.py | 118 +++++++++++++++++++++++++++ tests/test_db.py | 144 +++++++++++++++++++++++++++++++++ tests/test_module_contracts.py | 43 ++++++++++ 10 files changed, 345 insertions(+), 8 deletions(-) create mode 100644 tests/__init__.py create mode 100644 tests/_helpers.py create mode 100644 tests/test_appconfig.py create mode 100644 tests/test_db.py create mode 100644 tests/test_module_contracts.py diff --git a/app/appconfig.py b/app/appconfig.py index 7578376..8328268 100644 --- a/app/appconfig.py +++ b/app/appconfig.py @@ -109,7 +109,9 @@ def _assert_no_secrets(config): if isinstance(value, dict): for key, child in value.items(): lowered = str(key).lower() - if lowered in SECRET_FIELD_NAMES or lowered.endswith("_key"): + if lowered in SECRET_FIELD_NAMES or lowered.endswith( + ("_key", "_token", "_password") + ): raise ConfigError( f"config.json 不允许保存敏感字段: {'.'.join(path + [str(key)])}" ) diff --git a/docs/06-tasks.md b/docs/06-tasks.md index 2376f07..4cf0c17 100644 --- a/docs/06-tasks.md +++ b/docs/06-tasks.md @@ -29,7 +29,7 @@ | T-003 | `app/db.py` + SQLite 建表（batches/accounts/tasks，含 Excel 行定位、状态、时间戳、重试字段） | T-000 | `init_db` 幂等；`connect` 设置 WAL/busy_timeout/foreign_keys；账号/批次/任务/各 set_* 可用；schema 同架构 5.2 | DONE | | T-004 | `.gitignore`：排除 `config.json`、`config/ai_models.json`、`cmshopee.db`、`chrome_user_data_dir/`、`images/` | T-002, T-003 | 配置、密钥、凭证、业务数据、图片不被提交 | DONE | | T-005 | AI 模型清单后端：`config/ai_models.json` 读写 + category 过滤 + 测试连接 | T-002 | 本地明文 api_key；UI/API 打码显示；日志脱敏；至少 text/image 各一个；`get_model` 返回调用所需字段 | DONE | -| T-006 | 单元测试基座：`tests/` + appconfig/db/excel/prompts 最小测试 | T-002, T-003 | `python -m unittest discover -s tests` 可跑；不依赖真实 Shopee/AI；临时文件在测试目录清理 | TODO | +| T-006 | 单元测试基座：`tests/` + appconfig/db/excel/prompts 最小测试 | T-002, T-003 | `python -m unittest discover -s tests` 可跑；不依赖真实 Shopee/AI；临时文件在测试目录清理 | DONE | ## Phase 1 · 账号管理（④） diff --git a/docs/api.md b/docs/api.md index 76106f2..8dfa484 100644 --- a/docs/api.md +++ b/docs/api.md @@ -32,7 +32,7 @@ ai_config(config=None) -> dict # default_text_model/default_image response_timeout(config=None) -> int # = resolution_timeouts[resolution]（返回超时，随分辨率） ``` -`config.json` 不保存 AI Key；写入 `api_key` / `*_key` / `token` / `password` 等敏感字段时抛 `ConfigError`。AI Key 留给 `config/ai_models.json`。 +`config.json` 不保存 AI Key；写入 `api_key` / `*_key` / `token` / `*_token` / `password` / `*_password` 等敏感字段时抛 `ConfigError`。AI Key 留给 `config/ai_models.json`。 AI 模型清单（`config/ai_models.json`，含本地明文密钥，已建；UI 由 ⑤ 设置复用）： diff --git a/docs/current-state.md b/docs/current-state.md index 0c00260..9ecdade 100644 --- a/docs/current-state.md +++ b/docs/current-state.md @@ -6,10 +6,10 @@ ## 当前快照 - 日期：2026-06-27 -- 阶段：V0 单账号 CDP 流程已验证；V1 已完成 T-000 正式代码包结构、T-001 `app/editor.py` 模块化、T-002 `app/appconfig.py` 应用配置、T-003 SQLite 持久化地基、T-004 本地数据忽略规则、T-005 AI 模型清单后端。 +- 阶段：V0 单账号 CDP 流程已验证；V1 已完成 T-000 正式代码包结构、T-001 `app/editor.py` 模块化、T-002 `app/appconfig.py` 应用配置、T-003 SQLite 持久化地基、T-004 本地数据忽略规则、T-005 AI 模型清单后端、T-006 单元测试基座。 - 技术栈：Python 3.10+，自研 CDP（websocket-client + requests），SQLite（sqlite3）+ `config.json` + openpyxl + AI（服务商待定），GUI PySide6 5 Tab（已定）。 - 生产代码：已建立 `app/` 包 + 根入口 `main.py`；`app/cdp.py` 为已验证 CDP 底座；`app/editor.py` 已封装标题/封面/采集/更新按钮能力；`app/appconfig.py` 已实现 `config.json` 默认值/读写/更新、AI 参数与端口读取，以及 `config/ai_models.json` 模型清单 CRUD/过滤/打码/测试连接；`app/db.py` 已实现 SQLite schema、连接 PRAGMA、批次/账号/任务与阶段写库函数；`app/gui.py` 目前是入口占位，完整 PySide6 主窗口待 T-104。 -- 测试：当前以 `compileall` + 测试商品手动 CDP 验证为主；`tests/` 与 `python -m unittest discover -s tests` 由 T-006 建立，T-006 完成前不把缺少 `tests/` 视为验证失败。 +- 测试：`tests/` 已建立；T-006 后纯逻辑改动必须运行 `python -m unittest discover -s tests`，当前覆盖 appconfig/db，并对尚未实现的 app.excel/app.prompts 做契约占位 skip；CDP/Shopee 改动仍需测试商品手动验证。 - 数据：`config.json`、`config/ai_models.json`、`cmshopee.db`、`chrome_user_data_dir/`、`images/` 已由 `.gitignore` 排除；`app/appconfig.py` 首次读取缺失的 `config.json` 时会在本地写默认配置，`app/db.py` 调用 `init_db()` 时会在本地创建 SQLite DB。 ## 既定设计要点（文档已定） @@ -36,6 +36,7 @@ | `app/editor.py` | 已有 | T-001 产出：登录检测、打开商品页、读/写标题、读/下载封面、上传拖封面、更新按钮、apply_task | | `app/appconfig.py` | 已有 | T-002 产出：`config.json` 默认值、读写、更新、路径/端口/AI 参数读取；拒绝敏感字段写入 | | `app/db.py` | 已有 | T-003 产出：batches/accounts/tasks schema；WAL/busy_timeout/foreign_keys；账号/批次/任务与 set_* 阶段写库 | +| `tests/` | 已有 | T-006 产出：stdlib unittest 基座；appconfig/db 单元测试；excel/prompts 模块契约占位测试 | | `app/excel.py` / `app/config.py` / `app/chrome.py` / `app/workers.py` | 待建 | Phase 0-3 产出 | | `config.json` / `config/ai_models.json` / `cmshopee.db` / `chrome_user_data_dir/` / `images/` | 本地待建，已忽略 | 含配置、密钥、业务、登录态、图片，不提交版本库 | @@ -51,9 +52,9 @@ 任务状态以 [`06-tasks.md`](06-tasks.md) 为准，历史记录见 [`../progress.md`](../progress.md)。 -- 已完成：T-000（正式代码包结构）、T-001（`app/editor.py` 模块化）、T-002（`app/appconfig.py` + `config.json`）、T-003（`app/db.py` + SQLite 建表）、T-004（本地数据 gitignore）、T-005（AI 模型清单后端）。 +- 已完成：T-000（正式代码包结构）、T-001（`app/editor.py` 模块化）、T-002（`app/appconfig.py` + `config.json`）、T-003（`app/db.py` + SQLite 建表）、T-004（本地数据 gitignore）、T-005（AI 模型清单后端）、T-006（单元测试基座）。 - 正在进行：无。 -- 下一个可领取任务：**T-006（单元测试基座）**。 +- 下一个可领取任务：**T-101（`config` 生成 slug + 创建账号 user-data-dir）**。 ## 当前可运行内容 @@ -77,7 +78,7 @@ py -3 -c "import os,tempfile; from app import appconfig; d=tempfile.TemporaryDir python main.py py -3 -m app -# 单元测试（T-006 完成且 tests/ 存在后） +# 单元测试（T-006 后纯逻辑改动必跑） python -m unittest discover -s tests # 单账号闭环（不提交线上） diff --git a/progress.md b/progress.md index 4113fa6..c1d32e6 100644 --- a/progress.md +++ b/progress.md @@ -286,3 +286,12 @@ - 验证：`py -3 -m compileall app main.py` 通过；临时 `ai_models.json` 默认写入、分类过滤、key 打码、`get_model` 明文 key、重命名/禁用、缺少 url/model/key 的测试连接错误返回均通过；重复 name 与删除最后 image 模型均返回 `ConfigError`。 - 注意：本轮未使用真实 API Key 发外部网络请求；真实「测试连接」需用户在本地 `config/ai_models.json` 填入有效 url/model/api_key 后由 UI 或函数触发。 - 下一步：按任务看板领取 T-006。 + +## 【2026-06-27】T-006 单元测试基座 + +- 状态：DONE +- 变更：新增 `tests/` 基座，包含 `_helpers.py`、`test_appconfig.py`、`test_db.py`、`test_module_contracts.py`；覆盖 `config.json` 默认/更新/敏感字段拦截、`config/ai_models.json` CRUD/过滤/打码/约束、SQLite 初始化/PRAGMA/账号批次任务生命周期/失败状态；同步 `docs/06-tasks.md`、`docs/current-state.md`、`docs/api.md`。 +- 补强：`config.json` 敏感字段拦截从 exact `token/password` 扩展到 `*_token`、`*_password`，测试覆盖 `provider_token`。 +- 说明：`app.excel` 与 `app.prompts` 尚未实现，`test_module_contracts.py` 对这两个模块做契约占位测试；当前表现为 2 个 skip，后续模块文件出现后会检查公开函数是否齐全。 +- 验证：`py -3 -m compileall app main.py tests` 通过；`py -3 -m unittest discover -s tests` 通过（9 tests，skipped=2）；`python -m unittest discover -s tests` 通过（9 tests，skipped=2）；测试临时目录位于 `tests/` 下并已清理。 +- 下一步：按任务看板领取 T-101。 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ + diff --git a/tests/_helpers.py b/tests/_helpers.py new file mode 100644 index 0000000..a83ed4b --- /dev/null +++ b/tests/_helpers.py @@ -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) diff --git a/tests/test_appconfig.py b/tests/test_appconfig.py new file mode 100644 index 0000000..e60dab1 --- /dev/null +++ b/tests/test_appconfig.py @@ -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() diff --git a/tests/test_db.py b/tests/test_db.py new file mode 100644 index 0000000..b98ffaa --- /dev/null +++ b/tests/test_db.py @@ -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() diff --git a/tests/test_module_contracts.py b/tests/test_module_contracts.py new file mode 100644 index 0000000..7c6c530 --- /dev/null +++ b/tests/test_module_contracts.py @@ -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()