Coverage for api/handlers/authorization/company_employee_access.py: 16%

1from contextlib import suppress 

2from typing import Any 

3 

4from bson import ObjectId 

5from exceptions import NotFoundHTTPError, SubsidiaryIsNotAssigned 

6from handlers.authorization.check_role import has_role 

7from mongodb import clients_col, cm_clients_col 

8from sotrans_models.models.misc.client import ( 

9 ClientDBModel, 

10 ManagersClientsDBModel, 

11) 

12from sotrans_models.models.orders.order import OrderDBModel 

13from sotrans_models.models.roles import SotransRole 

14from sotrans_models.models.users import SotransOIDCUserModel 

15from utils.helper import get_subsidiary_oid 

16 

17################################################### 

18# OPTIMIZED: # 

19# assert_is_assigned_to_client_query # 

20# flexible_company_assertion_query # 

21################################################### 

22 

23 

24async def assert_is_assigned_to_client_query( 

25 executor: SotransOIDCUserModel, 

26 for_grabber: bool = False, 

27 with_sub: bool = True, 

28) -> dict[str, Any]: 

29 if has_role(executor, SotransRole.company_director): 

30 return {} 

31 assigned_clients = await get_assigned_clients_ids(executor) 

32 if for_grabber: 

33 assigned_clients = list(map(str, assigned_clients)) 

34 client_filter = {"client.id": {"$in": assigned_clients}} 

35 if with_sub: 

36 with suppress(SubsidiaryIsNotAssigned): 

37 sub_id = await get_subsidiary_oid(executor) 

38 client_filter = { 

39 "$or": [ # type: ignore[dict-item] 

40 client_filter, 

41 {OrderDBModel.client.subsidiary.id: sub_id}, 

42 ] 

43 } 

44 return client_filter 

45 

46 

47async def manager_as_higher_role_query( 

48 user: SotransOIDCUserModel, for_grabber: bool = False 

49) -> dict[str, Any]: 

50 if has_role(user, SotransRole.company_director): 

51 return {} 

52 clients_by_responsible = await assert_is_assigned_to_client_query( 

53 user, for_grabber, with_sub=False 

54 ) 

55 assigned_subsidiary_or_logistician_for_order = ( 

56 assigned_subsidiary_or_logistician_for_order_query( 

57 user, to_grabber=for_grabber 

58 ) 

59 ) 

60 complex_filter = [clients_by_responsible] 

61 if "$or" in assigned_subsidiary_or_logistician_for_order: 

62 complex_filter.extend( 

63 assigned_subsidiary_or_logistician_for_order["$or"] 

64 ) 

65 else: 

66 complex_filter.append(assigned_subsidiary_or_logistician_for_order) 

67 return {"$or": complex_filter} 

68 

69 

70async def flexible_company_assertion_query( 

71 user: SotransOIDCUserModel, to_grabber: bool = False 

72) -> dict[str, Any]: 

73 if has_role(user, SotransRole.company_manager): 

74 return await manager_as_higher_role_query(user, to_grabber) 

75 return assigned_subsidiary_or_logistician_for_order_query( 

76 user, to_grabber=to_grabber 

77 ) 

78 

79 

80################################################# 

81# WITH SUBSIDIARIES # 

82################################################# 

83 

84 

85def assigned_subsidiary_or_logistician_for_order_query( 

86 user: SotransOIDCUserModel, *, to_grabber: bool = False 

87): 

88 if has_role(user, SotransRole.company_director): 

89 return {} 

90 query = { 

91 OrderDBModel.assigned.company.employee.id: str(user.sub) 

92 if to_grabber 

93 else user.sub 

94 } 

95 if user.subsidiary_id: 

96 sub_id = ( 

97 user.subsidiary_id if to_grabber else ObjectId(user.subsidiary_id) 

98 ) 

99 subsidiary_order_q = { 

100 OrderDBModel.assigned.company.subsidiary.id: sub_id 

101 } 

102 subsidiary_client_q = {OrderDBModel.client.subsidiary.id: sub_id} 

103 query = {"$or": [subsidiary_client_q, subsidiary_order_q, query]} 

104 return query 

105 

106 

107async def get_assigned_clients_ids( 

108 executor: SotransOIDCUserModel, 

109) -> list[ObjectId]: 

110 if has_role(executor, SotransRole.company_director): 

111 assigned_clients = await clients_col.find_batch({}) 

112 return [ca["_id"] for ca in assigned_clients] 

113 elif has_role(executor, SotransRole.company_logistician): 

114 assigned_clients = await cm_clients_col.find_single( 

115 "manager_id", executor.sub 

116 ) 

117 if not assigned_clients: 

118 return [] 

119 return ManagersClientsDBModel(**assigned_clients).clients_ids 

120 else: 

121 return [] 

122 

123 

124async def assert_client_assigned( 

125 executor: SotransOIDCUserModel, client_id: ObjectId 

126): 

127 if has_role(executor, SotransRole.company_director): 

128 return True 

129 elif has_role(executor, SotransRole.company_logistician): 

130 client_ids = await get_assigned_clients_ids(executor) 

131 if client_id in client_ids: 

132 return True 

133 with suppress(SubsidiaryIsNotAssigned): 

134 sub_id = await get_subsidiary_oid(executor) 

135 client = await clients_col.find_single("_id", client_id) 

136 if client is None: 

137 raise NotFoundHTTPError("клиент") 

138 cm = ClientDBModel(**client) 

139 if cm.subsidiary and cm.subsidiary.id == sub_id: 

140 return True 

141 return False