签入版本

tests/fixtures/__init__.py

@@ -0,0 +1 @@
+"""测试夹具。"""

tests/fixtures/data.sql

@@ -0,0 +1,2 @@
+-- 集成测试基础数据
+-- 完整结构请使用 document/djangoadmin.tornado.elevue.sql

tests/fixtures/factories.py

@@ -0,0 +1,126 @@
+"""工厂模式：构造测试用模拟对象。"""
+import json
+
+from common.utils.helpers import parse_id_list
+
+
+class DictItem(dict):
+    """带 to_dict 的伪模型对象。"""
+
+    def to_dict(self):
+        return dict(self)
+
+
+class FakeHandler:
+    """模拟 Tornado handler（支持 get_query_argument / request.body / finish）。"""
+
+    def __init__(self, query=None, body=None):
+        self._query = query or {}
+        self.request = type("Request", (), {
+            "body": json.dumps(body).encode("utf-8") if body is not None else b"{}",
+            "path": "/api/v1/test",
+            "method": "POST",
+        })()
+        self.result = None
+
+    def get_query_argument(self, name, default=None):
+        return self._query.get(name, default)
+
+    def finish(self, result):
+        self.result = result
+        return self.result
+
+
+class FakeQuery:
+    """内存查询对象：支持 filter/order_by/count/all（不做真实过滤）。"""
+
+    def __init__(self, items):
+        self.items = list(items)
+
+    def filter(self, *criterion):
+        return self
+
+    def order_by(self, *cols):
+        return self
+
+    def all(self):
+        return self.items
+
+    def count(self):
+        return len(self.items)
+
+
+class FakeRepository:
+    """内存版仓储，支持 BaseService 模板方法所需原语。"""
+
+    def __init__(self, items=None):
+        self.items = items if items is not None else []
+        self._deleted = []
+
+    # ---------- 数据清洗 ----------
+    def _clean_model_data(self, data: dict) -> dict:
+        """模拟 BaseRepository._clean_model_data：透传数据。"""
+        return data if data else {}
+
+    # ---------- 查询 ----------
+    def filter(self, *criterion):
+        return FakeQuery(self.items)
+
+    def filter_by(self, **kwargs):
+        return FakeQuery(self.items)
+
+    def get_by_id(self, pk, soft_delete=True):
+        try:
+            pk = int(pk)
+        except (TypeError, ValueError):
+            return None
+        for item in self.items:
+            if getattr(item, "id", None) == pk:
+                return item
+        return None
+
+    def get_by_ids(self, ids, soft_delete=True):
+        valid = [i for i in ids if isinstance(i, int) and i > 0]
+        return [item for item in self.items if getattr(item, "id", None) in valid]
+
+    def exists(self, obj_id=None, soft_delete=True, **filters):
+        if obj_id is not None:
+            return self.get_by_id(obj_id) is not None
+        return any(self.items)
+
+    def exists_by_field(self, field, value, exclude_id=None, soft_delete=True):
+        for item in self.items:
+            if getattr(item, field, None) == value and getattr(item, "id", None) != exclude_id:
+                return True
+        return False
+
+    def paginate(self, query, page=1, limit=10):
+        items = list(query.items)
+        total = len(items)
+        start = (page - 1) * limit
+        return items[start:start + limit], total
+
+    def all(self, soft_delete=True):
+        return list(self.items)
+
+    # ---------- 写操作 ----------
+    def create(self, obj):
+        self.items.append(obj)
+        return obj
+
+    def update(self, pk, data):
+        return True
+
+    def soft_delete(self, ids):
+        self._deleted.extend(ids)
+        return len(ids)
+
+    def batch_delete(self, ids_str):
+        id_list = parse_id_list(str(ids_str))
+        if not id_list:
+            return 0
+        self._deleted.extend(id_list)
+        return len(id_list)
+
+    def close(self):
+        pass

tests/fixtures/users.json

@@ -0,0 +1,10 @@
+[
+  {
+    "username": "admin",
+    "password": "123456",
+    "realname": "超级管理员",
+    "nickname": "管理员",
+    "gender": 1,
+    "status": 1
+  }
+]