Simon
986 lines
36 KiB
Python
986 lines
36 KiB
Python
# =============================================================================
|
||
# 组织架构树构建与过滤 — 单元测试
|
||
# =============================================================================
|
||
# 测试覆盖：
|
||
# 1. build_org_tree() — 层级树构建（dept_ 前缀、parentid 根判定、多部门员工、空部门、孤儿部门、嵌套层级）
|
||
# 2. _build_flat_tree_by_name() — 降级扁平分组
|
||
# 3. filter_user_from_tree() — 递归过滤用户 + 空部门移除
|
||
# 4. count_tree_employees() — 递归统计员工数
|
||
# 5. get_org_tree_cached() — 缓存 key 区分端点 + 用户排除（mock 依赖）
|
||
#
|
||
# 设计原则：
|
||
# - build_org_tree / filter_user_from_tree / count_tree_employees / _build_flat_tree_by_name
|
||
# 是纯函数，无外部依赖，直接测试输入输出
|
||
# - get_org_tree_cached 是 async 函数，mock get_org_directory 和 get_cached_dept_list
|
||
# =============================================================================
|
||
|
||
import json
|
||
from unittest.mock import AsyncMock, patch
|
||
|
||
import pytest
|
||
|
||
from app.services.employee_directory import (
|
||
ORG_TREE_CACHE_KEY_PREFIX,
|
||
build_org_tree,
|
||
count_tree_employees,
|
||
filter_user_from_tree,
|
||
get_org_tree_cached,
|
||
_build_flat_tree_by_name,
|
||
)
|
||
|
||
|
||
# =============================================================================
|
||
# 辅助：构造测试数据
|
||
# =============================================================================
|
||
|
||
def _emp(employee_id: str, name: str, department: str = "", dept_ids=None):
|
||
"""构造员工目录条目。"""
|
||
return {
|
||
"employee_id": employee_id,
|
||
"name": name,
|
||
"department": department,
|
||
"dept_ids": dept_ids or [],
|
||
}
|
||
|
||
|
||
def _dept(dept_id: int, name: str, parentid: int = 0):
|
||
"""构造部门列表条目。"""
|
||
return {"id": dept_id, "name": name, "parentid": parentid}
|
||
|
||
|
||
# =============================================================================
|
||
# 一、build_org_tree() 测试
|
||
# =============================================================================
|
||
|
||
class TestBuildOrgTree:
|
||
"""build_org_tree 纯函数测试。"""
|
||
|
||
# ----- 1.1 dept_ 前缀与员工 UserID 不冲突 -----
|
||
|
||
def test_dept_prefix_no_collision_with_employee_id(self):
|
||
"""部门节点 id 加 dept_ 前缀，员工节点 id 用 UserID，两者不冲突。"""
|
||
directory = [_emp("zhangsan", "张三", "研发部", [2])]
|
||
dept_list = [_dept(1, "公司", 0), _dept(2, "研发部", 1)]
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
# 收集所有节点 id
|
||
all_ids = set()
|
||
|
||
def collect_ids(nodes):
|
||
for n in nodes:
|
||
all_ids.add(n["id"])
|
||
if "children" in n:
|
||
collect_ids(n["children"])
|
||
|
||
collect_ids(tree)
|
||
# 部门 id 有 dept_ 前缀，员工 id 无前缀
|
||
assert "dept_1" in all_ids
|
||
assert "dept_2" in all_ids
|
||
assert "zhangsan" in all_ids
|
||
# 不存在 "dept_zhangsan" 之类的冲突
|
||
assert "dept_zhangsan" not in all_ids
|
||
|
||
def test_dept_prefix_with_numeric_employee_id(self):
|
||
"""员工 UserID 为纯数字时，dept_ 前缀仍能区分。"""
|
||
directory = [_emp("1001", "员工1001", "研发部", [2])]
|
||
dept_list = [_dept(1, "公司", 0), _dept(2, "研发部", 1), _dept(1001, "数字部门", 1)]
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
all_ids = set()
|
||
|
||
def collect_ids(nodes):
|
||
for n in nodes:
|
||
all_ids.add(n["id"])
|
||
if "children" in n:
|
||
collect_ids(n["children"])
|
||
|
||
collect_ids(tree)
|
||
# 部门 1001 的 id 是 "dept_1001"，员工 1001 的 id 是 "1001"
|
||
assert "dept_1001" in all_ids
|
||
assert "1001" in all_ids
|
||
|
||
# ----- 1.2 parentid=0 根部门判定 -----
|
||
|
||
def test_parentid_0_as_root(self):
|
||
"""parentid=0 的部门作为根节点。"""
|
||
directory = [_emp("u1", "用户1", "研发", [2])]
|
||
dept_list = [
|
||
_dept(1, "公司", 0),
|
||
_dept(2, "研发部", 1),
|
||
]
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
# 只有一个根节点（公司）
|
||
assert len(tree) == 1
|
||
assert tree[0]["id"] == "dept_1"
|
||
assert tree[0]["label"] == "公司"
|
||
# 研发部是公司的子部门
|
||
assert len(tree[0]["children"]) == 1
|
||
assert tree[0]["children"][0]["id"] == "dept_2"
|
||
|
||
def test_multiple_root_departments(self):
|
||
"""多个 parentid=0 的部门都作为根节点。"""
|
||
directory = []
|
||
dept_list = [
|
||
_dept(1, "公司A", 0),
|
||
_dept(2, "公司B", 0),
|
||
]
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
assert len(tree) == 2
|
||
root_ids = {n["id"] for n in tree}
|
||
assert root_ids == {"dept_1", "dept_2"}
|
||
|
||
def test_parentid_1_fallback_when_no_parentid_0(self):
|
||
"""没有 parentid=0 时，parentid=1 的部门作为根（企微 dept 1 缺失场景）。"""
|
||
directory = [_emp("u1", "用户1", "研发", [2])]
|
||
# dept 1 不在列表中，dept 2 的 parentid=1
|
||
dept_list = [_dept(2, "研发部", 1)]
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
assert len(tree) == 1
|
||
assert tree[0]["id"] == "dept_2"
|
||
|
||
# ----- 1.3 多部门员工 -----
|
||
|
||
def test_multi_dept_employee_appears_in_all_depts(self):
|
||
"""员工属于多个部门时，在每个部门下都出现。"""
|
||
directory = [_emp("zhangsan", "张三", "研发,测试", [2, 3])]
|
||
dept_list = [
|
||
_dept(1, "公司", 0),
|
||
_dept(2, "研发部", 1),
|
||
_dept(3, "测试部", 1),
|
||
]
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
# 根节点是公司
|
||
assert len(tree) == 1
|
||
company = tree[0]
|
||
# 公司下有研发部和测试部
|
||
dept_nodes = {n["id"]: n for n in company["children"] if not n.get("isLeaf")}
|
||
assert "dept_2" in dept_nodes
|
||
assert "dept_3" in dept_nodes
|
||
|
||
# 研发部下有张三
|
||
dev_emps = [n for n in dept_nodes["dept_2"]["children"] if n.get("isLeaf")]
|
||
assert len(dev_emps) == 1
|
||
assert dev_emps[0]["id"] == "zhangsan"
|
||
|
||
# 测试部下也有张三
|
||
test_emps = [n for n in dept_nodes["dept_3"]["children"] if n.get("isLeaf")]
|
||
assert len(test_emps) == 1
|
||
assert test_emps[0]["id"] == "zhangsan"
|
||
|
||
# ----- 1.4 空部门（有部门节点但没有员工） -----
|
||
|
||
def test_empty_dept_still_appears(self):
|
||
"""有部门节点但没有员工的空部门仍出现在树中。"""
|
||
directory = [_emp("u1", "用户1", "研发", [2])]
|
||
dept_list = [
|
||
_dept(1, "公司", 0),
|
||
_dept(2, "研发部", 1),
|
||
_dept(3, "空部门", 1), # 没有员工
|
||
]
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
company = tree[0]
|
||
dept_ids = {n["id"] for n in company["children"] if not n.get("isLeaf")}
|
||
# 空部门仍然出现
|
||
assert "dept_3" in dept_ids
|
||
|
||
def test_empty_dept_has_empty_children(self):
|
||
"""空部门的 children 为空列表。"""
|
||
directory = []
|
||
dept_list = [_dept(1, "公司", 0), _dept(2, "空部门", 1)]
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
company = tree[0]
|
||
empty_dept = next(n for n in company["children"] if n["id"] == "dept_2")
|
||
assert empty_dept["children"] == []
|
||
|
||
# ----- 1.5 未分配部门员工 -----
|
||
|
||
def test_unassigned_employees_grouped(self):
|
||
"""没有 dept_ids 的员工归到"未分配部门"。"""
|
||
directory = [
|
||
_emp("u1", "用户1", "", []),
|
||
_emp("u2", "用户2", "某部门", []),
|
||
]
|
||
dept_list = [_dept(1, "公司", 0)]
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
# 应该有公司 + 未分配部门
|
||
top_ids = {n["id"] for n in tree}
|
||
assert "dept_1" in top_ids
|
||
assert "dept_unassigned" in top_ids
|
||
|
||
unassigned = next(n for n in tree if n["id"] == "dept_unassigned")
|
||
assert unassigned["label"] == "未分配部门"
|
||
assert len(unassigned["children"]) == 2
|
||
|
||
# ----- 1.6 孤儿部门（parentid 不在 dept_map 中） -----
|
||
|
||
def test_orphan_dept_as_root(self):
|
||
"""parentid 指向不存在的部门时，该部门作为根节点。"""
|
||
directory = [_emp("u1", "用户1", "孤儿", [99])]
|
||
dept_list = [
|
||
_dept(1, "公司", 0),
|
||
_dept(99, "孤儿部门", 88), # parentid=88 不在 dept_map
|
||
]
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
top_ids = {n["id"] for n in tree}
|
||
assert "dept_1" in top_ids
|
||
assert "dept_99" in top_ids # 孤儿部门作为根
|
||
|
||
# ----- 1.7 多层嵌套 -----
|
||
|
||
def test_nested_hierarchy_3_levels(self):
|
||
"""3 层嵌套：公司 → 研发部 → 前端组 → 员工。"""
|
||
directory = [_emp("u1", "前端工程师", "前端组", [3])]
|
||
dept_list = [
|
||
_dept(1, "公司", 0),
|
||
_dept(2, "研发部", 1),
|
||
_dept(3, "前端组", 2),
|
||
]
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
assert len(tree) == 1
|
||
company = tree[0]
|
||
assert company["id"] == "dept_1"
|
||
assert len(company["children"]) == 1
|
||
|
||
dev = company["children"][0]
|
||
assert dev["id"] == "dept_2"
|
||
assert len(dev["children"]) == 1
|
||
|
||
fe = dev["children"][0]
|
||
assert fe["id"] == "dept_3"
|
||
assert len(fe["children"]) == 1
|
||
|
||
emp = fe["children"][0]
|
||
assert emp["id"] == "u1"
|
||
assert emp["isLeaf"] is True
|
||
|
||
def test_deep_nested_hierarchy_10_levels(self):
|
||
"""10 层嵌套不栈溢出。"""
|
||
directory = [_emp("u1", "深层员工", "深层", [10])]
|
||
dept_list = [_dept(i, f"部门{i}", i - 1) for i in range(1, 11)]
|
||
dept_list[0] = _dept(1, "部门1", 0) # 根
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
# 递归到最深层
|
||
node = tree[0]
|
||
for i in range(1, 10):
|
||
assert node["id"] == f"dept_{i}"
|
||
dept_children = [c for c in node["children"] if not c.get("isLeaf")]
|
||
assert len(dept_children) == 1
|
||
node = dept_children[0]
|
||
# 第 10 层有员工
|
||
assert node["id"] == "dept_10"
|
||
emps = [c for c in node["children"] if c.get("isLeaf")]
|
||
assert len(emps) == 1
|
||
|
||
# ----- 1.8 空输入 -----
|
||
|
||
def test_empty_directory_and_empty_dept_list(self):
|
||
"""空 directory + 空 dept_list → 空树。"""
|
||
tree = build_org_tree([], [])
|
||
assert tree == []
|
||
|
||
def test_empty_directory_with_dept_list(self):
|
||
"""空 directory + 有 dept_list → 只有部门节点无员工。"""
|
||
dept_list = [_dept(1, "公司", 0)]
|
||
tree = build_org_tree([], dept_list)
|
||
assert len(tree) == 1
|
||
assert tree[0]["id"] == "dept_1"
|
||
assert tree[0]["children"] == []
|
||
|
||
def test_directory_with_empty_dept_list_falls_back_to_flat(self):
|
||
"""有 directory + 空 dept_list → 降级到 _build_flat_tree_by_name。"""
|
||
directory = [_emp("u1", "用户1", "研发部")]
|
||
tree = build_org_tree(directory, [])
|
||
# 降级路径使用 dept_name_ 前缀
|
||
assert len(tree) == 1
|
||
assert tree[0]["id"] == "dept_name_研发部"
|
||
|
||
# ----- 1.9 部门节点结构验证 -----
|
||
|
||
def test_dept_node_structure(self):
|
||
"""部门节点包含 id, label, dept_id, parentid, children 字段。"""
|
||
directory = [_emp("u1", "用户1", "研发", [2])]
|
||
dept_list = [_dept(1, "公司", 0), _dept(2, "研发部", 1)]
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
company = tree[0]
|
||
assert company["id"] == "dept_1"
|
||
assert company["label"] == "公司"
|
||
assert company["dept_id"] == 1
|
||
assert company["parentid"] == 0
|
||
assert "children" in company
|
||
|
||
def test_employee_node_structure(self):
|
||
"""员工节点包含 id, label, isLeaf, department 字段。"""
|
||
directory = [_emp("u1", "用户1", "研发", [2])]
|
||
dept_list = [_dept(1, "公司", 0), _dept(2, "研发部", 1)]
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
company = tree[0]
|
||
dev = company["children"][0]
|
||
emp = dev["children"][0]
|
||
assert emp["id"] == "u1"
|
||
assert emp["label"] == "用户1"
|
||
assert emp["isLeaf"] is True
|
||
assert emp["department"] == "研发"
|
||
|
||
# ----- 1.10 排序验证 -----
|
||
|
||
def test_departments_sorted_by_name(self):
|
||
"""子部门按名称排序（Python 默认 Unicode 码点序）。"""
|
||
directory = []
|
||
dept_list = [
|
||
_dept(1, "公司", 0),
|
||
_dept(4, "_Z部门", 1),
|
||
_dept(2, "A部门", 1),
|
||
_dept(3, "B部门", 1),
|
||
]
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
company = tree[0]
|
||
dept_names = [n["label"] for n in company["children"] if not n.get("isLeaf")]
|
||
# Python 按 Unicode 码点排序：A(0x41) < B(0x42) < _(0x5F)
|
||
assert dept_names == ["A部门", "B部门", "_Z部门"]
|
||
|
||
def test_employees_sorted_by_name(self):
|
||
"""同一部门下员工按姓名排序（Python 默认 Unicode 码点序）。"""
|
||
directory = [
|
||
_emp("u3", "张三", "研发", [2]),
|
||
_emp("u1", "阿大", "研发", [2]),
|
||
_emp("u2", "李四", "研发", [2]),
|
||
]
|
||
dept_list = [_dept(1, "公司", 0), _dept(2, "研发部", 1)]
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
company = tree[0]
|
||
dev = company["children"][0]
|
||
emp_names = [n["label"] for n in dev["children"] if n.get("isLeaf")]
|
||
# Python 按 Unicode 码点排序：张(U+5F20) < 李(U+674E) < 阿(U+963F)
|
||
assert emp_names == ["张三", "李四", "阿大"]
|
||
|
||
|
||
# =============================================================================
|
||
# 二、_build_flat_tree_by_name() 测试
|
||
# =============================================================================
|
||
|
||
class TestBuildFlatTreeByName:
|
||
"""_build_flat_tree_by_name 降级扁平分组测试。"""
|
||
|
||
def test_basic_grouping(self):
|
||
"""按部门名分组（Python 默认 Unicode 码点序排序）。"""
|
||
directory = [
|
||
_emp("u1", "张三", "研发部"),
|
||
_emp("u2", "李四", "研发部"),
|
||
_emp("u3", "王五", "测试部"),
|
||
]
|
||
tree = _build_flat_tree_by_name(directory)
|
||
|
||
assert len(tree) == 2
|
||
# Python 按 Unicode 码点排序：测(U+6D4B) < 研(U+7814)
|
||
assert tree[0]["label"] == "测试部"
|
||
assert tree[1]["label"] == "研发部"
|
||
assert len(tree[0]["children"]) == 1
|
||
assert len(tree[1]["children"]) == 2
|
||
|
||
def test_multi_dept_takes_first(self):
|
||
"""多部门员工取第一个部门名。"""
|
||
directory = [_emp("u1", "张三", "研发部,测试部")]
|
||
tree = _build_flat_tree_by_name(directory)
|
||
|
||
assert len(tree) == 1
|
||
assert tree[0]["label"] == "研发部"
|
||
|
||
def test_empty_dept_name_goes_to_unassigned(self):
|
||
"""空部门名归到"未分配部门"。"""
|
||
directory = [
|
||
_emp("u1", "张三", ""),
|
||
_emp("u2", "李四", None),
|
||
]
|
||
tree = _build_flat_tree_by_name(directory)
|
||
|
||
assert len(tree) == 1
|
||
assert tree[0]["label"] == "未分配部门"
|
||
assert len(tree[0]["children"]) == 2
|
||
|
||
def test_comma_only_dept_name_goes_to_unassigned(self):
|
||
"""部门名只有逗号时归到"未分配部门"。"""
|
||
directory = [_emp("u1", "张三", ",,")]
|
||
tree = _build_flat_tree_by_name(directory)
|
||
|
||
assert len(tree) == 1
|
||
assert tree[0]["label"] == "未分配部门"
|
||
|
||
def test_empty_directory(self):
|
||
"""空 directory → 空树。"""
|
||
tree = _build_flat_tree_by_name([])
|
||
assert tree == []
|
||
|
||
def test_flat_tree_node_structure(self):
|
||
"""扁平树节点结构正确。"""
|
||
directory = [_emp("u1", "张三", "研发部")]
|
||
tree = _build_flat_tree_by_name(directory)
|
||
|
||
dept_node = tree[0]
|
||
assert dept_node["id"] == "dept_name_研发部"
|
||
assert dept_node["label"] == "研发部"
|
||
assert dept_node["dept_id"] is None
|
||
assert dept_node["parentid"] == 0
|
||
|
||
emp_node = dept_node["children"][0]
|
||
assert emp_node["id"] == "u1"
|
||
assert emp_node["label"] == "张三"
|
||
assert emp_node["isLeaf"] is True
|
||
assert emp_node["department"] == "研发部"
|
||
|
||
|
||
# =============================================================================
|
||
# 三、filter_user_from_tree() 测试
|
||
# =============================================================================
|
||
|
||
class TestFilterUserFromTree:
|
||
"""filter_user_from_tree 递归过滤测试。"""
|
||
|
||
def _sample_tree(self):
|
||
"""构造测试用树：公司 → 研发部(张三,李四) + 测试部(王五)。"""
|
||
return [
|
||
{
|
||
"id": "dept_1",
|
||
"label": "公司",
|
||
"dept_id": 1,
|
||
"parentid": 0,
|
||
"children": [
|
||
{
|
||
"id": "dept_2",
|
||
"label": "研发部",
|
||
"dept_id": 2,
|
||
"parentid": 1,
|
||
"children": [
|
||
{"id": "zhangsan", "label": "张三", "isLeaf": True, "department": "研发部"},
|
||
{"id": "lisi", "label": "李四", "isLeaf": True, "department": "研发部"},
|
||
],
|
||
},
|
||
{
|
||
"id": "dept_3",
|
||
"label": "测试部",
|
||
"dept_id": 3,
|
||
"parentid": 1,
|
||
"children": [
|
||
{"id": "wangwu", "label": "王五", "isLeaf": True, "department": "测试部"},
|
||
],
|
||
},
|
||
],
|
||
}
|
||
]
|
||
|
||
def test_filter_single_user(self):
|
||
"""过滤单个用户，其他用户保留。"""
|
||
tree = self._sample_tree()
|
||
result = filter_user_from_tree(tree, "zhangsan")
|
||
|
||
# 收集所有员工 id
|
||
def get_emp_ids(nodes):
|
||
ids = set()
|
||
for n in nodes:
|
||
if n.get("isLeaf"):
|
||
ids.add(n["id"])
|
||
elif "children" in n:
|
||
ids.update(get_emp_ids(n["children"]))
|
||
return ids
|
||
|
||
emp_ids = get_emp_ids(result)
|
||
assert "zhangsan" not in emp_ids
|
||
assert "lisi" in emp_ids
|
||
assert "wangwu" in emp_ids
|
||
|
||
def test_filter_user_empty_dept_removed(self):
|
||
"""过滤后变空的部门节点被移除。"""
|
||
tree = [
|
||
{
|
||
"id": "dept_1",
|
||
"label": "公司",
|
||
"dept_id": 1,
|
||
"parentid": 0,
|
||
"children": [
|
||
{
|
||
"id": "dept_2",
|
||
"label": "研发部",
|
||
"dept_id": 2,
|
||
"parentid": 1,
|
||
"children": [
|
||
{"id": "zhangsan", "label": "张三", "isLeaf": True, "department": "研发部"},
|
||
],
|
||
},
|
||
{
|
||
"id": "dept_3",
|
||
"label": "测试部",
|
||
"dept_id": 3,
|
||
"parentid": 1,
|
||
"children": [
|
||
{"id": "lisi", "label": "李四", "isLeaf": True, "department": "测试部"},
|
||
{"id": "wangwu", "label": "王五", "isLeaf": True, "department": "测试部"},
|
||
],
|
||
},
|
||
],
|
||
}
|
||
]
|
||
|
||
# 过滤张三 → 研发部变空 → 研发部被移除
|
||
result = filter_user_from_tree(tree, "zhangsan")
|
||
|
||
assert len(result) == 1
|
||
company = result[0]
|
||
dept_ids = [n["id"] for n in company["children"] if not n.get("isLeaf")]
|
||
assert "dept_2" not in dept_ids # 研发部被移除
|
||
assert "dept_3" in dept_ids # 测试部保留
|
||
|
||
def test_filter_user_not_in_tree(self):
|
||
"""过滤不存在的用户，树不变。"""
|
||
tree = self._sample_tree()
|
||
result = filter_user_from_tree(tree, "nonexistent")
|
||
|
||
assert len(result) == len(tree)
|
||
# 树结构应与原始相同（所有员工都在）
|
||
def count_emps(nodes):
|
||
c = 0
|
||
for n in nodes:
|
||
if n.get("isLeaf"):
|
||
c += 1
|
||
elif "children" in n:
|
||
c += count_emps(n["children"])
|
||
return c
|
||
|
||
assert count_emps(result) == 3
|
||
|
||
def test_filter_empty_tree(self):
|
||
"""空树过滤返回空列表。"""
|
||
result = filter_user_from_tree([], "anyone")
|
||
assert result == []
|
||
|
||
def test_filter_user_in_multiple_depts(self):
|
||
"""用户出现在多个部门时，所有实例都被移除。"""
|
||
tree = [
|
||
{
|
||
"id": "dept_1",
|
||
"label": "公司",
|
||
"dept_id": 1,
|
||
"parentid": 0,
|
||
"children": [
|
||
{
|
||
"id": "dept_2",
|
||
"label": "研发部",
|
||
"dept_id": 2,
|
||
"parentid": 1,
|
||
"children": [
|
||
{"id": "zhangsan", "label": "张三", "isLeaf": True, "department": "研发"},
|
||
{"id": "lisi", "label": "李四", "isLeaf": True, "department": "研发"},
|
||
],
|
||
},
|
||
{
|
||
"id": "dept_3",
|
||
"label": "测试部",
|
||
"dept_id": 3,
|
||
"parentid": 1,
|
||
"children": [
|
||
{"id": "zhangsan", "label": "张三", "isLeaf": True, "department": "测试"},
|
||
],
|
||
},
|
||
],
|
||
}
|
||
]
|
||
|
||
result = filter_user_from_tree(tree, "zhangsan")
|
||
|
||
def get_emp_ids(nodes):
|
||
ids = set()
|
||
for n in nodes:
|
||
if n.get("isLeaf"):
|
||
ids.add(n["id"])
|
||
elif "children" in n:
|
||
ids.update(get_emp_ids(n["children"]))
|
||
return ids
|
||
|
||
emp_ids = get_emp_ids(result)
|
||
assert "zhangsan" not in emp_ids
|
||
assert "lisi" in emp_ids # 李四保留
|
||
|
||
def test_filter_all_users_empties_tree(self):
|
||
"""过滤所有用户后，整个树变空。"""
|
||
tree = self._sample_tree()
|
||
# 依次过滤所有 3 个用户
|
||
result = tree
|
||
for uid in ["zhangsan", "lisi", "wangwu"]:
|
||
result = filter_user_from_tree(result, uid)
|
||
|
||
assert result == []
|
||
|
||
def test_filter_preserves_dept_node_fields(self):
|
||
"""过滤后部门节点保留原有字段（dept_id, parentid 等）。"""
|
||
tree = self._sample_tree()
|
||
result = filter_user_from_tree(tree, "zhangsan")
|
||
|
||
company = result[0]
|
||
assert company["dept_id"] == 1
|
||
assert company["parentid"] == 0
|
||
assert company["label"] == "公司"
|
||
|
||
def test_filter_does_not_mutate_original(self):
|
||
"""过滤不修改原始树（返回新对象）。"""
|
||
tree = self._sample_tree()
|
||
original_count = count_tree_employees(tree)
|
||
filter_user_from_tree(tree, "zhangsan")
|
||
# 原始树不受影响
|
||
assert count_tree_employees(tree) == original_count
|
||
|
||
|
||
# =============================================================================
|
||
# 四、count_tree_employees() 测试
|
||
# =============================================================================
|
||
|
||
class TestCountTreeEmployees:
|
||
"""count_tree_employees 递归统计测试。"""
|
||
|
||
def test_count_flat_tree(self):
|
||
"""扁平树统计员工数。"""
|
||
tree = [
|
||
{"id": "u1", "label": "张三", "isLeaf": True, "department": ""},
|
||
{"id": "u2", "label": "李四", "isLeaf": True, "department": ""},
|
||
]
|
||
assert count_tree_employees(tree) == 2
|
||
|
||
def test_count_nested_tree(self):
|
||
"""嵌套树统计员工数。"""
|
||
tree = [
|
||
{
|
||
"id": "dept_1",
|
||
"label": "公司",
|
||
"children": [
|
||
{
|
||
"id": "dept_2",
|
||
"label": "研发部",
|
||
"children": [
|
||
{"id": "u1", "label": "张三", "isLeaf": True},
|
||
{"id": "u2", "label": "李四", "isLeaf": True},
|
||
],
|
||
},
|
||
{"id": "u3", "label": "王五", "isLeaf": True},
|
||
],
|
||
}
|
||
]
|
||
assert count_tree_employees(tree) == 3
|
||
|
||
def test_count_empty_tree(self):
|
||
"""空树员工数为 0。"""
|
||
assert count_tree_employees([]) == 0
|
||
|
||
def test_count_only_dept_nodes(self):
|
||
"""只有部门节点无员工时为 0。"""
|
||
tree = [
|
||
{
|
||
"id": "dept_1",
|
||
"label": "公司",
|
||
"children": [
|
||
{"id": "dept_2", "label": "研发部", "children": []},
|
||
],
|
||
}
|
||
]
|
||
assert count_tree_employees(tree) == 0
|
||
|
||
def test_count_mixed_nodes(self):
|
||
"""混合节点（部门+员工）正确统计。"""
|
||
tree = [
|
||
{
|
||
"id": "dept_1",
|
||
"label": "公司",
|
||
"children": [
|
||
{"id": "u1", "label": "张三", "isLeaf": True},
|
||
{
|
||
"id": "dept_2",
|
||
"label": "研发部",
|
||
"children": [
|
||
{"id": "u2", "label": "李四", "isLeaf": True},
|
||
],
|
||
},
|
||
],
|
||
},
|
||
{"id": "u3", "label": "王五", "isLeaf": True},
|
||
]
|
||
assert count_tree_employees(tree) == 3
|
||
|
||
|
||
# =============================================================================
|
||
# 五、get_org_tree_cached() 测试（mock 依赖）
|
||
# =============================================================================
|
||
|
||
class TestGetOrgTreeCached:
|
||
"""get_org_tree_cached 缓存与过滤测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_cache_key_per_endpoint(self):
|
||
"""agent 和 h5 端点使用不同的缓存 key。"""
|
||
from tests.conftest import MockRedis
|
||
|
||
mock_redis = MockRedis()
|
||
directory = [_emp("u1", "张三", "研发", [2])]
|
||
dept_list = [_dept(1, "公司", 0), _dept(2, "研发部", 1)]
|
||
|
||
with patch(
|
||
"app.services.employee_directory.get_org_directory",
|
||
new_callable=AsyncMock,
|
||
return_value=(directory, True),
|
||
):
|
||
with patch(
|
||
"app.services.employee_directory.get_cached_dept_list",
|
||
new_callable=AsyncMock,
|
||
return_value=dept_list,
|
||
):
|
||
# 调用 agent 端点
|
||
await get_org_tree_cached(None, mock_redis, "agent", "nobody")
|
||
# 调用 h5 端点
|
||
await get_org_tree_cached(None, mock_redis, "h5", "nobody")
|
||
|
||
# 验证两个不同的缓存 key
|
||
assert f"{ORG_TREE_CACHE_KEY_PREFIX}:agent" in mock_redis._data
|
||
assert f"{ORG_TREE_CACHE_KEY_PREFIX}:h5" in mock_redis._data
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_user_exclusion_from_cached_tree(self):
|
||
"""缓存包含全量员工，读取后过滤当前用户。"""
|
||
from tests.conftest import MockRedis
|
||
|
||
mock_redis = MockRedis()
|
||
directory = [
|
||
_emp("u1", "张三", "研发", [2]),
|
||
_emp("u2", "李四", "研发", [2]),
|
||
_emp("current_user", "当前用户", "研发", [2]),
|
||
]
|
||
dept_list = [_dept(1, "公司", 0), _dept(2, "研发部", 1)]
|
||
|
||
with patch(
|
||
"app.services.employee_directory.get_org_directory",
|
||
new_callable=AsyncMock,
|
||
return_value=(directory, True),
|
||
):
|
||
with patch(
|
||
"app.services.employee_directory.get_cached_dept_list",
|
||
new_callable=AsyncMock,
|
||
return_value=dept_list,
|
||
):
|
||
# 第一次调用：构建树并缓存
|
||
tree1 = await get_org_tree_cached(None, mock_redis, "agent", "current_user")
|
||
# 当前用户应被排除
|
||
assert count_tree_employees(tree1) == 2 # 张三 + 李四
|
||
|
||
def get_emp_ids(nodes):
|
||
ids = set()
|
||
for n in nodes:
|
||
if n.get("isLeaf"):
|
||
ids.add(n["id"])
|
||
elif "children" in n:
|
||
ids.update(get_emp_ids(n["children"]))
|
||
return ids
|
||
|
||
assert "current_user" not in get_emp_ids(tree1)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_cache_hit_skips_rebuild(self):
|
||
"""缓存命中时不重新构建树。"""
|
||
from tests.conftest import MockRedis
|
||
|
||
mock_redis = MockRedis()
|
||
# 预填充缓存
|
||
cached_tree = [
|
||
{
|
||
"id": "dept_1",
|
||
"label": "公司",
|
||
"dept_id": 1,
|
||
"parentid": 0,
|
||
"children": [
|
||
{"id": "u1", "label": "张三", "isLeaf": True, "department": ""},
|
||
{"id": "u2", "label": "李四", "isLeaf": True, "department": ""},
|
||
],
|
||
}
|
||
]
|
||
await mock_redis.setex(
|
||
f"{ORG_TREE_CACHE_KEY_PREFIX}:agent",
|
||
1800,
|
||
json.dumps(cached_tree, ensure_ascii=False),
|
||
)
|
||
|
||
# mock get_org_directory — 如果被调用说明缓存未命中
|
||
mock_get_dir = AsyncMock(return_value=([], True))
|
||
mock_get_dept = AsyncMock(return_value=[])
|
||
|
||
with patch("app.services.employee_directory.get_org_directory", mock_get_dir):
|
||
with patch("app.services.employee_directory.get_cached_dept_list", mock_get_dept):
|
||
tree = await get_org_tree_cached(None, mock_redis, "agent", "u1")
|
||
|
||
# 缓存命中，不应调用 get_org_directory
|
||
mock_get_dir.assert_not_called()
|
||
mock_get_dept.assert_not_called()
|
||
|
||
# u1 被过滤
|
||
assert count_tree_employees(tree) == 1
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_cache_miss_builds_and_caches(self):
|
||
"""缓存未命中时构建树并写入缓存。"""
|
||
from tests.conftest import MockRedis
|
||
|
||
mock_redis = MockRedis()
|
||
directory = [_emp("u1", "张三", "研发", [2])]
|
||
dept_list = [_dept(1, "公司", 0), _dept(2, "研发部", 1)]
|
||
|
||
with patch(
|
||
"app.services.employee_directory.get_org_directory",
|
||
new_callable=AsyncMock,
|
||
return_value=(directory, True),
|
||
):
|
||
with patch(
|
||
"app.services.employee_directory.get_cached_dept_list",
|
||
new_callable=AsyncMock,
|
||
return_value=dept_list,
|
||
):
|
||
tree = await get_org_tree_cached(None, mock_redis, "agent", "nobody")
|
||
|
||
# 树已构建
|
||
assert len(tree) == 1
|
||
# 缓存已写入
|
||
assert f"{ORG_TREE_CACHE_KEY_PREFIX}:agent" in mock_redis._data
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_no_redis_builds_tree_without_cache(self):
|
||
"""redis=None 时仍能构建树（不读写缓存）。"""
|
||
directory = [_emp("u1", "张三", "研发", [2])]
|
||
dept_list = [_dept(1, "公司", 0), _dept(2, "研发部", 1)]
|
||
|
||
with patch(
|
||
"app.services.employee_directory.get_org_directory",
|
||
new_callable=AsyncMock,
|
||
return_value=(directory, True),
|
||
):
|
||
with patch(
|
||
"app.services.employee_directory.get_cached_dept_list",
|
||
new_callable=AsyncMock,
|
||
return_value=dept_list,
|
||
):
|
||
tree = await get_org_tree_cached(None, None, "agent", "nobody")
|
||
|
||
assert len(tree) == 1
|
||
assert count_tree_employees(tree) == 1
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_degraded_flat_tree_when_no_dept_list(self):
|
||
"""dept_list 为空时降级到扁平树。"""
|
||
from tests.conftest import MockRedis
|
||
|
||
mock_redis = MockRedis()
|
||
directory = [_emp("u1", "张三", "研发部")]
|
||
|
||
with patch(
|
||
"app.services.employee_directory.get_org_directory",
|
||
new_callable=AsyncMock,
|
||
return_value=(directory, True),
|
||
):
|
||
with patch(
|
||
"app.services.employee_directory.get_cached_dept_list",
|
||
new_callable=AsyncMock,
|
||
return_value=[], # 空部门列表
|
||
):
|
||
tree = await get_org_tree_cached(None, mock_redis, "agent", "nobody")
|
||
|
||
# 降级路径：dept_name_ 前缀
|
||
assert len(tree) == 1
|
||
assert tree[0]["id"] == "dept_name_研发部"
|
||
|
||
|
||
# =============================================================================
|
||
# 六、兼容性测试：resolve_target 与搜索接口不受 dept_ids 影响
|
||
# =============================================================================
|
||
|
||
class TestCompatibility:
|
||
"""验证新增 dept_ids 字段不影响现有功能。"""
|
||
|
||
def test_directory_entry_has_dept_ids_field(self):
|
||
"""验证 directory 条目结构包含 dept_ids 字段（企微路径）。
|
||
|
||
这不是直接测试 get_org_directory（需要 mock 企微 API），
|
||
而是验证 build_org_tree 对 dept_ids 字段的消费方式正确。
|
||
"""
|
||
# 模拟 get_org_directory 返回的数据结构
|
||
directory = [
|
||
{
|
||
"employee_id": "u1",
|
||
"name": "张三",
|
||
"department": "研发部",
|
||
"dept_ids": [2],
|
||
}
|
||
]
|
||
dept_list = [_dept(1, "公司", 0), _dept(2, "研发部", 1)]
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
# 员工正确出现在部门下
|
||
assert count_tree_employees(tree) == 1
|
||
|
||
def test_directory_entry_without_dept_ids(self):
|
||
"""dept_ids 字段缺失时（旧缓存兼容），build_org_tree 不报错。"""
|
||
directory = [
|
||
{
|
||
"employee_id": "u1",
|
||
"name": "张三",
|
||
"department": "研发部",
|
||
# dept_ids 缺失
|
||
}
|
||
]
|
||
dept_list = [_dept(1, "公司", 0), _dept(2, "研发部", 1)]
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
# dept_ids 缺失 → emp.get("dept_ids") or [] → [] → 归入未分配部门
|
||
top_ids = {n["id"] for n in tree}
|
||
assert "dept_unassigned" in top_ids
|
||
|
||
def test_employee_with_unknown_dept_id(self):
|
||
"""员工 dept_id 不在 dept_list 中时的行为验证。
|
||
|
||
已知边界情况：员工 dept_id 不在 dept_map 中时，
|
||
员工会被分配到 dept_employees[unknown_id] 但不会被任何 build_dept_node 处理，
|
||
导致该员工从树中"消失"。
|
||
|
||
本测试验证此行为并记录为已知边界情况。
|
||
"""
|
||
directory = [_emp("u1", "张三", "某部门", [99])] # dept 99 不在 dept_list
|
||
dept_list = [_dept(1, "公司", 0), _dept(2, "研发部", 1)]
|
||
|
||
tree = build_org_tree(directory, dept_list)
|
||
|
||
# 已知边界情况：员工 u1 不在树中（dept 99 不在 dept_map）
|
||
# 理想行为应该是归入"未分配部门"，但当前实现会丢失该员工
|
||
assert count_tree_employees(tree) == 0
|
||
# 记录此行为 — 见测试报告中的 Known Issues
|