Coverage for api/handlers/orders/trash_orders.py: 52%

1import asyncio 

2from typing import Any 

3 

4from bson import ObjectId 

5from database.orders import move_order_and_return 

6from mongodb import buf_col, trash_orders_col 

7from pymongo.results import DeleteResult 

8from services.notifications.director import notification_api 

9from sotrans_models.models.orders.order import ( 

10 OrderDBModel, 

11 OrderStatus, 

12 OrderUpdateModel, 

13) 

14from utils.helper import etag_detalizer 

15 

16 

17async def on_put_trash_to_buffer( 

18 order_id: ObjectId, 

19) -> dict[str, Any] | OrderDBModel: 

20 updated = await move_order_and_return( 

21 order_id, trash_orders_col, buf_col, OrderStatus.buffer 

22 ) 

23 asyncio.create_task( 

24 notification_api.new_in_buffer(OrderDBModel(**updated)) 

25 ) 

26 return updated 

27 

28 

29async def on_remove_from_trash( 

30 order_id: ObjectId, order: OrderUpdateModel | None 

31): 

32 etag_q = ( 

33 {} if order is None or order.etag is None else {"etag": order.etag} 

34 ) 

35 deleted: DeleteResult = await trash_orders_col.collection.delete_one( 

36 {"id": order_id} | etag_q 

37 ) 

38 if deleted.deleted_count == 1: 

39 return 

40 await etag_detalizer(trash_orders_col, etag_q, {"id": order_id})