240 lines
9.2 KiB
Python
240 lines
9.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
hr-org 组织管理模块自测脚本
|
|
=================================
|
|
覆盖:
|
|
* 模块导入
|
|
* 23 端点注册
|
|
* org_unit CRUD / 树构建 / 停用 / 移动
|
|
* org_field_def / org_field_value (EAV, 缺参校验)
|
|
* org_unit_change 时间轴
|
|
* org_import 空数据 & ≥200 行校验
|
|
* 合同公司 CRUD(含统一社会信用代码校验)
|
|
|
|
standalone 模式: 用 sqlite3 内存库实现 DataStore, 不依赖 sqlor。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
import traceback
|
|
|
|
|
|
# 确保能 import 到 apps 包
|
|
sys.path.insert(0, __file__.rsplit("/apps/", 1)[0])
|
|
|
|
import hr_org # noqa: E402
|
|
|
|
|
|
class SqliteStore(hr_org.DataStore):
|
|
"""内存 sqlite3 实现 DataStore。"""
|
|
|
|
def __init__(self):
|
|
self.conn = sqlite3.connect(":memory:")
|
|
self.conn.row_factory = sqlite3.Row
|
|
self._create_schema()
|
|
|
|
def _create_schema(self):
|
|
cur = self.conn.cursor()
|
|
cur.executescript("""
|
|
CREATE TABLE org_unit (
|
|
id TEXT PRIMARY KEY, org_code TEXT, org_name TEXT, parent_id TEXT,
|
|
org_type TEXT, leader_id TEXT, effective_date TEXT, expire_date TEXT,
|
|
status TEXT, sort_no INTEGER, remark TEXT, created_by TEXT,
|
|
updated_by TEXT, created_at TEXT, updated_at TEXT
|
|
);
|
|
CREATE TABLE org_field_def (
|
|
id TEXT PRIMARY KEY, field_code TEXT, field_name TEXT, field_type TEXT,
|
|
entity_type TEXT, options TEXT, required INTEGER, sort_no INTEGER,
|
|
status TEXT, created_by TEXT, updated_by TEXT, created_at TEXT, updated_at TEXT
|
|
);
|
|
CREATE TABLE org_field_value (
|
|
id TEXT PRIMARY KEY, org_id TEXT, field_def_id TEXT, field_value TEXT,
|
|
created_by TEXT, updated_by TEXT, created_at TEXT, updated_at TEXT
|
|
);
|
|
CREATE TABLE org_unit_change (
|
|
id TEXT PRIMARY KEY, org_id TEXT, change_type TEXT, change_date TEXT,
|
|
before_data TEXT, after_data TEXT, operator TEXT, created_at TEXT
|
|
);
|
|
CREATE TABLE org_contract_company (
|
|
id TEXT PRIMARY KEY, company_code TEXT, company_name TEXT, credit_code TEXT,
|
|
legal_person TEXT, contact_info TEXT, status TEXT, created_by TEXT,
|
|
updated_by TEXT, created_at TEXT, updated_at TEXT
|
|
);
|
|
CREATE TABLE roster_employee (
|
|
id TEXT PRIMARY KEY, org_id TEXT, org_path TEXT
|
|
);
|
|
CREATE TABLE sys_audit_log (
|
|
id TEXT PRIMARY KEY, module TEXT, action TEXT, entity TEXT,
|
|
entity_id TEXT, before_data TEXT, after_data TEXT, operator TEXT,
|
|
created_at TEXT
|
|
);
|
|
""")
|
|
self.conn.commit()
|
|
|
|
def query(self, sql, params=None):
|
|
cur = self.conn.cursor()
|
|
cur.execute(sql, params or [])
|
|
rows = cur.fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
def execute(self, sql, params=None):
|
|
cur = self.conn.cursor()
|
|
cur.execute(sql, params or [])
|
|
self.conn.commit()
|
|
return cur.rowcount
|
|
|
|
|
|
PASS = 0
|
|
FAIL = 0
|
|
|
|
|
|
def check(name, cond, detail=""):
|
|
global PASS, FAIL
|
|
if cond:
|
|
PASS += 1
|
|
print("[PASS] %s" % name)
|
|
else:
|
|
FAIL += 1
|
|
print("[FAIL] %s %s" % (name, detail))
|
|
|
|
|
|
def main():
|
|
store = SqliteStore()
|
|
hr_org.register(store)
|
|
op = "selftest_user"
|
|
|
|
# 1. 模块导入
|
|
check("模块导入 hr_org", hr_org is not None)
|
|
|
|
# 2. 23 端点注册
|
|
core = hr_org.core_endpoints()
|
|
check("23 核心端点注册", len(core) == 23,
|
|
"实际 %d 个" % len(core))
|
|
check("全部端点含模块级端点", len(hr_org.endpoints()) == 26,
|
|
"实际 %d 个" % len(hr_org.endpoints()))
|
|
|
|
# 3. org_unit 新增
|
|
r = hr_org.org_unit_save(store, {
|
|
"org_code": "ORG001", "org_name": "总部", "sort_no": 1}, op)
|
|
check("org_unit 新增", r["code"] == 0, str(r))
|
|
root_id = r["data"]["id"]
|
|
r2 = hr_org.org_unit_save(store, {
|
|
"org_code": "ORG002", "org_name": "研发部", "parent_id": root_id,
|
|
"sort_no": 2}, op)
|
|
check("org_unit 新增子组织", r2["code"] == 0, str(r2))
|
|
child_id = r2["data"]["id"]
|
|
|
|
# 编码唯一
|
|
r = hr_org.org_unit_save(store, {"org_code": "ORG001", "org_name": "x"}, op)
|
|
check("org_code 唯一性校验", r["code"] != 0, str(r))
|
|
|
|
# 4. 树构建
|
|
tree = hr_org.org_unit_tree(store, {}, op)
|
|
check("组织树构建", tree["code"] == 0 and len(tree["data"]) == 1
|
|
and len(tree["data"][0]["children"]) == 1, str(tree))
|
|
|
|
# 5. 移动 & 循环校验
|
|
r = hr_org.org_unit_move(store, {"id": root_id, "parent_id": child_id}, op)
|
|
check("移动环校验拒绝", r["code"] != 0, str(r))
|
|
grand = hr_org.org_unit_save(store, {
|
|
"org_code": "ORG003", "org_name": "后端组", "parent_id": child_id,
|
|
"sort_no": 3}, op)
|
|
grand_id = grand["data"]["id"]
|
|
r = hr_org.org_unit_move(store, {"id": grand_id, "parent_id": root_id}, op)
|
|
check("组织移动(子树联动)", r["code"] == 0, str(r))
|
|
|
|
# 6. 删除有下级的组织
|
|
r = hr_org.org_unit_delete(store, {"id": child_id}, op)
|
|
check("删除有下级组织被拒", r["code"] != 0, str(r))
|
|
|
|
# 7. 停用级联
|
|
r = hr_org.org_unit_deactivate(store, {"id": root_id}, op)
|
|
check("停用级联子树", r["code"] == 0 and len(r["data"]["deactivated"]) >= 2, str(r))
|
|
|
|
# 8. 生效时间
|
|
r = hr_org.org_unit_effective(store, {
|
|
"id": root_id, "effective_date": "2024-01-01", "expire_date": "2030-01-01"}, op)
|
|
check("生效时间维护", r["code"] == 0, str(r))
|
|
r = hr_org.org_unit_effective(store, {
|
|
"id": root_id, "effective_date": "2025-01-01", "expire_date": "2020-01-01"}, op)
|
|
check("生效/失效日期校验", r["code"] != 0, str(r))
|
|
|
|
# 9. org_field_def
|
|
r = hr_org.org_field_def_save(store, {
|
|
"field_code": "office_area", "field_name": "办公面积", "field_type": "number"}, op)
|
|
check("字段定义新增", r["code"] == 0, str(r))
|
|
fd_id = r["data"]["id"]
|
|
check("字段定义查询", hr_org.org_field_def_list(store, {}, op)["code"] == 0)
|
|
check("字段定义详情", hr_org.org_field_def_get(store, {"id": fd_id}, op)["code"] == 0)
|
|
|
|
# 10. org_field_value 缺参校验
|
|
r = hr_org.org_field_value_save(store, {"field_value": "100"}, op)
|
|
check("org_field_value 缺 org_id 校验", r["code"] != 0 and "org_id" in r["message"], str(r))
|
|
r = hr_org.org_field_value_save(store, {"org_id": root_id}, op)
|
|
check("org_field_value 缺 field_def_id 校验", r["code"] != 0 and "field_def_id" in r["message"], str(r))
|
|
r = hr_org.org_field_value_save(store, {
|
|
"org_id": root_id, "field_def_id": fd_id, "field_value": "100"}, op)
|
|
check("org_field_value 正常写入", r["code"] == 0, str(r))
|
|
|
|
# 11. org_unit_change 时间轴
|
|
changes = hr_org.org_unit_change_list(store, {"org_id": root_id}, op)
|
|
check("时间轴记录生成(create/modify/effective)",
|
|
changes["code"] == 0 and changes["data"]["total"] >= 2, str(changes))
|
|
r = hr_org.org_unit_change_save(store, {
|
|
"org_id": root_id, "change_type": "split",
|
|
"change_date": "2024-06-01"}, op)
|
|
check("时间轴手动记录", r["code"] == 0, str(r))
|
|
r = hr_org.org_unit_change_save(store, {
|
|
"org_id": root_id, "change_type": "bad_type"}, op)
|
|
check("时间轴非法类型拒绝", r["code"] != 0, str(r))
|
|
|
|
# 12. org_import 空数据
|
|
r = hr_org.org_import_import(store, {"rows": []}, op)
|
|
check("导入空数据校验", r["code"] != 0, str(r))
|
|
|
|
# 13. org_import ≥200 行确认
|
|
big = [{"org_code": "IMP%03d" % i, "org_name": "导入组织%d" % i}
|
|
for i in range(205)]
|
|
r = hr_org.org_import_import(store, {"rows": big}, op)
|
|
check("≥200 行需二次确认", r["code"] == 2, str(r))
|
|
r = hr_org.org_import_import(store, {"rows": big, "confirmed": True}, op)
|
|
check("≥200 行确认后导入", r["code"] == 0, str(r))
|
|
|
|
# 14. 合同公司 CRUD
|
|
r = hr_org.org_contract_company_save(store, {
|
|
"company_code": "C001", "company_name": "测试合同公司",
|
|
"credit_code": "91330100MA27XJ1234"}, op)
|
|
check("合同公司新增", r["code"] == 0, str(r))
|
|
cc_id = r["data"]["id"]
|
|
r = hr_org.org_contract_company_save(store, {
|
|
"company_code": "C002", "company_name": "坏信用代码",
|
|
"credit_code": "bad"}, op)
|
|
check("统一社会信用代码校验", r["code"] != 0, str(r))
|
|
r = hr_org.org_contract_company_get(store, {"id": cc_id}, op)
|
|
check("合同公司详情", r["code"] == 0, str(r))
|
|
r = hr_org.org_contract_company_delete(store, {"id": cc_id}, op)
|
|
check("合同公司删除", r["code"] == 0, str(r))
|
|
|
|
# 15. org_tree / org_chart as_of
|
|
r = hr_org.org_tree_view(store, {"as_of": "2024-06-01"}, op)
|
|
check("org_tree as_of 历史架构", r["code"] == 0, str(r))
|
|
r = hr_org.org_chart_view(store, {"as_of": "2024-06-01", "export_image": True}, op)
|
|
check("org_chart 导出图片数据", r["code"] == 0 and "image" in r["data"], str(r))
|
|
|
|
# 16. dispatch 兜底
|
|
check("dispatch 未知端点", hr_org.dispatch("nope", {}).get("code") != 0)
|
|
|
|
print("\n===== 自测结果: PASS=%d FAIL=%d =====" % (PASS, FAIL))
|
|
if FAIL:
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except Exception:
|
|
traceback.print_exc()
|
|
sys.exit(2)
|