待办：项目待办模型 + /projects/{id}/todos 接口 + 控制台待办页
This commit is contained in:
host committed
1 parent
c10fda4e4a
commit
b3b3b35042
8 files changed
+962
-23
No files matched your search
@@ -0,0 +1,138 @@
|
||||
"""项目待办模块单测(cloud/todos.py)。
|
||||
|
||||
不跑真实模型、不联网：临时 SQLite 文件 + 模块函数直调。
|
||||
覆盖：建表幂等、创建校验与默认值、状态五态与完成时间、
|
||||
PATCH 只改给定字段、过滤（状态/q）、排序追加、硬删除、
|
||||
跨项目隔离（模块层按 project_id 查，服务端层再做 404）。
|
||||
"""
|
||||
|
||||
import os
|
||||
import sqlite3
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
from cloud.todos import (
|
||||
create_todo,
|
||||
delete_todo,
|
||||
ensure_todos_table,
|
||||
get_todo,
|
||||
list_todos,
|
||||
update_todo,
|
||||
)
|
||||
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||||
from cloud.todos import (
|
||||
create_todo,
|
||||
delete_todo,
|
||||
ensure_todos_table,
|
||||
get_todo,
|
||||
list_todos,
|
||||
update_todo,
|
||||
)
|
||||
|
||||
|
||||
class TodosTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||||
self.conn = sqlite3.connect(self.db_path)
|
||||
self.conn.row_factory = sqlite3.Row
|
||||
ensure_todos_table(self.conn)
|
||||
self.conn.commit()
|
||||
self.now = 1700000000.0
|
||||
self.project = "proj-1"
|
||||
|
||||
def tearDown(self):
|
||||
self.conn.close()
|
||||
self.tmp.cleanup()
|
||||
|
||||
def _make(self, title="写文档", **kw):
|
||||
kw.setdefault("project_id", self.project)
|
||||
return create_todo(
|
||||
self.conn, kw.pop("project_id"), "user-1", title,
|
||||
now=self.now, **kw)
|
||||
|
||||
def test_ensure_is_idempotent(self):
|
||||
ensure_todos_table(self.conn)
|
||||
ensure_todos_table(self.conn)
|
||||
self.conn.commit()
|
||||
rows = self.conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type = 'table'"
|
||||
" AND name = 'todos'").fetchall()
|
||||
self.assertEqual(len(rows), 1)
|
||||
|
||||
def test_create_defaults_and_validation(self):
|
||||
t = self._make()
|
||||
self.assertEqual(t["status"], "pending")
|
||||
self.assertEqual(t["priority"], "medium")
|
||||
self.assertEqual(t["description"], "")
|
||||
self.assertIsNone(t["task_id"])
|
||||
self.assertIsNone(t["due_at"])
|
||||
self.assertIsNone(t["completed_at"])
|
||||
self.assertEqual(t["sort_order"], 0)
|
||||
with self.assertRaises(ValueError):
|
||||
self._make(title=" ")
|
||||
with self.assertRaises(ValueError):
|
||||
self._make(status="done")
|
||||
with self.assertRaises(ValueError):
|
||||
self._make(priority="urgent")
|
||||
with self.assertRaises(ValueError):
|
||||
self._make(title="x" * 501)
|
||||
|
||||
def test_sort_order_appends(self):
|
||||
a = self._make("第一")
|
||||
b = self._make("第二")
|
||||
c = self._make("第三", sort_order=10)
|
||||
self.assertEqual((a["sort_order"], b["sort_order"],
|
||||
c["sort_order"]), (0, 1, 10))
|
||||
titles = [t["title"] for t in list_todos(self.conn, self.project)]
|
||||
self.assertEqual(titles, ["第一", "第二", "第三"])
|
||||
|
||||
def test_status_completed_sets_and_clears_completed_at(self):
|
||||
t = self._make()
|
||||
done = update_todo(self.conn, t["id"], {"status": "completed"})
|
||||
self.assertIsNotNone(done["completed_at"])
|
||||
back = update_todo(self.conn, t["id"], {"status": "in_progress"})
|
||||
self.assertIsNone(back["completed_at"])
|
||||
self.assertEqual(back["status"], "in_progress")
|
||||
|
||||
def test_update_only_given_fields(self):
|
||||
t = self._make("原标题", description="原说明",
|
||||
priority="low", task_id="task-1")
|
||||
u = update_todo(self.conn, t["id"], {"title": "新标题"})
|
||||
self.assertEqual(u["title"], "新标题")
|
||||
self.assertEqual(u["description"], "原说明")
|
||||
self.assertEqual(u["priority"], "low")
|
||||
self.assertEqual(u["task_id"], "task-1")
|
||||
self.assertIsNone(update_todo(self.conn, "nope", {"title": "x"}))
|
||||
|
||||
def test_due_at_ms_tolerance(self):
|
||||
t = self._make(due_at=1700000000000)
|
||||
self.assertEqual(t["due_at"], 1700000000.0)
|
||||
with self.assertRaises(ValueError):
|
||||
self._make(due_at="明天")
|
||||
|
||||
def test_list_filters(self):
|
||||
self._make("写文档", status="completed")
|
||||
self._make("修 bug", status="in_progress")
|
||||
self._make("写测试", status="pending")
|
||||
done = list_todos(self.conn, self.project, status="completed")
|
||||
self.assertEqual([t["title"] for t in done], ["写文档"])
|
||||
q = list_todos(self.conn, self.project, q="写")
|
||||
self.assertEqual(sorted(t["title"] for t in q), ["写文档", "写测试"])
|
||||
other = list_todos(self.conn, "proj-2")
|
||||
self.assertEqual(other, [])
|
||||
|
||||
def test_delete(self):
|
||||
t = self._make()
|
||||
deleted = delete_todo(self.conn, t["id"])
|
||||
self.assertEqual(deleted["id"], t["id"])
|
||||
self.assertIsNone(get_todo(self.conn, t["id"]))
|
||||
self.assertIsNone(delete_todo(self.conn, t["id"]))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,162 @@
|
||||
"""项目待办服务端单测：POST/GET /projects/{id}/todos、
|
||||
GET/PATCH/DELETE /projects/{id}/todos/{tid}。
|
||||
|
||||
不跑真实模型、不联网：本地临时端口 + 临时 SQLite，与
|
||||
test_automations_server.py 同款骨架。覆盖鉴权、权限、
|
||||
校验、跨项目 404、状态完成时间、删除语义。
|
||||
"""
|
||||
|
||||
import http.client
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
from cloud.auth import AuthStore
|
||||
from cloud.server import create_server
|
||||
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||||
from cloud.auth import AuthStore
|
||||
from cloud.server import create_server
|
||||
|
||||
|
||||
class TodoServerTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||||
self.auth = AuthStore(self.db_path)
|
||||
self.owner_id, self.owner_key = self.auth.create_user("owner")
|
||||
self.outsider_id, self.outsider_key = self.auth.create_user(
|
||||
"outsider")
|
||||
self.server = create_server(self.db_path, "127.0.0.1", 0)
|
||||
self.port = self.server.server_address[1]
|
||||
self.thread = threading.Thread(
|
||||
target=self.server.serve_forever, daemon=True)
|
||||
self.thread.start()
|
||||
status, self.project = self._req(
|
||||
"POST", "/projects", {"name": "demo"}, self.owner_key)
|
||||
self.assertEqual(status, 201)
|
||||
status, self.project2 = self._req(
|
||||
"POST", "/projects", {"name": "other"}, self.owner_key)
|
||||
self.assertEqual(status, 201)
|
||||
|
||||
def tearDown(self):
|
||||
self.server.shutdown()
|
||||
self.server.server_close()
|
||||
self.thread.join(timeout=5)
|
||||
try:
|
||||
self.auth.conn.close()
|
||||
except Exception:
|
||||
pass
|
||||
self.tmp.cleanup()
|
||||
|
||||
def _req(self, method, path, body=None, key=None):
|
||||
conn = http.client.HTTPConnection(
|
||||
"127.0.0.1", self.port, timeout=10)
|
||||
hdrs = {}
|
||||
payload = None
|
||||
if body is not None:
|
||||
payload = json.dumps(body).encode("utf-8")
|
||||
hdrs["Content-Type"] = "application/json"
|
||||
if key is not None:
|
||||
hdrs["X-API-Key"] = key
|
||||
try:
|
||||
conn.request(method, path, body=payload, headers=hdrs)
|
||||
resp = conn.getresponse()
|
||||
data = resp.read()
|
||||
finally:
|
||||
conn.close()
|
||||
try:
|
||||
parsed = json.loads(data.decode("utf-8")) if data else None
|
||||
except ValueError:
|
||||
parsed = None
|
||||
return resp.status, parsed
|
||||
|
||||
def _url(self, project_id=None):
|
||||
return "/projects/%s/todos" % (project_id or self.project["id"])
|
||||
|
||||
def _create(self, project_id=None, **extra):
|
||||
body = {"title": "写文档"}
|
||||
body.update(extra)
|
||||
return self._req("POST", self._url(project_id), body,
|
||||
self.owner_key)
|
||||
|
||||
def test_create_list_get(self):
|
||||
status, todo = self._create()
|
||||
self.assertEqual(status, 201)
|
||||
self.assertEqual(todo["status"], "pending")
|
||||
self.assertEqual(todo["priority"], "medium")
|
||||
status, body = self._req(
|
||||
"GET", self._url(), key=self.owner_key)
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(body["total"], 1)
|
||||
self.assertEqual(body["counts"]["pending"], 1)
|
||||
status, got = self._req(
|
||||
"GET", self._url() + "/" + todo["id"], key=self.owner_key)
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(got["id"], todo["id"])
|
||||
|
||||
def test_auth_and_permission(self):
|
||||
status, _ = self._req("GET", self._url())
|
||||
self.assertEqual(status, 401)
|
||||
status, _ = self._req("POST", self._url(),
|
||||
{"title": "x"}, self.outsider_key)
|
||||
self.assertEqual(status, 403)
|
||||
status, _ = self._req("GET", self._url(), key=self.outsider_key)
|
||||
self.assertEqual(status, 403)
|
||||
|
||||
def test_validation_and_cross_project_404(self):
|
||||
status, _ = self._create(title=" ")
|
||||
self.assertEqual(status, 400)
|
||||
status, _ = self._create(status="done")
|
||||
self.assertEqual(status, 400)
|
||||
status, todo = self._create()
|
||||
self.assertEqual(status, 201)
|
||||
# 单条属于别的项目时一律 404，不泄露存在性
|
||||
status, _ = self._req(
|
||||
"GET", self._url(self.project2["id"]) + "/" + todo["id"],
|
||||
key=self.owner_key)
|
||||
self.assertEqual(status, 404)
|
||||
status, _ = self._req(
|
||||
"GET", self._url() + "/nope", key=self.owner_key)
|
||||
self.assertEqual(status, 404)
|
||||
|
||||
def test_patch_completed_and_delete(self):
|
||||
_, todo = self._create()
|
||||
status, done = self._req(
|
||||
"PATCH", self._url() + "/" + todo["id"],
|
||||
{"status": "completed"}, self.owner_key)
|
||||
self.assertEqual(status, 200)
|
||||
self.assertIsNotNone(done["completed_at"])
|
||||
status, back = self._req(
|
||||
"PATCH", self._url() + "/" + todo["id"],
|
||||
{"status": "blocked"}, self.owner_key)
|
||||
self.assertEqual(status, 200)
|
||||
self.assertIsNone(back["completed_at"])
|
||||
status, _ = self._req(
|
||||
"DELETE", self._url() + "/" + todo["id"],
|
||||
key=self.owner_key)
|
||||
self.assertEqual(status, 200)
|
||||
status, _ = self._req(
|
||||
"GET", self._url() + "/" + todo["id"], key=self.owner_key)
|
||||
self.assertEqual(status, 404)
|
||||
|
||||
def test_list_filter(self):
|
||||
self._create()
|
||||
self._create(title="修 bug", status="in_progress")
|
||||
status, body = self._req(
|
||||
"GET", self._url() + "?status=in_progress",
|
||||
key=self.owner_key)
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(len(body["todos"]), 1)
|
||||
# 计数与 total 是全项目口径，不受过滤影响
|
||||
self.assertEqual(body["total"], 2)
|
||||
self.assertEqual(body["counts"]["pending"], 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user