Merge pull request #166 from openqj/feat-console-pagination-tabs

console: 任务列表服务端分页 + 控制台分页面签

cloud/tests/test_task_list_pagination.py

@@ -0,0 +1,135 @@
+"""任务列表服务端分页：limit/offset/总数/服务端过滤/状态统计。
+
+用真实 HTTP 服务 + 临时 SQLite：建 5 个任务（2 completed、
+1 failed、2 queued 交错 prompt），分别断言：
+- 无 limit 时旧行为不变：全量、created_at 正序、total=5；
+- limit=2&order=desc 时第 1 页是最新的 2 条，total=5；
+- offset=2 时第 2 页是接下来 2 条，与第 1 页不重叠；
+- status 过滤只回该状态，total 是过滤后计数；
+- q 过滤按 id/prompt 子串（不区分大小写）；
+- limit=0 / limit=101 / offset=-1 回 400；
+- status_counts 是全项目统计，不受当前过滤与分页影响。
+"""
+
+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:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+
+
+class ListPaginationTestCase(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.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)
+        self.ids = []
+        for prompt in ["alpha one", "beta two", "alpha three", "gamma four", "alpha five"]:
+            status, t = self._req("POST", "/tasks",
+                                 {"project_id": self.project["id"], "prompt": prompt},
+                                 self.owner_key)
+            self.assertEqual(status, 201)
+            self.ids.append(t["id"])
+        # 直接改库状态：前 2 条 completed，第 3 条 failed，其余 queued。
+        import sqlite3
+        conn = sqlite3.connect(self.db_path)
+        conn.execute("UPDATE tasks SET status='completed' WHERE id IN (?, ?)",
+                     (self.ids[0], self.ids[1]))
+        conn.execute("UPDATE tasks SET status='failed' WHERE id=?", (self.ids[2],))
+        conn.commit()
+        conn.close()
+
+    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 _list(self, qs=""):
+        return self._req("GET", "/tasks?project_id=" + self.project["id"] + qs,
+                         None, self.owner_key)
+
+    def test_no_limit_is_old_behaviour(self):
+        status, d = self._list()
+        self.assertEqual(status, 200)
+        self.assertEqual([t["id"] for t in d["tasks"]], self.ids)
+        self.assertEqual(d["total"], 5)
+
+    def test_desc_pages_do_not_overlap(self):
+        status, p1 = self._list("&limit=2&order=desc")
+        self.assertEqual(status, 200)
+        self.assertEqual(p1["total"], 5)
+        self.assertEqual([t["id"] for t in p1["tasks"]],
+                         [self.ids[4], self.ids[3]])
+        status, p2 = self._list("&limit=2&offset=2&order=desc")
+        self.assertEqual(status, 200)
+        self.assertEqual([t["id"] for t in p2["tasks"]],
+                         [self.ids[2], self.ids[1]])
+        self.assertEqual(set(t["id"] for t in p1["tasks"]).isdisjoint(
+            set(t["id"] for t in p2["tasks"])), True)
+
+    def test_status_and_q_filters(self):
+        status, d = self._list("&status=completed")
+        self.assertEqual(status, 200)
+        self.assertEqual(d["total"], 2)
+        self.assertEqual(set(t["status"] for t in d["tasks"]), {"completed"})
+        status, d = self._list("&q=ALPHA")
+        self.assertEqual(status, 200)
+        self.assertEqual(d["total"], 3)
+
+    def test_status_counts_ignore_filter_and_page(self):
+        status, d = self._list("&status=completed&limit=1&order=desc")
+        self.assertEqual(status, 200)
+        self.assertEqual(len(d["tasks"]), 1)
+        self.assertEqual(d["status_counts"], {
+            "completed": 2, "failed": 1, "queued": 2})
+
+    def test_invalid_pagination_is_400(self):
+        for qs in ["&limit=0", "&limit=101", "&limit=2&offset=-1", "&limit=abc"]:
+            status, _ = self._list(qs)
+            self.assertEqual(status, 400, qs)