Coverage for api/handlers/suggestions.py: 25%

1from contextlib import suppress 

2from typing import Any 

3 

4from database.text_search.mongo_search import update_query_by_search 

5from handlers.authorization.check_role import has_role 

6from mongodb import subsidiaries_col, users_col 

7from sotrans_models.models.roles import SotransRole 

8from sotrans_models.models.users import ( 

9 SotransOIDCUserModel, 

10 SotransUserDBFieldsModel, 

11) 

12from utils.helper import get_org_oid 

13 

14 

15async def on_get_sub_suggestions( 

16 search_query: str, 

17 skip: int, 

18 limit: int, 

19): 

20 query: dict = {} 

21 if search_query: 

22 await update_query_by_search(subsidiaries_col, search_query, query) 

23 subsidiaries = await subsidiaries_col.find_batch(query, skip, limit) 

24 return [{"value": str(e["_id"]), "label": e["name"]} for e in subsidiaries] 

25 

26 

27async def on_get_users_suggestions( 

28 user: SotransOIDCUserModel, 

29 search_query: str, 

30 limit: int, 

31 skip: int, 

32 minimal_role: SotransRole | None, 

33) -> list[dict[str, Any]]: 

34 roles_to_search = [] 

35 if minimal_role: 

36 roles_to_search.append(minimal_role.value) 

37 bigger_roles: dict[SotransRole, list[SotransRole]] = { 

38 SotransRole.carrier_logistician: [ 

39 SotransRole.company_logistician, 

40 SotransRole.carrier_logistician, 

41 SotransRole.company_manager, 

42 SotransRole.company_director, 

43 SotransRole.carrier_director, 

44 SotransRole.admin, 

45 ], 

46 SotransRole.company_logistician: [ 

47 SotransRole.company_manager, 

48 SotransRole.company_director, 

49 SotransRole.admin, 

50 ], 

51 SotransRole.company_manager: [ 

52 SotransRole.company_director, 

53 SotransRole.admin, 

54 ], 

55 SotransRole.company_director: [SotransRole.admin], 

56 SotransRole.carrier_director: [ 

57 SotransRole.admin, 

58 SotransRole.company_director, 

59 ], 

60 } 

61 roles_to_search.extend( 

62 role.value for role in bigger_roles.get(minimal_role, []) 

63 ) 

64 org_id = get_org_oid(user) 

65 query = { 

66 SotransUserDBFieldsModel.organization_id: str(org_id), 

67 } 

68 await update_query_by_search(users_col, search_query, query) 

69 if has_role(user, SotransRole.carrier_director) and not has_role( 

70 user, SotransRole.company_director 

71 ): 

72 for role in (SotransRole.company_director, SotransRole.admin): 

73 with suppress(ValueError): 

74 roles_to_search.remove(role.value) 

75 if roles_to_search: 

76 query.update({"role": {"$in": roles_to_search}}) # type: ignore[dict-item] 

77 return await users_col.find_batch(query, skip, limit)