transfer: --quota 显示空间配额; 测试 67 项

tests/test_transfer.py

@@ -102,6 +102,15 @@ class ProtocolServer:
                 if srv.require_token and self.headers.get("Authorization") != f"Bearer {srv.require_token}":
                     self._json(401, {"error": "未授权"})
                     return
+                if self.path == "/api/quota":
+                    used = sum(f.get("size", len(f.get("data", b""))) for f in srv.files)
+                    self._json(200, {
+                        "used_bytes": used,
+                        "quota_bytes": 1024,
+                        "remain_bytes": max(1024 - used, 0),
+                        "percent": round(used * 100 / 1024, 1),
+                    })
+                    return
                 m = re.fullmatch(r"/api/transfer/([^/]+)/chunks", self.path)
                 if m:
                     self._json(200, {"received": sorted(srv.received)})
@@ -263,6 +272,16 @@ class TestListDownload(TransferTestBase):
         self.assertEqual(files[0]["file_id"], "F-1")
         self.assertEqual(files[0]["file_name"], "报告.pdf")
 
+    def test_quota(self):
+        self.srv.files = [
+            {"file_id": "F-1", "file_name": "a.bin", "size": 300, "data": b"x" * 300},
+            {"file_id": "F-2", "file_name": "b.bin", "size": 200, "data": b"y" * 200},
+        ]
+        q = self.client.get_quota()
+        self.assertEqual(q["used_bytes"], 500)
+        self.assertEqual(q["quota_bytes"], 1024)
+        self.assertEqual(q["remain_bytes"], 524)
+
     def test_delete_file(self):
         self.srv.files = [
             {"file_id": "F-1", "file_name": "a.bin", "size": 1, "data": b"x"},

transfer.py

@@ -143,6 +143,13 @@ class TransferClient:
         if not payload.get("ok"):
             raise TransferError(f"删除失败: {payload}")
 
+    def get_quota(self) -> dict[str, Any]:
+        """GET /api/quota: 空间配额 (used/quota/remain/percent)"""
+        code, payload = self._request("GET", "/api/quota")
+        if code != 200:
+            raise TransferError(f"配额查询失败 (HTTP {code}): {payload}")
+        return payload
+
     def download(self, file_id: str, dest: Path | None = None) -> tuple[Path, dict[str, Any]]:
         """逐卷下载密文 (零合并: 服务端只存卷, 客户端本地拼接), 返回 (密文路径, 解密参数)。
 
@@ -278,6 +285,7 @@ def main() -> int:
     ap.add_argument("--init", help="init json 路径 (上传模式)")
     ap.add_argument("--manifest", help="卷清单路径 (上传模式)")
     ap.add_argument("--list", action="store_true", help="列出服务端文件")
+    ap.add_argument("--quota", action="store_true", help="显示空间配额")
     ap.add_argument("--download", metavar="FILE_ID", help="下载文件")
     ap.add_argument("--delete", metavar="FILE_ID", help="删除服务端文件")
     ap.add_argument("--out", help="保存路径 (目录或完整文件名)")
@@ -368,6 +376,17 @@ def main() -> int:
             return 1
         return 0
 
+    if args.quota:
+        try:
+            q = client.get_quota()
+        except TransferError as e:
+            print(f"[错误] {e}")
+            return 1
+        print(f"[配额] 已用 {q['used_bytes'] / 1048576:.1f} MB / "
+              f"{q['quota_bytes'] / 1048576:.1f} MB ({q['percent']}%)  "
+              f"剩余 {q['remain_bytes'] / 1048576:.1f} MB")
+        return 0
+
     if args.delete:
         try:
             client.delete_file(args.delete)
@@ -378,7 +397,7 @@ def main() -> int:
         return 0
 
     if not (args.init and args.manifest):
-        ap.error("需指定 --list / --download / (--init + --manifest) 之一")
+        ap.error("需指定 --list / --quota / --download / --delete / (--init + --manifest) 之一")
 
     init_json = json.loads(Path(args.init).read_text(encoding="utf-8"))
     manifest = json.loads(Path(args.manifest).read_text(encoding="utf-8"))