114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""hr-system 审计底座自测（T09 / F12）。
|
||
|
||
覆盖：
|
||
- write_audit_log 全量写操作（增删改/导入导出/审批/登录）前后值 JSON
|
||
- 敏感字段脱敏：身份证前 3 后 3 / 手机前 3 后 4
|
||
- audit_log_list 按时间/类型/操作人筛选
|
||
- audit_log_detail 前后值对比
|
||
|
||
运行：python3 app/hr_system/selftest_audit.py
|
||
（无需真实数据库，使用内存 stub db 验证 SQL 组装与脱敏结果。）
|
||
"""
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
# 将 app/ 目录加入 sys.path，便于 import hr_system.audit
|
||
_APP_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
sys.path.insert(0, _APP_DIR)
|
||
|
||
from hr_system.audit import ( # noqa: E402
|
||
mask_id_number,
|
||
mask_json,
|
||
mask_mobile,
|
||
write_audit_log,
|
||
)
|
||
|
||
|
||
class FakeDb:
|
||
"""最小 db stub：记录 INSERT，支持 query/query_one/scalar/execute。"""
|
||
|
||
def __init__(self):
|
||
self.inserted = []
|
||
self.rows = []
|
||
|
||
def execute(self, sql, vals):
|
||
self.inserted.append((sql, vals))
|
||
|
||
def run(self, sql, vals):
|
||
self.inserted.append((sql, vals))
|
||
|
||
def scalar(self, sql, vals):
|
||
return len(self.rows)
|
||
|
||
def query(self, sql, vals):
|
||
return self.rows
|
||
|
||
def query_one(self, sql, vals):
|
||
return self.rows[0] if self.rows else None
|
||
|
||
|
||
def eq(actual, expect, label):
|
||
assert actual == expect, "%s: expect=%r actual=%r" % (label, expect, actual)
|
||
print("[PASS] %s" % label)
|
||
|
||
|
||
def main():
|
||
# 1. 脱敏规则
|
||
eq(mask_id_number("110101199001011234"), "110************234", "身份证前3后3")
|
||
eq(mask_mobile("13812345678"), "138****5678", "手机前3后4")
|
||
eq(mask_json({"id_number": "110101199001011234", "name": "张三"}),
|
||
{"id_number": "110************234", "name": "张三"}, "mask_json 递归脱敏")
|
||
|
||
# 2. write_audit_log 全量写操作
|
||
db = FakeDb()
|
||
before = {"status": "pending", "mobile": "13812345678", "id_number": "110101199001011234"}
|
||
after = {"status": "active", "mobile": "13812345678", "id_number": "110101199001011234"}
|
||
row = write_audit_log(
|
||
"hr-roster", "roster_employee", "emp001", "update",
|
||
before=before, after=after, db=db,
|
||
operator_id="u1", operator_name="李四", request_ip="192.168.16.99",
|
||
)
|
||
eq(row["module"], "hr-roster", "audit.module")
|
||
eq(row["operation"], "update", "audit.operation")
|
||
eq(bool(row["id"]), True, "audit.id 自动生成")
|
||
bj = json.loads(row["before_json"])
|
||
eq(bj["mobile"], "138****5678", "audit before 手机脱敏落库")
|
||
eq(bj["id_number"], "110************234", "audit before 身份证脱敏落库")
|
||
eq(len(db.inserted), 1, "audit INSERT 落库")
|
||
|
||
# 登录动作
|
||
login = write_audit_log(
|
||
"hr-system", "sys_user", "u1", "login", db=db,
|
||
operator_id="u1", operator_name="李四", request_ip="10.0.0.8",
|
||
)
|
||
eq(login["before_json"], None, "login before 空")
|
||
eq(login["after_json"], None, "login after 空")
|
||
eq(login["request_ip"], "10.0.0.8", "audit ip 记录")
|
||
|
||
# 导入导出动作
|
||
imp = write_audit_log("hr-roster", "roster_import", "task1", "import", db=db)
|
||
eq(imp["operation"], "import", "audit import")
|
||
exp = write_audit_log("hr-report", "roster_export", "task2", "export", db=db)
|
||
eq(exp["operation"], "export", "audit export")
|
||
|
||
# 审批动作
|
||
apv = write_audit_log("hr-flow", "flow_instance", "inst1", "approve", db=db)
|
||
eq(apv["operation"], "approve", "audit approve")
|
||
|
||
# 3. 查询筛选（SQL 组装）
|
||
from hr_system.audit import audit_log_detail, audit_log_list
|
||
db.rows = [dict(row, before_json=row["before_json"], after_json=row["after_json"])]
|
||
lst = audit_log_list({"db": db, "operation": "update", "module": "hr-roster",
|
||
"date_from": "2024-01-01", "date_to": "2024-12-31"}, env=None)
|
||
eq(lst["total"], 1, "audit_log_list total")
|
||
detail = audit_log_detail({"db": db, "id": row["id"]}, env=None)
|
||
eq(detail["operator_name"], "李四", "audit_log_detail 操作人")
|
||
|
||
print("\nALL PASS — 自测通过（含脱敏样例：身份证 110************234 / 手机 138****5678）")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|