Coverage for src/cstlcore/memberships/dependencies.py: 83%

1from fastapi import Depends, HTTPException 

2from sqlmodel import Session, select 

3 

4from cstlcore.auth.dependencies import get_current_user 

5from cstlcore.collections.dependencies import get_existing_collection 

6from cstlcore.collections.models import Collection 

7from cstlcore.constellations.dependencies import get_existing_constellation 

8from cstlcore.constellations.models import Constellation 

9from cstlcore.database.dependencies import get_session 

10from cstlcore.maps.dependencies import get_existing_map # Make sure this exists 

11from cstlcore.maps.models import Map 

12from cstlcore.memberships.access import require_constellation_access 

13from cstlcore.memberships.models import AccessEnum, ConstellationMembership 

14from cstlcore.users.dependencies import get_user_by_id 

15from cstlcore.users.models import User 

16 

17 

18def require_strictly_higher_access( 

19 constellation: Constellation = Depends(get_existing_constellation), 

20 user: User = Depends(get_user_by_id), 

21 current_user: User = Depends(get_current_user), 

22 session: Session = Depends(get_session), 

23): 

24 if current_user.id == user.id: 

25 raise HTTPException( 

26 status_code=400, 

27 detail="You cannot perform this action on yourself.", 

28 ) 

29 

30 current_membership = session.exec( 

31 select(ConstellationMembership).where( 

32 ConstellationMembership.constellation_id == constellation.id, 

33 ConstellationMembership.user_id == current_user.id, 

34 ) 

35 ).first() 

36 target_membership = session.exec( 

37 select(ConstellationMembership).where( 

38 ConstellationMembership.constellation_id == constellation.id, 

39 ConstellationMembership.user_id == user.id, 

40 ) 

41 ).first() 

42 

43 if not target_membership: 

44 raise HTTPException( 

45 status_code=404, 

46 detail="Membership not found for this user in constellation", 

47 ) 

48 

49 if not current_membership or current_membership.access <= target_membership.access: 

50 raise HTTPException( 

51 status_code=403, 

52 detail="You must have a higher access level than the target user.", 

53 ) 

54 

55 

56def make_collection_access_dependency(required_level: AccessEnum): 

57 def dependency( 

58 user: User = Depends(get_current_user), 

59 collection: Collection = Depends(get_existing_collection), 

60 session: Session = Depends(get_session), 

61 ) -> Collection: 

62 require_constellation_access( 

63 user=user, 

64 constellation=collection.constellation, 

65 required_level=required_level, 

66 session=session, 

67 ) 

68 return collection 

69 

70 return dependency 

71 

72 

73def make_constellation_access_dependency(required_level: AccessEnum): 

74 def dependency( 

75 user: User = Depends(get_current_user), 

76 constellation: Constellation = Depends(get_existing_constellation), 

77 session: Session = Depends(get_session), 

78 ) -> Constellation: 

79 return require_constellation_access( 

80 user=user, 

81 constellation=constellation, 

82 required_level=required_level, 

83 session=session, 

84 ) 

85 

86 return dependency 

87 

88 

89def make_map_access_dependency(required_level: AccessEnum): 

90 def dependency( 

91 user: User = Depends(get_current_user), 

92 _map: Map = Depends(get_existing_map), 

93 session: Session = Depends(get_session), 

94 ) -> Map: 

95 # Access is checked through the map's collection 

96 require_constellation_access( 

97 user=user, 

98 constellation=_map.collection.constellation, 

99 required_level=required_level, 

100 session=session, 

101 ) 

102 return _map 

103 

104 return dependency 

105 

106 

107require_read_access_collections = make_collection_access_dependency(AccessEnum.READ) 

108require_write_access_collections = make_collection_access_dependency(AccessEnum.WRITE) 

109require_admin_access_collections = make_collection_access_dependency(AccessEnum.ADMIN) 

110 

111require_read_access = make_constellation_access_dependency(AccessEnum.READ) 

112require_write_access = make_constellation_access_dependency(AccessEnum.WRITE) 

113require_admin_access = make_constellation_access_dependency(AccessEnum.ADMIN) 

114 

115require_read_access_maps = make_map_access_dependency(AccessEnum.READ) 

116require_write_access_maps = make_map_access_dependency(AccessEnum.WRITE) 

117require_admin_access_maps = make_map_access_dependency(AccessEnum.ADMIN)