codex-app-server/cloud/tests/test_project_todos_server.py

"""项目待办服务端单测：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()