diff --git a/src/bk-user/bkuser/biz/tenant.py b/src/bk-user/bkuser/biz/tenant.py index aa91c5e9a..1de7d9f05 100644 --- a/src/bk-user/bkuser/biz/tenant.py +++ b/src/bk-user/bkuser/biz/tenant.py @@ -375,13 +375,12 @@ def convert_data_source_department_to_tenant_department( continue # 部门基础信息 data_source_department_info = data_source_departments[data_source_department_id] - # 只要一个子部门被授权，都是存在子部门 - children_flag = [True for child in data_source_department_info.child_ids if child in bound_departments_ids] + # 协同数据源：只要一个子部门被授权给该租户，has_children=True data.append( TenantDepartmentBaseInfo( id=tenant_department.id, name=data_source_department_info.name, - has_children=any(children_flag), + has_children=bool(set(data_source_department_info.child_ids) & set(bound_departments_ids)), ) ) return data diff --git a/src/bk-user/tests/apis/web/conftest.py b/src/bk-user/tests/apis/web/conftest.py new file mode 100644 index 000000000..e2f32ab4f --- /dev/null +++ b/src/bk-user/tests/apis/web/conftest.py @@ -0,0 +1,69 @@ +# -*- 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 typing import List + +import pytest +from bkuser.apps.data_source.constants import DataSourcePluginEnum +from bkuser.apps.data_source.models import ( + DataSource, + DataSourceDepartment, + DataSourceUser, +) +from bkuser.apps.tenant.models import TenantDepartment, TenantUser + +from tests.test_utils.data_source import ( + create_data_source_departments_with_relations, + create_data_source_users_with_relations, +) +from tests.test_utils.helpers import generate_random_string +from tests.test_utils.tenant import create_tenant_departments, create_tenant_users + + +@pytest.fixture() +def local_data_source(default_tenant) -> DataSource: + """ + 创建测试数据源 + """ + return DataSource.objects.create( + name=generate_random_string(), owner_tenant_id=default_tenant, plugin_id=DataSourcePluginEnum.LOCAL + ) + + +@pytest.fixture() +def local_data_source_departments(local_data_source) -> List[DataSourceDepartment]: + """ + 创建测试数据源部门，并以首个对象作为父部门 + """ + return create_data_source_departments_with_relations(local_data_source) + + +@pytest.fixture() +def local_data_source_users(local_data_source, local_data_source_departments) -> List[DataSourceUser]: + """ + 创建测试数据源用户， 并以首个对象作为leader, 随机关联部门 + """ + return create_data_source_users_with_relations(local_data_source, local_data_source_departments) + + +@pytest.fixture() +def tenant_users(default_tenant, local_data_source_users) -> List[TenantUser]: + """ + 根据测试数据源用户，创建租户用户 + """ + return create_tenant_users(default_tenant, local_data_source_users) + + +@pytest.fixture() +def tenant_departments(default_tenant, local_data_source, local_data_source_departments) -> List[TenantDepartment]: + """ + 根据测试数据源部门，创建租户部门 + """ + return create_tenant_departments(default_tenant, local_data_source_departments) diff --git a/src/bk-user/tests/apis/web/organization/test_organization.py b/src/bk-user/tests/apis/web/organization/test_organization.py index e5d6382b5..fd7ce8b83 100644 --- a/src/bk-user/tests/apis/web/organization/test_organization.py +++ b/src/bk-user/tests/apis/web/organization/test_organization.py @@ -8,7 +8,6 @@ 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 typing import List import pytest from bkuser.apps.data_source.models import ( @@ -17,56 +16,58 @@ DataSourceUserLeaderRelation, ) from bkuser.apps.tenant.models import Tenant, TenantDepartment, TenantManager, TenantUser -from bkuser.auth.models import User from django.urls import reverse +from django.utils import timezone from rest_framework import status from rest_framework.test import APIClient +from tests.test_utils.helpers import generate_random_string + pytestmark = pytest.mark.django_db class TestTenantListApi: def test_list_tenants( self, - api_client: APIClient, - bk_user: User, - default_tenant: str, - random_tenant: str, - default_tenant_departments: List[TenantDepartment], + api_client, + bk_user, + default_tenant, + tenant_departments, ): resp = api_client.get(reverse("organization.tenant.list")) # 至少会有一个当前用户所属的租户 + current_tenant_id = bk_user.get_property("tenant_id") assert len(resp.data) >= 1 - resp_data = resp.data - for tenant_info in resp_data: - if tenant_info["id"] != bk_user.get_property("tenant_id"): + assert current_tenant_id == resp.data[0]["id"] + + for tenant_info in resp.data: + if tenant_info["id"] != current_tenant_id: assert not tenant_info["departments"] else: assert len(tenant_info["departments"]) >= 1 class TestTenantRetrieveUpdateApi: - NOT_EXIST_TENANT_ID = 9999 - - def test_retrieve_tenant(self, api_client: APIClient, bk_user: User): + def test_retrieve_tenant(self, api_client, bk_user, default_tenant): tenant_id = bk_user.get_property("tenant_id") - resp = api_client.get(reverse("organization.tenant.retrieve_update", kwargs={"id": tenant_id})) - resp_data = resp.data + resp = api_client.get( + reverse("organization.tenant.retrieve_update", kwargs={"id": bk_user.get_property("tenant_id")}) + ) + tenant = Tenant.objects.get(id=tenant_id) - assert tenant_id == resp_data["id"] - assert tenant.id == resp_data["id"] - assert tenant.name == resp_data["name"] - assert tenant.updated_at_display == resp_data["updated_at"] - assert tenant.logo == resp_data["logo"] - assert tenant.feature_flags == resp_data["feature_flags"] - assert TenantManager.objects.filter(tenant=tenant).count() == len(resp_data["managers"]) + assert tenant.id == resp.data["id"] + assert tenant.name == resp.data["name"] + assert tenant.updated_at_display == resp.data["updated_at"] + assert tenant.logo == resp.data["logo"] + assert tenant.feature_flags == resp.data["feature_flags"] + assert TenantManager.objects.filter(tenant=tenant).count() == len(resp.data["managers"]) - for item in resp_data["managers"]: + for item in resp.data["managers"]: assert TenantManager.objects.filter(tenant=tenant, tenant_user_id=item["id"]).exists() - def test_retrieve_other_tenant(self, api_client: APIClient, random_tenant: str): + def test_retrieve_other_tenant(self, api_client, random_tenant): resp = api_client.get(reverse("organization.tenant.retrieve_update", kwargs={"id": random_tenant})) resp_data = resp.data tenant = Tenant.objects.get(id=random_tenant) @@ -79,12 +80,12 @@ def test_retrieve_other_tenant(self, api_client: APIClient, random_tenant: str): # 非当前用户所在租户，不返回管理员 assert not resp_data["managers"] - def test_retrieve_not_exist_tenant(self, api_client: APIClient): - resp = api_client.get(reverse("organization.tenant.retrieve_update", kwargs={"id": self.NOT_EXIST_TENANT_ID})) + def test_retrieve_not_exist_tenant(self, api_client): + resp = api_client.get(reverse("organization.tenant.retrieve_update", kwargs={"id": generate_random_string()})) assert resp.status_code == status.HTTP_404_NOT_FOUND - def test_update_tenant(self, api_client: APIClient, default_tenant: str, default_tenant_users: List[TenantUser]): - new_manager_ids = [user.id for user in default_tenant_users] + def test_update_tenant(self, api_client, default_tenant, tenant_users): + new_manager_ids = [user.id for user in tenant_users] update_data = { "id": "fake-tenant-updated", "name": "fake-tenant-updated", @@ -105,9 +106,7 @@ def test_update_tenant(self, api_client: APIClient, default_tenant: str, default TenantManager.objects.filter(tenant=tenant).values_list("tenant_user_id", flat=True) ) - def test_update_other_tenant( - self, api_client: APIClient, random_tenant: str, default_tenant_users: List[TenantUser] - ): + def test_update_other_tenant(self, api_client, random_tenant, tenant_users): resp = api_client.put( reverse("organization.tenant.retrieve_update", kwargs={"id": random_tenant}), data={ @@ -115,7 +114,7 @@ def test_update_other_tenant( "name": "fake-tenant-updated", "feature_flags": {"user_number_visible": False}, "logo": "aabb", - "manager_ids": [user.id for user in default_tenant_users], + "manager_ids": [user.id for user in tenant_users], }, ) # 进行更新非当前用户的租户，异常返回 @@ -123,15 +122,13 @@ def test_update_other_tenant( class TestTenantUserListApi: - def test_list_tenant_users( - self, api_client: APIClient, default_tenant: str, default_tenant_users: List[TenantUser] - ): + def test_list_tenant_users(self, api_client, default_tenant, tenant_users): resp = api_client.get(reverse("organization.tenant.users.list", kwargs={"id": default_tenant})) assert TenantUser.objects.filter(tenant_id=default_tenant).count() == resp.data["count"] for item in resp.data["results"]: tenant_user = TenantUser.objects.filter(id=item["id"]).first() - assert tenant_user + assert tenant_user is not None assert tenant_user.data_source_user.username == item["username"] assert tenant_user.data_source_user.full_name == item["full_name"] assert tenant_user.data_source_user.email == item["email"] @@ -141,8 +138,8 @@ def test_list_tenant_users( class TestTenantDepartmentChildrenListApi: - def test_retrieve_children(self, api_client: APIClient, default_tenant_departments: List[TenantDepartment]): - for item in default_tenant_departments: + def test_retrieve_children(self, api_client, tenant_departments): + for item in tenant_departments: resp = api_client.get(reverse("organization.children.list", kwargs={"id": item.id})) children = DataSourceDepartmentRelation.objects.get(department=item.data_source_department).get_children() @@ -161,10 +158,10 @@ class TestTenantDepartmentUserListApi: def test_list_department_users( self, api_client: APIClient, - default_tenant_departments: List[TenantDepartment], - default_tenant_users: List[TenantUser], + tenant_departments, + tenant_users, ): - for tenant_department in default_tenant_departments: + for tenant_department in tenant_departments: resp = api_client.get(reverse("departments.users.list", kwargs={"id": tenant_department.id})) resp_data = resp.data @@ -182,15 +179,16 @@ def test_list_department_users( class TestTenantUserRetrieveApi: - def test_retrieve_user(self, api_client: APIClient, default_tenant_users: List[TenantUser]): - for tenant_user in default_tenant_users: + def test_retrieve_user(self, api_client, tenant_users): + for tenant_user in tenant_users: resp = api_client.get(reverse("department.users.retrieve", kwargs={"id": tenant_user.id})) resp_data = resp.data data_source_user = tenant_user.data_source_user assert tenant_user.id == resp_data["id"] - assert tenant_user.account_expired_at.strftime("%Y-%m-%d %H:%M:%S") == resp_data["account_expired_at"] + real_account_expired_at = timezone.localtime(tenant_user.account_expired_at) + assert real_account_expired_at.strftime("%Y-%m-%d %H:%M:%S") == resp_data["account_expired_at"] assert data_source_user.username == resp_data["username"] assert data_source_user.full_name == resp_data["full_name"] diff --git a/src/bk-user/tests/biz/conftest.py b/src/bk-user/tests/biz/conftest.py new file mode 100644 index 000000000..e2f32ab4f --- /dev/null +++ b/src/bk-user/tests/biz/conftest.py @@ -0,0 +1,69 @@ +# -*- 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 typing import List + +import pytest +from bkuser.apps.data_source.constants import DataSourcePluginEnum +from bkuser.apps.data_source.models import ( + DataSource, + DataSourceDepartment, + DataSourceUser, +) +from bkuser.apps.tenant.models import TenantDepartment, TenantUser + +from tests.test_utils.data_source import ( + create_data_source_departments_with_relations, + create_data_source_users_with_relations, +) +from tests.test_utils.helpers import generate_random_string +from tests.test_utils.tenant import create_tenant_departments, create_tenant_users + + +@pytest.fixture() +def local_data_source(default_tenant) -> DataSource: + """ + 创建测试数据源 + """ + return DataSource.objects.create( + name=generate_random_string(), owner_tenant_id=default_tenant, plugin_id=DataSourcePluginEnum.LOCAL + ) + + +@pytest.fixture() +def local_data_source_departments(local_data_source) -> List[DataSourceDepartment]: + """ + 创建测试数据源部门，并以首个对象作为父部门 + """ + return create_data_source_departments_with_relations(local_data_source) + + +@pytest.fixture() +def local_data_source_users(local_data_source, local_data_source_departments) -> List[DataSourceUser]: + """ + 创建测试数据源用户， 并以首个对象作为leader, 随机关联部门 + """ + return create_data_source_users_with_relations(local_data_source, local_data_source_departments) + + +@pytest.fixture() +def tenant_users(default_tenant, local_data_source_users) -> List[TenantUser]: + """ + 根据测试数据源用户，创建租户用户 + """ + return create_tenant_users(default_tenant, local_data_source_users) + + +@pytest.fixture() +def tenant_departments(default_tenant, local_data_source, local_data_source_departments) -> List[TenantDepartment]: + """ + 根据测试数据源部门，创建租户部门 + """ + return create_tenant_departments(default_tenant, local_data_source_departments) diff --git a/src/bk-user/tests/biz/test_data_source.py b/src/bk-user/tests/biz/test_data_source.py index bcf3d9f3e..fad9e91cf 100644 --- a/src/bk-user/tests/biz/test_data_source.py +++ b/src/bk-user/tests/biz/test_data_source.py @@ -15,7 +15,6 @@ DataSourceDepartment, DataSourceDepartmentRelation, DataSourceDepartmentUserRelation, - DataSourceUser, DataSourceUserLeaderRelation, ) from bkuser.biz.data_source import DataSourceDepartmentHandler, DataSourceUserHandler @@ -24,7 +23,7 @@ class TestDataSourceDepartmentHandler: - def test_get_department_info_map_by_ids(self, local_data_source_departments: List[DataSourceDepartment]): + def test_get_department_info_map_by_ids(self, local_data_source_departments): departments_map = DataSourceDepartmentHandler.get_department_info_map_by_ids( [department.id for department in local_data_source_departments] ) @@ -51,15 +50,15 @@ def test_get_department_info_map_by_ids(self, local_data_source_departments: Lis [14, 24, 34], ], ) - def test_not_exist_get_department_info_map_by_ids(self, not_exist_data_source_department_ids: List[int]): + def test_not_exist_get_department_info_map_by_ids(self, not_exist_data_source_department_ids): departments_map = DataSourceDepartmentHandler.get_department_info_map_by_ids( not_exist_data_source_department_ids ) + data_source_departments = DataSourceDepartment.objects.filter(id__in=not_exist_data_source_department_ids) + assert data_source_departments.count() == len(departments_map.keys()) assert not departments_map - def test_get_user_department_ids_map( - self, local_data_source_departments: List[DataSourceDepartment], local_data_source_users: List[DataSourceUser] - ): + def test_get_user_department_ids_map(self, local_data_source_departments, local_data_source_users): user_ids = [user.id for user in local_data_source_users] user_department_relationship_map = DataSourceDepartmentHandler.get_user_department_ids_map(user_ids) @@ -87,7 +86,7 @@ def test_not_exist_get_user_department_ids_map(self, not_exist_data_source_user_ class TestDataSourceUserHandler: - def test_get_user_leader_ids_map(self, local_data_source_users: List[DataSourceUser]): + def test_get_user_leader_ids_map(self, local_data_source_users): data_source_user_ids = [item.id for item in local_data_source_users] leader_ids_map = DataSourceUserHandler.get_user_leader_ids_map(data_source_user_ids) diff --git a/src/bk-user/tests/biz/test_tenant.py b/src/bk-user/tests/biz/test_tenant.py index eb1a955d9..acb0a3cd6 100644 --- a/src/bk-user/tests/biz/test_tenant.py +++ b/src/bk-user/tests/biz/test_tenant.py @@ -12,7 +12,6 @@ import pytest from bkuser.apps.data_source.models import ( - DataSourceDepartment, DataSourceDepartmentRelation, DataSourceDepartmentUserRelation, DataSourceUserLeaderRelation, @@ -24,12 +23,12 @@ class TestTenantUserHandler: - def test_get_tenant_user_leaders_map_by_id(self, default_tenant_users: List[TenantUser]): + def test_get_tenant_user_leaders_map_by_id(self, tenant_users): tenant_user_leader_map = TenantUserHandler.get_tenant_user_leaders_map_by_id( - [item.id for item in default_tenant_users] + [item.id for item in tenant_users] ) - for tenant_user in default_tenant_users: + for tenant_user in tenant_users: tenant_user_leader_ids = [leader_info.id for leader_info in tenant_user_leader_map.get(tenant_user.id, [])] data_source_leader_ids = DataSourceUserLeaderRelation.objects.filter( user=tenant_user.data_source_user @@ -42,65 +41,62 @@ def test_get_tenant_user_leaders_map_by_id(self, default_tenant_users: List[Tena def test_get_tenant_user_departments_map_by_id( self, - default_tenant: str, - default_tenant_users: List[TenantUser], - default_tenant_departments: List[TenantDepartment], + default_tenant, + tenant_users, + tenant_departments, ): tenant_user_departments_map = TenantUserHandler.get_tenant_user_departments_map_by_id( - [user.id for user in default_tenant_users] + [user.id for user in tenant_users] ) - for tenant_user in default_tenant_users: + for tenant_user in tenant_users: tenant_departments = tenant_user_departments_map.get(tenant_user.id, []) - tenant_departments_ids = [tenant_department.id for tenant_department in tenant_departments] - data_source_department = TenantDepartment.objects.filter( - id__in=tenant_departments_ids, tenant=default_tenant - ) + data_source_departments = TenantDepartment.objects.filter( + id__in=[tenant_department.id for tenant_department in tenant_departments], tenant=default_tenant + ).values_list("data_source_department", flat=True) - user_related_department = DataSourceDepartmentUserRelation.objects.filter( + user_related_departments = DataSourceDepartmentUserRelation.objects.filter( user_id=tenant_user.data_source_user_id ) - assert not set(user_related_department.values_list("department_id", flat=True)) - set( - data_source_department.values_list("data_source_department_id", flat=True) + assert not set(user_related_departments.values_list("department_id", flat=True)) - set( + data_source_departments.values_list("data_source_department_id", flat=True) ) - def test_get_tenant_user_ids_by_tenant(self, default_tenant: str): + def test_get_tenant_user_ids_by_tenant(self, default_tenant): tenant_user_ids = TenantUserHandler.get_tenant_user_ids_by_tenant(default_tenant) - assert len(tenant_user_ids) == TenantUser.objects.filter(tenant=default_tenant).count() - assert not set(tenant_user_ids) - set( - TenantUser.objects.filter(tenant=default_tenant).values_list("id", flat=True) - ) + assert TenantUser.objects.filter(id__in=tenant_user_ids).count() == len(tenant_user_ids) class TestTenantDepartmentHandler: def test_convert_data_source_department_to_tenant_department( self, - default_tenant: str, - local_data_source_departments: List[DataSourceDepartment], - default_tenant_departments: List[TenantDepartment], + default_tenant, + local_data_source_departments, + tenant_departments, ): data_source_department_ids = [department.id for department in local_data_source_departments] - tenant_departments = TenantDepartmentHandler.convert_data_source_department_to_tenant_department( + tenant_department_info_list = TenantDepartmentHandler.convert_data_source_department_to_tenant_department( default_tenant, data_source_department_ids ) assert len(data_source_department_ids) == len(tenant_departments) - for department in tenant_departments: - tenant_department = TenantDepartment.objects.get(id=department.id) + for tenant_department_info in tenant_department_info_list: + tenant_department = TenantDepartment.objects.get(id=tenant_department_info.id) assert tenant_department.data_source_department_id in data_source_department_ids - assert department.name == tenant_department.data_source_department.name + assert tenant_department_info.name == tenant_department.data_source_department.name - children_id = ( + child_ids = ( DataSourceDepartmentRelation.objects.get(department=tenant_department.data_source_department) .get_children() .values_list("department_id", flat=True) ) - - tenant_children_departments = TenantDepartment.objects.filter(data_source_department_id__in=children_id) - assert department.has_children == tenant_children_departments.exists() + assert ( + tenant_department_info.has_children + == TenantDepartment.objects.filter(data_source_department_id__in=child_ids).exists() + ) @pytest.mark.parametrize( "not_exist_data_source_department_ids", diff --git a/src/bk-user/tests/conftest.py b/src/bk-user/tests/conftest.py index 78e7b38bb..3d0f3d252 100644 --- a/src/bk-user/tests/conftest.py +++ b/src/bk-user/tests/conftest.py @@ -8,24 +8,13 @@ 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 typing import List import pytest -from bkuser.apps.data_source.models import ( - DataSource, - DataSourceDepartment, - DataSourceUser, -) -from bkuser.apps.tenant.models import TenantDepartment, TenantUser from bkuser.auth.models import User from tests.test_utils.auth import create_user -from tests.test_utils.data_source import ( - create_data_source_departments_with_relationship, - create_data_source_users_with_relationship, -) from tests.test_utils.helpers import generate_random_string -from tests.test_utils.tenant import create_tenant, create_tenant_departments, create_tenant_users +from tests.test_utils.tenant import create_tenant @pytest.fixture() @@ -41,56 +30,6 @@ def random_tenant() -> str: @pytest.fixture() -def bk_user(default_tenant: str) -> User: +def bk_user() -> User: """生成随机用户""" return create_user() - - -@pytest.fixture() -def local_data_source(default_tenant: str) -> DataSource: - """ - 创建测试数据源 - """ - return DataSource.objects.create(name="local_data_source", owner_tenant_id=default_tenant, plugin_id="local") - - -@pytest.fixture() -def local_data_source_departments(local_data_source: DataSource) -> List[DataSourceDepartment]: - """ - 创建测试数据源部门，并以首个对象作为父部门 - """ - return create_data_source_departments_with_relationship(local_data_source.id) - - -@pytest.fixture() -def local_data_source_users( - local_data_source: DataSource, local_data_source_departments: List[DataSourceDepartment] -) -> List[DataSourceUser]: - """ - 创建测试数据源用户， 并以首个对象作为leader, 随机关联部门 - """ - return create_data_source_users_with_relationship( - local_data_source.id, [department.id for department in local_data_source_departments] - ) - - -@pytest.fixture() -def default_tenant_users( - default_tenant: str, local_data_source: DataSource, local_data_source_users: List[DataSourceUser] -) -> List[TenantUser]: - """ - 根据测试数据源用户，创建租户用户 - """ - return create_tenant_users(default_tenant, local_data_source.id, [user.id for user in local_data_source_users]) - - -@pytest.fixture() -def default_tenant_departments( - default_tenant: str, local_data_source: DataSource, local_data_source_departments: List[DataSourceDepartment] -) -> List[TenantDepartment]: - """ - 根据测试数据源部门，创建租户部门 - """ - return create_tenant_departments( - default_tenant, local_data_source.id, [department.id for department in local_data_source_departments] - ) diff --git a/src/bk-user/tests/test_utils/data_source.py b/src/bk-user/tests/test_utils/data_source.py index d3c497672..54b3c627b 100644 --- a/src/bk-user/tests/test_utils/data_source.py +++ b/src/bk-user/tests/test_utils/data_source.py @@ -18,42 +18,63 @@ DataSourceUser, DataSourceUserLeaderRelation, ) +from tests.test_utils.helpers import generate_random_string -def create_data_source_departments_with_relationship(data_source_id: int): +def create_data_source_departments_with_relations(data_source) -> List[DataSourceDepartment]: """ 创建数据源部门，并以首个对象为其余对象的父部门 """ - data_source_departments: List[DataSourceDepartment] = [] - for i in range(10): - department = DataSourceDepartment.objects.create(data_source_id=data_source_id, name=f"fake_dept_{i}") - data_source_departments.append(department) + departments = [DataSourceDepartment(data_source=data_source, name=generate_random_string()) for _ in range(10)] + DataSourceDepartment.objects.bulk_create(departments) + data_source_departments = list(DataSourceDepartment.objects.filter(data_source=data_source)) # 添加部门关系 root = DataSourceDepartmentRelation.objects.create( - department=data_source_departments[0], data_source_id=data_source_id + department=data_source_departments[0], data_source=data_source, parent=None ) - for item in data_source_departments[1:]: - DataSourceDepartmentRelation.objects.create(department=item, data_source_id=data_source_id, parent=root) + # 组织树重建 + DataSourceDepartmentRelation.objects.rebuild() + + for data_source_department in data_source_departments[1:]: + DataSourceDepartmentRelation.objects.create( + department=data_source_department, data_source=data_source, parent=root + ) + + # 组织树重建 + DataSourceDepartmentRelation.objects.rebuild() return data_source_departments -def create_data_source_users_with_relationship(data_source_id: int, department_ids: List[int]): +def create_data_source_users_with_relations(data_source, departments) -> List[DataSourceUser]: """ 创建数据源用户，并以首个对象为其余对象的上级, 随机关联部门 """ - data_source_users: List[DataSourceUser] = [] - for i in range(10): - fake_user = DataSourceUser.objects.create( - full_name=f"fake-user-{i}", username=f"fake-user-{i}", phone="1312345678", data_source_id=data_source_id + users = [ + DataSourceUser( + full_name=generate_random_string(), + username=generate_random_string(), + email=f"{generate_random_string()}@qq.com", + phone="13123456789", + data_source=data_source, ) - data_source_users.append(fake_user) + for _ in range(10) + ] + DataSourceUser.objects.bulk_create(users) + data_source_users = list(DataSourceUser.objects.filter(data_source=data_source)) # 添加上下级关系 - leader = data_source_users[0] - for user in data_source_users[1:]: - DataSourceUserLeaderRelation.objects.get_or_create(user=user, leader=leader) + user_relations = [ + DataSourceUserLeaderRelation(user=data_source_user, leader=data_source_users[0]) + for data_source_user in data_source_users[1:] + ] + DataSourceUserLeaderRelation.objects.bulk_create(user_relations) + + # 添加部门-人员关系 + user_department_relations = [ + DataSourceDepartmentUserRelation(user=data_source_user, department=random.choice(departments)) + for data_source_user in data_source_users + ] + DataSourceDepartmentUserRelation.objects.bulk_create(user_department_relations) - for item in data_source_users: - DataSourceDepartmentUserRelation.objects.get_or_create(user=item, department_id=random.choice(department_ids)) return data_source_users diff --git a/src/bk-user/tests/test_utils/tenant.py b/src/bk-user/tests/test_utils/tenant.py index 284da1fde..36f94ec09 100644 --- a/src/bk-user/tests/test_utils/tenant.py +++ b/src/bk-user/tests/test_utils/tenant.py @@ -32,34 +32,34 @@ def create_tenant(tenant_id: Optional[str] = DEFAULT_TENANT) -> Tenant: return tenant -def create_tenant_users(tenant_id: str, data_source_id: int, data_source_user_ids: List[int]) -> List[TenantUser]: +def create_tenant_users(tenant, data_source_users) -> List[TenantUser]: """ 创建租户用户 """ - tenant_users: List[TenantUser] = [] - for user_id in data_source_user_ids: - tenant_user = TenantUser.objects.create( - data_source_user_id=user_id, - data_source_id=data_source_id, - tenant_id=tenant_id, + tenant_users = [ + TenantUser( + data_source_user=user, + data_source=user.data_source, + tenant_id=tenant, id=generate_uuid(), ) - tenant_users.append(tenant_user) - return tenant_users + for user in data_source_users + ] + TenantUser.objects.bulk_create(tenant_users) + return list(TenantUser.objects.filter(tenant_id=tenant)) -def create_tenant_departments( - tenant_id: str, data_source_id: int, data_source_department_ids: List[int] -) -> List[TenantDepartment]: +def create_tenant_departments(tenant, data_source_departments) -> List[TenantDepartment]: """ 创建租户部门 """ - tenant_departments: List[TenantDepartment] = [] - for department in data_source_department_ids: - tenant_department = TenantDepartment.objects.create( - data_source_department_id=department, - data_source_id=data_source_id, - tenant_id=tenant_id, + tenant_departments = [ + TenantDepartment( + data_source_department=department, + data_source=department.data_source, + tenant_id=tenant, ) - tenant_departments.append(tenant_department) - return tenant_departments + for department in data_source_departments + ] + TenantDepartment.objects.bulk_create(tenant_departments) + return list(TenantDepartment.objects.filter(tenant_id=tenant))