# -*- 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()