Coverage for api/handlers/resources/common.py: 31%

1from bson import ObjectId 

2from dbcc import MongoTableEngine 

3from exceptions import NoAccessHTTPError, NotFoundHTTPError 

4from handlers.authorization.check_role import has_role 

5from sotrans_models.models._base import EntityHash 

6from sotrans_models.models.resources.drivers import DriverDBModel 

7from sotrans_models.models.resources.trailers import TrailerDBModel 

8from sotrans_models.models.resources.trucks import TruckDBModel 

9from sotrans_models.models.roles import SotransRole 

10from sotrans_models.models.users import SotransOIDCUserModel 

11 

12 

13async def assert_resource_manage( 

14 resource_id: ObjectId, 

15 engine: MongoTableEngine, 

16 executor: SotransOIDCUserModel, 

17 etag: EntityHash | None = None, 

18) -> DriverDBModel | TruckDBModel | TrailerDBModel: 

19 resource_name = { 

20 "trucks": "тягачи", 

21 "trailers": "прицепы", 

22 "drivers": "водители", 

23 }[engine.collection_name] 

24 if etag is not None: 

25 resource = await engine.collection.find_one( 

26 {"_id": resource_id, "etag": etag} 

27 ) 

28 if resource is None: 

29 resource = await engine.find_single("_id", resource_id) 

30 if resource is not None: 

31 raise NotFoundHTTPError("данные устарели") 

32 raise NotFoundHTTPError(f"{resource_name}") 

33 else: 

34 resource = await engine.find_single("_id", resource_id) 

35 if resource is None: 

36 raise NotFoundHTTPError(resource_name) 

37 resource_model = { 

38 "drivers": DriverDBModel, 

39 "trucks": TruckDBModel, 

40 "trailers": TrailerDBModel, 

41 }[engine.collection_name](**resource) 

42 if not has_role( 

43 executor, SotransRole.company_logistician 

44 ) and executor.organization_id != str(resource_model.organization_id): 

45 raise NoAccessHTTPError(resource_name) 

46 return resource_model