Coverage for api/endpoints/orders/confirmed_orders.py: 79%

1from fastapi import APIRouter, Depends 

2from handlers.orders.confirm_orders import ( 

3 on_confirmed_to_appointment, 

4 on_draw_up_order, 

5 on_get_confirmed, 

6 on_get_one_confirmed, 

7 on_notify_winner, 

8 on_put_back_to_auc, 

9 on_put_to_active, 

10) 

11from keycloak import idp 

12from sotrans_models.models._mongo import PydanticObjectIdPath 

13from sotrans_models.models.orders.order import ( 

14 ConfirmedOrderUpdateModel, 

15 OrderDBModel, 

16 OrderUpdateModel, 

17) 

18from sotrans_models.models.responses import ( 

19 AucEndNotificationCall, 

20 ErrorRepr, 

21 GenericGetListResponse, 

22 MultiselectError, 

23) 

24from sotrans_models.models.roles import SotransRole 

25from sotrans_models.models.users import SotransOIDCUserModel 

26from typing_extensions import Annotated 

27from utils.access_wrapper import get_active_user 

28from utils.data_grabber import BaseGetListQueryParams, BaseGetOneQueryParams 

29from utils.helper import multiexec 

30 

31confirmed_orders_router = APIRouter( 

32 prefix="/confirmed-orders", tags=["confirmed", "orders"] 

33) 

34 

35 

36@confirmed_orders_router.put("/{order_ids}/to-exchange") 

37async def put_order_to_auction( 

38 order_ids: str, 

39 user: Annotated[ 

40 SotransOIDCUserModel, 

41 Depends( 

42 idp.get_current_user( 

43 required_role_names=[SotransRole.company_logistician] 

44 ) 

45 ), 

46 ], 

47 order: OrderUpdateModel, 

48) -> MultiselectError: 

49 """Accepts start_price(required), assigned(optional) and auction_end_time(optional).""" 

50 return await multiexec(order_ids, on_put_back_to_auc, user, order) 

51 

52 

53@confirmed_orders_router.put("/{order_ids}/to-active") 

54async def put_order_to_active( 

55 order_ids: str, 

56 user: Annotated[ 

57 SotransOIDCUserModel, 

58 Depends( 

59 idp.get_current_user( 

60 required_role_names=[SotransRole.company_logistician] 

61 ) 

62 ), 

63 ], 

64 order: OrderUpdateModel | None = None, 

65) -> MultiselectError: 

66 """ 

67 Throws Not Found on wrong subsidiary/employee, another than unverified status. 

68 Accepts end_price and assigned(both are optional). 

69 """ 

70 return await multiexec(order_ids, on_put_to_active, user, order) 

71 

72 

73@confirmed_orders_router.put( 

74 "/{order_id}/to-unverified", 

75 response_model_exclude_unset=True, 

76 responses={404: {"model": ErrorRepr}}, 

77) 

78async def draw_up_order( 

79 order: ConfirmedOrderUpdateModel, 

80 order_id: PydanticObjectIdPath, 

81 user=Depends(get_active_user([SotransRole.carrier_logistician])), 

82) -> OrderDBModel: 

83 """Restricted carrier by organization_id, by confirmation_end_time, by id""" 

84 return await on_draw_up_order(order_id, user, order) 

85 

86 

87@confirmed_orders_router.get("", response_model_exclude_unset=True) 

88async def get_confirmed_orders( 

89 user=Depends(get_active_user([SotransRole.carrier_logistician])), 

90 params: BaseGetListQueryParams = Depends(), 

91) -> GenericGetListResponse[OrderDBModel]: 

92 return await on_get_confirmed(user, params) 

93 

94 

95@confirmed_orders_router.put("/{order_ids}/to-appointment") 

96async def confirmed_to_appointment( 

97 order_ids: str, 

98 user: Annotated[ 

99 SotransOIDCUserModel, 

100 Depends( 

101 idp.get_current_user( 

102 required_role_names=[SotransRole.company_logistician] 

103 ) 

104 ), 

105 ], 

106 order: OrderUpdateModel | None = None, 

107) -> MultiselectError: 

108 return await multiexec(order_ids, on_confirmed_to_appointment, user, order) 

109 

110 

111@confirmed_orders_router.post("/notify_winner") 

112async def notify_winner( 

113 _: Annotated[ 

114 SotransOIDCUserModel, 

115 Depends( 

116 idp.get_current_user(required_role_names=[SotransRole.scheduled]) 

117 ), 

118 ], 

119 call_data: AucEndNotificationCall, 

120): 

121 await on_notify_winner(call_data) 

122 

123 

124@confirmed_orders_router.get( 

125 "/{order_id}", 

126 responses={404: {"model": ErrorRepr}}, 

127 response_model_exclude_unset=True, 

128) 

129async def get_confirmed_by_id( 

130 order_id: PydanticObjectIdPath, 

131 user=Depends(get_active_user([SotransRole.carrier_logistician])), 

132 params: BaseGetOneQueryParams = Depends(), 

133) -> OrderDBModel: 

134 return await on_get_one_confirmed(order_id, user, params)