test: add DataSourceParser & Exporter unittest

src/bk-user/bkuser/apps/sync/syncers.py

@@ -153,9 +153,9 @@ def _generate_tree_id(data_source_id: int, root_node_idx: int) -> int:
         在 MPTT 中，单个 tree_id 只能用于一棵树，因此需要为不同的树分配不同的 ID
 
         # FIXME (su) 抽象成 TreeIdProvider，利用 Redis 锁，提供在并发情况下，安全获取最大 tree_id + 1 的能力
-        分配规则：data_source_id * 10000 + root_node_idx
+        分配规则：data_source_id * 1000 + root_node_idx
         """
-        return data_source_id * 10**5 + root_node_idx
+        return data_source_id * 10**4 + root_node_idx
 
 
 class DataSourceUserSyncer:

src/bk-user/bkuser/biz/data_source_organization.py

@@ -22,6 +22,7 @@
     DataSourceUserLeaderRelation,
 )
 from bkuser.apps.tenant.models import Tenant, TenantUser
+from bkuser.plugins.local.utils import gen_code
 from bkuser.utils.uuid import generate_uuid
 
 
@@ -77,7 +78,9 @@ def create_user(
         # TODO：补充日志
         with transaction.atomic():
             # 创建数据源用户
-            user = DataSourceUser.objects.create(data_source=data_source, **base_user_info.model_dump())
+            user = DataSourceUser.objects.create(
+                data_source=data_source, code=gen_code(base_user_info.username), **base_user_info.model_dump()
+            )
 
             # 批量创建数据源用户-部门关系
             department_user_relation_objs = [

src/bk-user/bkuser/biz/exporters.py

@@ -155,7 +155,7 @@ def _build_user_departments_map(self) -> Dict[int, List[int]]:
             .values("user_id", "department_id")
         )
         return {
-            user_id: [r["department_id"] for r in group]
+            user_id: sorted([r["department_id"] for r in group])
             for user_id, group in groupby(relations, key=lambda r: r["user_id"])
         }
 
@@ -171,7 +171,7 @@ def _build_user_leaders_map(self) -> Dict[int, List[int]]:
             .values("user_id", "leader_id")
         )
         return {
-            user_id: [r["leader_id"] for r in group]
+            user_id: sorted([r["leader_id"] for r in group])
             for user_id, group in groupby(relations, key=lambda r: r["user_id"])
         }
 

src/bk-user/bkuser/biz/tenant.py

@@ -25,6 +25,7 @@
     DataSourceUserHandler,
 )
 from bkuser.plugins.local.models import PasswordInitialConfig
+from bkuser.plugins.local.utils import gen_code
 from bkuser.utils.uuid import generate_uuid
 
 
@@ -273,7 +274,9 @@ def create_with_managers(
             tenant_manager_objs = []
             for i in managers:
                 # 创建数据源用户
-                data_source_user = DataSourceUser.objects.create(data_source=data_source, **i.model_dump())
+                data_source_user = DataSourceUser.objects.create(
+                    data_source=data_source, code=gen_code(i.username), **i.model_dump()
+                )
                 # 创建对应的租户用户
                 tenant_user = TenantUser.objects.create(
                     data_source_user=data_source_user,

src/bk-user/bkuser/plugins/local/exceptions.py

@@ -31,6 +31,10 @@ class DuplicateColumnName(LocalDataSourcePluginError):
     """待导入文件中存在重复列名"""
 
 
+class RequiredFieldIsEmpty(LocalDataSourcePluginError):
+    """待导入文件中必填字段为空"""
+
+
 class DuplicateUsername(LocalDataSourcePluginError):
     """待导入文件中存在重复用户"""
 

src/bk-user/bkuser/plugins/local/parser.py

@@ -9,7 +9,6 @@
 specific language governing permissions and limitations under the License.
 """
 from collections import Counter
-from hashlib import sha256
 from typing import List
 
 import phonenumbers
@@ -21,10 +20,12 @@
     CustomColumnNameInvalid,
     DuplicateColumnName,
     DuplicateUsername,
+    RequiredFieldIsEmpty,
     SheetColumnsNotMatch,
     UserLeaderInvalid,
     UserSheetNotExists,
 )
+from bkuser.plugins.local.utils import gen_code
 from bkuser.plugins.models import RawDataSourceDepartment, RawDataSourceUser
 
 
@@ -125,10 +126,11 @@ def _validate_and_prepare(self):  # noqa: C901
             info = dict(zip(self.all_field_names, [cell.value for cell in row], strict=True))
             for field_name in self.required_field_names:
                 if not info.get(field_name):
-                    raise ValueError(_("待导入文件中必填字段 {} 存在空值").format(field_name))
+                    raise RequiredFieldIsEmpty(_("待导入文件中必填字段 {} 存在空值").format(field_name))
 
             usernames.append(info["username"])
-            leaders.extend([ld.strip() for ld in info["leaders"].split(",") if ld])
+            if user_leaders := info["leaders"]:
+                leaders.extend([ld.strip() for ld in user_leaders.split(",") if ld])
 
         # 6. 检查用户名是否有重复的
         if duplicate_usernames := [n for n, cnt in Counter(usernames).items() if cnt > 1]:
@@ -146,7 +148,7 @@ def _parse_departments(self):
                     organizations.add(org.strip())
 
         # 组织路径：本数据源部门 Code 映射表
-        org_code_map = {org: self.gen_code(org) for org in organizations}
+        org_code_map = {org: gen_code(org) for org in organizations}
         for org in organizations:
             parent_org, __, dept_name = org.rpartition("/")
             self.departments.append(
@@ -159,10 +161,10 @@ def _parse_users(self):
 
             department_codes, leader_codes = [], []
             if organizations := properties.pop("organizations"):
-                department_codes = [self.gen_code(org.strip()) for org in organizations.split(",") if org]
+                department_codes = [gen_code(org.strip()) for org in organizations.split(",") if org]
 
             if leaders := properties.pop("leaders"):
-                leader_codes = [self.gen_code(ld.strip()) for ld in leaders.split(",") if ld]
+                leader_codes = [gen_code(ld.strip()) for ld in leaders.split(",") if ld]
 
             phone_number = str(properties.pop("phone_number"))
             # 默认认为是不带国际代码的
@@ -177,15 +179,9 @@ def _parse_users(self):
             properties = {k: str(v) for k, v in properties.items() if v is not None}
             self.users.append(
                 RawDataSourceUser(
-                    code=self.gen_code(properties["username"]),
+                    code=gen_code(properties["username"]),
                     properties=properties,
                     leaders=leader_codes,
                     departments=department_codes,
                 )
             )
-
-    @staticmethod
-    def gen_code(username_or_org: str) -> str:
-        # 本地数据源组织没有提供用户及部门 code 的方式，
-        # 因此使用 sha256 计算以避免冲突，也便于后续插入 DB 时进行比较
-        return sha256(username_or_org.encode("utf-8")).hexdigest()

src/bk-user/bkuser/plugins/local/utils.py

@@ -0,0 +1,17 @@
+# -*- coding: utf-8 -*-
+"""
+TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-用户管理(Bk-User) available.
+Copyright (C) 2017-2021 THL A29 Limited, a Tencent company. All rights reserved.
+Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at http://opensource.org/licenses/MIT
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+specific language governing permissions and limitations under the License.
+"""
+from hashlib import sha256
+
+
+def gen_code(username_or_org: str) -> str:
+    # 本地数据源数据没有提供用户及部门 code 的方式，
+    # 因此使用 sha256 计算以避免冲突，也便于后续插入 DB 时进行比较
+    return sha256(username_or_org.encode("utf-8")).hexdigest()

src/bk-user/tests/biz/test_exporters.py

@@ -10,18 +10,80 @@
 """
 import pytest
 from bkuser.apps.data_source.models import DataSourceUser
+from bkuser.biz.exporters import DataSourceUserExporter
 
 pytestmark = pytest.mark.django_db
 
 
 class TestDataSourceExporter:
+    """测试用户数据导出 & 模板获取"""
+
     def test_get_template(self, bare_local_data_source, tenant_user_custom_fields):
-        # TODO (su) 获取模板，确认模板列名，自定义字段列
-        ...
+        exporter = DataSourceUserExporter(bare_local_data_source)
+        tmpl = exporter.get_template()
+
+        assert "users" in tmpl.sheetnames
+        assert [cell.value for cell in tmpl["users"][exporter.col_name_row_idx]] == [
+            "用户名/username",
+            "姓名/full_name",
+            "邮箱/email",
+            "手机号/phone_number",
+            "组织/organizations",
+            "直接上级/leaders",
+            "年龄/age",
+            "性别/gender",
+            "籍贯/region",
+        ]
 
     def test_export(self, full_local_data_source, tenant_user_custom_fields):
         # 初始化数据中，是没有 extras 的值的，这里更新下，以便于验证导出器的功能
-        DataSourceUser.objects.filter(data_source=full_local_data_source).update(
-            extras={"age": "20", "gender": "male", "region": "guangdong"}
-        )
-        # TODO (su) 导出数据，确认数据准确性，特别是自定义字段
+        exists_users = DataSourceUser.objects.filter(data_source=full_local_data_source)
+        for idx, user in enumerate(exists_users):
+            user.extras = {"age": str(20 + idx), "gender": "male", "region": "region-" + str(idx)}
+            user.save()
+
+        # 导出数据，确认数据准确性，特别是自定义字段
+        wk = DataSourceUserExporter(full_local_data_source).export()
+        assert "users" in wk.sheetnames
+
+        # 表格中第三行开始才是数据
+        min_data_row_index = 3
+        for idx, row in enumerate(wk["users"].iter_rows(min_row=min_data_row_index)):
+            assert row[0].value == exists_users[idx].username
+            assert row[1].value == exists_users[idx].full_name
+            assert row[2].value == exists_users[idx].email
+            assert row[3].value == f"+{exists_users[idx].phone_country_code}{exists_users[idx].phone}"
+            # 第四第五列分别是组织，直接上级，不在这个循环做检查
+            assert row[6].value == str(20 + idx)
+            assert row[7].value == "male"
+            assert row[8].value == "region-" + str(idx)
+
+        # 检查组织信息
+        assert [cell.value for cell in wk["users"]["E"][2:]] == [
+            "公司",
+            "公司/部门A, 公司/部门A/中心AA",
+            "公司/部门A, 公司/部门B",
+            "公司/部门A/中心AA",
+            "公司/部门A/中心AA/小组AAA",
+            "公司/部门A/中心AB",
+            "公司/部门A/中心AB",
+            "公司/部门B/中心BA, 公司/部门A/中心AB/小组ABA",
+            "公司/部门A/中心AB/小组ABA",
+            "公司/部门B/中心BA/小组BAA",
+            "",
+        ]
+
+        # 检查 leader 信息
+        assert [cell.value for cell in wk["users"]["F"][2:]] == [
+            "",
+            "zhangsan",
+            "zhangsan",
+            "lisi",
+            "zhaoliu",
+            "lisi, wangwu",
+            "wangwu",
+            "wangwu, maiba",
+            "lushi",
+            "lushi",
+            "",
+        ]