169 lines
6.1 KiB
Python
169 lines
6.1 KiB
Python
"""冒烟测试：覆盖 demo 关键路径。
|
||
|
||
不依赖 pytest，可用 ``python -m unittest discover -s tests`` 跑。
|
||
但 fixture 用法对齐 pytest；这里把 fixture 用 ``setUp`` 模拟。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import json
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from contextlib import redirect_stderr, redirect_stdout
|
||
|
||
# 注入 scripts/ 到 sys.path
|
||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||
_SCRIPTS_DIR = os.path.join(os.path.dirname(_HERE), "scripts")
|
||
if _SCRIPTS_DIR not in sys.path:
|
||
sys.path.insert(0, _SCRIPTS_DIR)
|
||
|
||
# 与 main.py 同样的 mock 注入：没有真实平台库时启用 mock
|
||
if not os.environ.get("TODO_SKILL_USE_REAL_PLATFORM"):
|
||
try:
|
||
import jiangchang_skill_core # noqa: F401
|
||
except ImportError:
|
||
_mock_platform = os.path.join(_SCRIPTS_DIR, "_mock_platform")
|
||
if _mock_platform not in sys.path:
|
||
sys.path.insert(0, _mock_platform)
|
||
|
||
|
||
class IsolatedEnv:
|
||
"""临时数据根目录，测试结束自动清理。"""
|
||
|
||
def __init__(self):
|
||
self._tmp = tempfile.mkdtemp(prefix="todo_test_")
|
||
self._prev_root = os.environ.get("JIANGCHANG_DATA_ROOT")
|
||
self._prev_user = os.environ.get("JIANGCHANG_USER_ID")
|
||
os.environ["JIANGCHANG_DATA_ROOT"] = os.path.join(self._tmp, "data")
|
||
os.environ["JIANGCHANG_USER_ID"] = "test_user"
|
||
os.makedirs(os.environ["JIANGCHANG_DATA_ROOT"], exist_ok=True)
|
||
|
||
def reset_config(self):
|
||
from jiangchang_skill_core import config
|
||
config.reset_cache()
|
||
|
||
def __enter__(self):
|
||
self.reset_config()
|
||
return self
|
||
|
||
def __exit__(self, *exc):
|
||
if self._prev_root is None:
|
||
os.environ.pop("JIANGCHANG_DATA_ROOT", None)
|
||
else:
|
||
os.environ["JIANGCHANG_DATA_ROOT"] = self._prev_root
|
||
if self._prev_user is None:
|
||
os.environ.pop("JIANGCHANG_USER_ID", None)
|
||
else:
|
||
os.environ["JIANGCHANG_USER_ID"] = self._prev_user
|
||
return False
|
||
|
||
|
||
def _run_cli(argv: list[str]) -> tuple[int, str, str]:
|
||
"""调用 cli.app.main，返回 (rc, stdout, stderr)。"""
|
||
from cli.app import main
|
||
out = io.StringIO()
|
||
err = io.StringIO()
|
||
with redirect_stdout(out), redirect_stderr(err):
|
||
rc = main(argv)
|
||
return rc, out.getvalue(), err.getvalue()
|
||
|
||
|
||
class TestCliSmoke(unittest.TestCase):
|
||
def test_empty_argv_shows_usage_and_nonzero(self):
|
||
rc, out, _ = _run_cli([])
|
||
self.assertEqual(rc, 1)
|
||
self.assertIn("待办事项 demo", out)
|
||
|
||
def test_version_json(self):
|
||
with IsolatedEnv():
|
||
rc, out, _ = _run_cli(["version"])
|
||
self.assertEqual(rc, 0)
|
||
payload = json.loads(out.strip())
|
||
self.assertEqual(payload["skill"], "todo-skill")
|
||
self.assertIn("version", payload)
|
||
|
||
def test_init_db_creates_tables_and_is_idempotent(self):
|
||
with IsolatedEnv():
|
||
rc, out, _ = _run_cli(["init-db"])
|
||
self.assertEqual(rc, 0)
|
||
payload = json.loads(out.strip())
|
||
self.assertTrue(payload["ok"])
|
||
self.assertTrue(os.path.isfile(payload["db_path"]))
|
||
# 再来一次应仍成功
|
||
rc2, out2, _ = _run_cli(["init-db"])
|
||
self.assertEqual(rc2, 0)
|
||
self.assertEqual(json.loads(out2.strip())["db_path"], payload["db_path"])
|
||
|
||
def test_add_list_done_stats_full_flow(self):
|
||
with IsolatedEnv():
|
||
_run_cli(["init-db"])
|
||
rc, out, _ = _run_cli(["todo", "add", "任务A", "-p", "high"])
|
||
self.assertEqual(rc, 0)
|
||
item = json.loads(out.strip())
|
||
self.assertEqual(item["title"], "任务A")
|
||
self.assertEqual(item["status"], "pending")
|
||
todo_id = item["id"]
|
||
|
||
rc, out, _ = _run_cli(["todo", "list"])
|
||
self.assertEqual(rc, 0)
|
||
listing = json.loads(out.strip())
|
||
self.assertEqual(listing["count"], 1)
|
||
|
||
rc, out, _ = _run_cli(["todo", "done", str(todo_id)])
|
||
self.assertEqual(rc, 0)
|
||
result = json.loads(out.strip())
|
||
self.assertTrue(result["items"][0]["ok"])
|
||
self.assertEqual(result["items"][0]["todo"]["status"], "done")
|
||
|
||
rc, out, _ = _run_cli(["todo", "stats"])
|
||
stats = json.loads(out.strip())
|
||
self.assertEqual(stats, {"total": 1, "pending": 0, "done": 1})
|
||
|
||
def test_done_batch_with_comma_ids(self):
|
||
with IsolatedEnv():
|
||
_run_cli(["init-db"])
|
||
_run_cli(["todo", "add", "T1"])
|
||
_run_cli(["todo", "add", "T2"])
|
||
rc, out, _ = _run_cli(["todo", "done", "1,2"])
|
||
self.assertEqual(rc, 0)
|
||
result = json.loads(out.strip())
|
||
self.assertEqual(result["count"], 2)
|
||
self.assertTrue(all(r["ok"] for r in result["items"]))
|
||
|
||
def test_done_invalid_id_returns_error(self):
|
||
with IsolatedEnv():
|
||
_run_cli(["init-db"])
|
||
rc, out, _ = _run_cli(["todo", "done", "999"])
|
||
self.assertEqual(rc, 1)
|
||
result = json.loads(out.strip())
|
||
self.assertFalse(result["items"][0]["ok"])
|
||
|
||
def test_remind_async_pattern_emits_progress(self):
|
||
with IsolatedEnv():
|
||
_run_cli(["init-db"])
|
||
_run_cli(["todo", "add", "T1"])
|
||
rc, out, _ = _run_cli(["remind", "-s", "1"])
|
||
self.assertEqual(rc, 0)
|
||
# stdout 应含 emit 进度事件（mock 把 activity 打到 stdout）
|
||
self.assertIn('"event": "emit"', out)
|
||
self.assertIn('"event": "finish"', out)
|
||
self.assertIn('"status": "ok"', out)
|
||
# 最后一行是 CLI 结果 JSON
|
||
last_line = [l for l in out.strip().splitlines() if l.startswith("{")][-1]
|
||
payload = json.loads(last_line)
|
||
self.assertTrue(payload["ok"])
|
||
self.assertEqual(payload["task_type"], "remind")
|
||
|
||
def test_health_returns_zero(self):
|
||
with IsolatedEnv():
|
||
rc, out, _ = _run_cli(["health"])
|
||
self.assertEqual(rc, 0)
|
||
self.assertIn("todo-skill health: ok", out)
|
||
self.assertIn("platform_kit_version:", out)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|