test: 添加测试数据

pear_admin/extensions/__init__.py

@@ -1,8 +1,10 @@
 from flask import Flask
 
 from .init_db import db, migrate
+from .init_script import register_script
 
 
 def register_extensions(app: Flask):
     db.init_app(app)
     migrate.init_app(app, db)
+    register_script(app)

pear_admin/extensions/init_script.py

@@ -0,0 +1,61 @@
+import csv
+import os
+
+from flask import Flask, current_app
+
+from pear_admin.extensions import db
+from pear_admin.orms import DepartmentORM, RightsORM, RoleORM, UserORM
+
+
+def dict_to_orm(d, o):
+    for k, v in d.items():
+        setattr(o, k, v or None)
+
+
+def csv_to_databases(path, orm):
+    with open(path, encoding="utf-8") as file:
+        for d in csv.DictReader(file):
+            o = orm()
+            dict_to_orm(d, o)
+            db.session.add(o)
+            db.session.flush()
+        db.session.commit()
+
+
+def register_script(app: Flask):
+    @app.cli.command()
+    def init():
+        db.drop_all()
+        db.create_all()
+
+        root = current_app.config.get("ROOT_PATH")
+
+        rights_data_path = os.path.join(root, "static", "data", "ums_rights.csv")
+        csv_to_databases(rights_data_path, RightsORM)
+
+        role_data_path = os.path.join(root, "static", "data", "ums_role.csv")
+        csv_to_databases(role_data_path, RoleORM)
+
+        with open(role_data_path, encoding="utf-8") as file:
+            for d in csv.DictReader(file):
+                role: RoleORM = RoleORM.query.get(d["id"])
+                id_list = [int(_id) for _id in d["rights_ids"].split(":")]
+                role.rights_list = RightsORM.query.filter(
+                    RightsORM.id.in_(id_list)
+                ).all()
+                db.session.commit()
+
+        department_data_path = os.path.join(
+            root, "static", "data", "ums_department.csv"
+        )
+        csv_to_databases(department_data_path, DepartmentORM)
+
+        user_data_path = os.path.join(root, "static", "data", "ums_user.csv")
+        csv_to_databases(user_data_path, UserORM)
+
+        with open(user_data_path, encoding="utf-8") as file:
+            for d in csv.DictReader(file):
+                user: UserORM = UserORM.query.get(d["id"])
+                id_list = [int(_id) for _id in d["role_ids"].split(":")]
+                user.role_list = RoleORM.query.filter(RoleORM.id.in_(id_list)).all()
+                db.session.commit()

pear_admin/orms/__init__.py

@@ -12,7 +12,7 @@ user_role = db.Table(
     db.Column("role_id", db.Integer, db.ForeignKey("ums_role.id"), comment="角色编号"),
 )
 
-role_permission = db.Table(
+role_rights = db.Table(
     "ums_role_rights",  # 用户-权限中间表名称
     db.Column("id", db.Integer, primary_key=True, autoincrement=True, comment="标识"),
     db.Column("rights_id", db.Integer, db.ForeignKey("ums_rights.id"), comment="用户编号"),

pear_admin/orms/role.py

@@ -11,13 +11,13 @@ class RoleORM(BaseORM):
     code = db.Column(db.String(20), nullable=False, comment="角色标识符")
     desc = db.Column(db.Text)
 
-    permission_ids = db.Column(
+    rights_ids = db.Column(
         db.String(512),
         comment="权限ids,1,2,5。冗余字段，用户缓存用户权限",
     )
 
-    permission_list = db.relationship(
-        "RightsORM", secondary="ums_role_permission", backref=db.backref("role")
+    rights_list = db.relationship(
+        "RightsORM", secondary="ums_role_rights", backref=db.backref("role")
     )
 
     def json(self):
@@ -25,5 +25,5 @@ class RoleORM(BaseORM):
             "id": self.id,
             "name": self.name,
             "desc": self.desc,
-            "permission_ids": self.permission_ids,
+            "rights_ids": self.rights_ids,
         }