Coverage for src/cstlcore/memberships/router.py: 35%

1from fastapi import APIRouter, Body, Depends, HTTPException 

2from sqlmodel import Session, select 

3 

4from cstlcore.auth.dependencies import get_current_user 

5from cstlcore.constellations.dependencies import get_existing_constellation 

6from cstlcore.constellations.models import Constellation 

7from cstlcore.database.dependencies import get_session 

8from cstlcore.memberships.dependencies import ( 

9 require_admin_access, 

10 require_strictly_higher_access, 

11) 

12from cstlcore.memberships.models import ( 

13 AccessEnum, 

14 ConstellationMembership, 

15 ConstellationMembershipCreate, 

16 ConstellationMembershipPublic, 

17 MemberPublic, 

18) 

19from cstlcore.memberships.services import as_member 

20from cstlcore.users.dependencies import get_user_by_id 

21from cstlcore.users.models import User 

22 

23router = APIRouter() 

24 

25 

26@router.get("/memberships", response_model=list[ConstellationMembershipPublic]) 

27async def get_memberships(session: Session = Depends(get_session)): 

28 db_memberships = session.exec(select(ConstellationMembership)).all() 

29 return db_memberships 

30 

31 

32@router.post("/memberships", response_model=MemberPublic) 

33async def create_membership( 

34 membership: ConstellationMembershipCreate, session: Session = Depends(get_session) 

35): 

36 # Validate that exactly one of user_id or user_email is provided 

37 if not membership.user_id and not membership.user_email: 

38 raise HTTPException( 

39 status_code=400, detail="Either user_id or user_email must be provided" 

40 ) 

41 

42 if membership.user_id and membership.user_email: 

43 raise HTTPException( 

44 status_code=400, detail="Cannot provide both user_id and user_email" 

45 ) 

46 

47 # Determine the user_id to use 

48 user_id = membership.user_id 

49 if membership.user_email: 

50 # Look up user by email 

51 user = session.exec( 

52 select(User).where(User.email == membership.user_email) 

53 ).first() 

54 if not user: 

55 raise HTTPException( 

56 status_code=404, 

57 detail=f"User with email {membership.user_email} not found", 

58 ) 

59 user_id = user.id 

60 # At this point, user_id is guaranteed to be not None due to validation above 

61 assert user_id is not None 

62 

63 # Check if membership already exists 

64 existing_membership = session.exec( 

65 select(ConstellationMembership).where( 

66 ConstellationMembership.constellation_id == membership.constellation_id, 

67 ConstellationMembership.user_id == user_id, 

68 ) 

69 ).first() 

70 

71 if existing_membership: 

72 raise HTTPException( 

73 status_code=409, 

74 detail="Membership already exists for this user and constellation", 

75 ) 

76 

77 # Create the membership with the resolved user_id 

78 db_membership = ConstellationMembership( 

79 constellation_id=membership.constellation_id, 

80 user_id=user_id, 

81 access=membership.access, 

82 ) 

83 session.add(db_membership) 

84 session.commit() 

85 session.refresh(db_membership) 

86 return as_member(db_membership) 

87 

88 

89@router.get( 

90 "/constellations/{constellation_id}/members", response_model=list[MemberPublic] 

91) 

92async def get_constellation_members( 

93 constellation: Constellation = Depends(get_existing_constellation), 

94 session: Session = Depends(get_session), 

95): 

96 memberships = session.exec( 

97 select(ConstellationMembership).where( 

98 ConstellationMembership.constellation_id == constellation.id 

99 ) 

100 ).all() 

101 

102 return [as_member(membership) for membership in memberships] 

103 

104 

105@router.patch( 

106 "/constellations/{constellation_id}/members/{user_id}/access", 

107 response_model=MemberPublic, 

108 dependencies=[Depends(require_strictly_higher_access)], 

109) 

110async def update_member_access( 

111 access: AccessEnum = Body(..., embed=True), 

112 constellation: Constellation = Depends(require_admin_access), 

113 user: User = Depends(get_user_by_id), 

114 current_user: User = Depends(get_current_user), 

115 session: Session = Depends(get_session), 

116): 

117 if current_user.id == user.id: 

118 raise HTTPException( 

119 status_code=400, 

120 detail="You cannot change your own access level", 

121 ) 

122 

123 membership = session.exec( 

124 select(ConstellationMembership).where( 

125 ConstellationMembership.constellation_id == constellation.id, 

126 ConstellationMembership.user_id == user.id, 

127 ) 

128 ).first() 

129 

130 if not membership: 

131 raise HTTPException( 

132 status_code=404, 

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

134 ) 

135 

136 membership.access = access 

137 session.add(membership) 

138 session.commit() 

139 session.refresh(membership) 

140 

141 return as_member(membership) 

142 

143 

144@router.delete( 

145 "/constellations/{constellation_id}/members/{user_id}", 

146 dependencies=[Depends(require_strictly_higher_access)], 

147 status_code=204, 

148) 

149async def delete_member( 

150 constellation: Constellation = Depends(require_admin_access), 

151 user: User = Depends(get_user_by_id), 

152 current_user: User = Depends(get_current_user), 

153 session: Session = Depends(get_session), 

154): 

155 if current_user.id == user.id: 

156 raise HTTPException( 

157 status_code=400, 

158 detail="You cannot remove yourself from the constellation", 

159 ) 

160 

161 membership = session.exec( 

162 select(ConstellationMembership).where( 

163 ConstellationMembership.constellation_id == constellation.id, 

164 ConstellationMembership.user_id == user.id, 

165 ) 

166 ).first() 

167 

168 if not membership: 

169 raise HTTPException( 

170 status_code=404, 

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

172 ) 

173 

174 session.delete(membership) 

175 session.commit()