Coverage for app/backend/src/couchers/servicers/groups.py: 76%

1import logging 

2 

3import grpc 

4from google.protobuf import empty_pb2 

5from sqlalchemy import select 

6from sqlalchemy.orm import Session 

7from sqlalchemy.sql import delete, func 

8 

9from couchers.context import CouchersContext 

10from couchers.db import can_moderate_node, get_node_parents_recursively 

11from couchers.event_log import log_event 

12from couchers.models import ( 

13 Cluster, 

14 ClusterRole, 

15 ClusterSubscription, 

16 Discussion, 

17 Event, 

18 EventOccurrence, 

19 Page, 

20 PageType, 

21 User, 

22) 

23from couchers.proto import groups_pb2, groups_pb2_grpc 

24from couchers.servicers.discussions import discussion_to_pb 

25from couchers.servicers.events import apply_occurrence_pagination, event_to_pb, occurrences_next_page_token 

26from couchers.servicers.pages import page_to_pb 

27from couchers.sql import users_visible, where_moderated_content_visible, where_users_column_visible 

28from couchers.utils import Timestamp_from_datetime 

29 

30logger = logging.getLogger(__name__) 

31 

32MAX_PAGINATION_LENGTH = 25 

33 

34 

35def _parents_to_pb(session: Session, cluster: Cluster) -> list[groups_pb2.Parent]: 

36 parents = get_node_parents_recursively(session, cluster.parent_node_id) 

37 return [ 

38 groups_pb2.Parent( 

39 community=groups_pb2.CommunityParent( 

40 community_id=node_id, 

41 name=cluster.name, 

42 slug=cluster.slug, 

43 description=cluster.description, 

44 ) 

45 ) 

46 for node_id, parent_node_id, level, cluster in parents 

47 ] + [ 

48 groups_pb2.Parent( 

49 group=groups_pb2.GroupParent( 

50 group_id=cluster.id, 

51 name=cluster.name, 

52 slug=cluster.slug, 

53 description=cluster.description, 

54 ) 

55 ) 

56 ] 

57 

58 

59def group_to_pb(session: Session, cluster: Cluster, context: CouchersContext) -> groups_pb2.Group: 

60 can_moderate = can_moderate_node(session, context.user_id, cluster.parent_node_id) 

61 

62 member_count = session.execute( 

63 where_users_column_visible( 

64 select(func.count()).select_from(ClusterSubscription).where(ClusterSubscription.cluster_id == cluster.id), 

65 context, 

66 ClusterSubscription.user_id, 

67 ) 

68 ).scalar_one() 

69 is_member = ( 

70 session.execute( 

71 select(ClusterSubscription) 

72 .where(ClusterSubscription.user_id == context.user_id) 

73 .where(ClusterSubscription.cluster_id == cluster.id) 

74 ).scalar_one_or_none() 

75 is not None 

76 ) 

77 

78 admin_count = session.execute( 

79 where_users_column_visible( 

80 select(func.count()) 

81 .select_from(ClusterSubscription) 

82 .where(ClusterSubscription.cluster_id == cluster.id) 

83 .where(ClusterSubscription.role == ClusterRole.admin), 

84 context, 

85 ClusterSubscription.user_id, 

86 ) 

87 ).scalar_one() 

88 is_admin = ( 

89 session.execute( 

90 select(ClusterSubscription) 

91 .where(ClusterSubscription.user_id == context.user_id) 

92 .where(ClusterSubscription.cluster_id == cluster.id) 

93 .where(ClusterSubscription.role == ClusterRole.admin) 

94 ).scalar_one_or_none() 

95 is not None 

96 ) 

97 

98 return groups_pb2.Group( 

99 group_id=cluster.id, 

100 name=cluster.name, 

101 slug=cluster.slug, 

102 description=cluster.description, 

103 created=Timestamp_from_datetime(cluster.created), 

104 parents=_parents_to_pb(session, cluster), 

105 main_page=page_to_pb(session, cluster.main_page, context), 

106 member=is_member, 

107 admin=is_admin, 

108 member_count=member_count, 

109 admin_count=admin_count, 

110 can_moderate=can_moderate, 

111 ) 

112 

113 

114class Groups(groups_pb2_grpc.GroupsServicer): 

115 def GetGroup(self, request: groups_pb2.GetGroupReq, context: CouchersContext, session: Session) -> groups_pb2.Group: 

116 cluster = session.execute( 

117 select(Cluster) 

118 .where(~Cluster.is_official_cluster) # not an official group 

119 .where(Cluster.id == request.group_id) 

120 ).scalar_one_or_none() 

121 if not cluster: 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true

122 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "group_not_found") 

123 

124 return group_to_pb(session, cluster, context) 

125 

126 def ListAdmins( 

127 self, request: groups_pb2.ListAdminsReq, context: CouchersContext, session: Session 

128 ) -> groups_pb2.ListAdminsRes: 

129 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH) 

130 next_admin_id = int(request.page_token) if request.page_token else 0 

131 cluster = session.execute( 

132 select(Cluster).where(~Cluster.is_official_cluster).where(Cluster.id == request.group_id) 

133 ).scalar_one_or_none() 

134 if not cluster: 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true

135 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "group_not_found") 

136 

137 admins = ( 

138 session.execute( 

139 select(User) 

140 .where(users_visible(context)) 

141 .join(ClusterSubscription, ClusterSubscription.user_id == User.id) 

142 .where(ClusterSubscription.cluster_id == cluster.id) 

143 .where(ClusterSubscription.role == ClusterRole.admin) 

144 .where(User.id >= next_admin_id) 

145 .order_by(User.id) 

146 .limit(page_size + 1) 

147 ) 

148 .scalars() 

149 .all() 

150 ) 

151 return groups_pb2.ListAdminsRes( 

152 admin_user_ids=[admin.id for admin in admins[:page_size]], 

153 next_page_token=str(admins[-1].id) if len(admins) > page_size else None, 

154 ) 

155 

156 def ListMembers( 

157 self, request: groups_pb2.ListMembersReq, context: CouchersContext, session: Session 

158 ) -> groups_pb2.ListMembersRes: 

159 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH) 

160 next_member_id = int(request.page_token) if request.page_token else 0 

161 cluster = session.execute( 

162 select(Cluster).where(~Cluster.is_official_cluster).where(Cluster.id == request.group_id) 

163 ).scalar_one_or_none() 

164 if not cluster: 164 ↛ 165line 164 didn't jump to line 165 because the condition on line 164 was never true

165 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "group_not_found") 

166 

167 members = ( 

168 session.execute( 

169 select(User) 

170 .join(ClusterSubscription, ClusterSubscription.user_id == User.id) 

171 .where(users_visible(context)) 

172 .where(ClusterSubscription.cluster_id == cluster.id) 

173 .where(User.id >= next_member_id) 

174 .order_by(User.id) 

175 .limit(page_size + 1) 

176 ) 

177 .scalars() 

178 .all() 

179 ) 

180 return groups_pb2.ListMembersRes( 

181 member_user_ids=[member.id for member in members[:page_size]], 

182 next_page_token=str(members[-1].id) if len(members) > page_size else None, 

183 ) 

184 

185 def ListPlaces( 

186 self, request: groups_pb2.ListPlacesReq, context: CouchersContext, session: Session 

187 ) -> groups_pb2.ListPlacesRes: 

188 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH) 

189 next_page_id = int(request.page_token) if request.page_token else 0 

190 cluster = session.execute( 

191 select(Cluster).where(~Cluster.is_official_cluster).where(Cluster.id == request.group_id) 

192 ).scalar_one_or_none() 

193 if not cluster: 

194 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "group_not_found") 

195 places = ( 

196 cluster.owned_pages.where(Page.type == PageType.place) 

197 .where(Page.id >= next_page_id) 

198 .order_by(Page.id) 

199 .limit(page_size + 1) 

200 .all() 

201 ) 

202 return groups_pb2.ListPlacesRes( 

203 places=[page_to_pb(session, page, context) for page in places[:page_size]], 

204 next_page_token=str(places[-1].id) if len(places) > page_size else None, 

205 ) 

206 

207 def ListGuides( 

208 self, request: groups_pb2.ListGuidesReq, context: CouchersContext, session: Session 

209 ) -> groups_pb2.ListGuidesRes: 

210 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH) 

211 next_page_id = int(request.page_token) if request.page_token else 0 

212 cluster = session.execute( 

213 select(Cluster).where(~Cluster.is_official_cluster).where(Cluster.id == request.group_id) 

214 ).scalar_one_or_none() 

215 if not cluster: 

216 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "group_not_found") 

217 guides = ( 

218 cluster.owned_pages.where(Page.type == PageType.guide) 

219 .where(Page.id >= next_page_id) 

220 .order_by(Page.id) 

221 .limit(page_size + 1) 

222 .all() 

223 ) 

224 return groups_pb2.ListGuidesRes( 

225 guides=[page_to_pb(session, page, context) for page in guides[:page_size]], 

226 next_page_token=str(guides[-1].id) if len(guides) > page_size else None, 

227 ) 

228 

229 def ListEvents( 

230 self, request: groups_pb2.ListEventsReq, context: CouchersContext, session: Session 

231 ) -> groups_pb2.ListEventsRes: 

232 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH) 

233 

234 cluster = session.execute( 

235 select(Cluster).where(~Cluster.is_official_cluster).where(Cluster.id == request.group_id) 

236 ).scalar_one_or_none() 

237 if not cluster: 237 ↛ 238line 237 didn't jump to line 238 because the condition on line 237 was never true

238 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "group_not_found") 

239 

240 query = ( 

241 select(EventOccurrence) 

242 .join(Event, Event.id == EventOccurrence.event_id) 

243 .where(Event.owner_cluster == cluster) 

244 ) 

245 query = where_moderated_content_visible(query, context, EventOccurrence, is_list_operation=True) 

246 

247 query = apply_occurrence_pagination(query, request.page_token, request.past) 

248 

249 query = query.limit(page_size + 1) 

250 occurrences = session.execute(query).scalars().all() 

251 

252 return groups_pb2.ListEventsRes( 

253 events=[event_to_pb(session, occurrence, context) for occurrence in occurrences[:page_size]], 

254 next_page_token=occurrences_next_page_token(occurrences, page_size), 

255 ) 

256 

257 def ListDiscussions( 

258 self, request: groups_pb2.ListDiscussionsReq, context: CouchersContext, session: Session 

259 ) -> groups_pb2.ListDiscussionsRes: 

260 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH) 

261 next_page_id = int(request.page_token) if request.page_token else 0 

262 cluster = session.execute( 

263 select(Cluster).where(~Cluster.is_official_cluster).where(Cluster.id == request.group_id) 

264 ).scalar_one_or_none() 

265 if not cluster: 265 ↛ 266line 265 didn't jump to line 266 because the condition on line 265 was never true

266 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "community_not_found") 

267 query = select(Discussion).where(Discussion.owner_cluster_id == cluster.id).where(Discussion.id >= next_page_id) 

268 query = where_moderated_content_visible(query, context, Discussion, is_list_operation=True) 

269 discussions = session.execute(query.order_by(Discussion.id).limit(page_size + 1)).scalars().all() 

270 return groups_pb2.ListDiscussionsRes( 

271 discussions=[discussion_to_pb(session, discussion, context) for discussion in discussions[:page_size]], 

272 next_page_token=str(discussions[-1].id) if len(discussions) > page_size else None, 

273 ) 

274 

275 def JoinGroup( 

276 self, request: groups_pb2.JoinGroupReq, context: CouchersContext, session: Session 

277 ) -> empty_pb2.Empty: 

278 cluster = session.execute( 

279 select(Cluster).where(~Cluster.is_official_cluster).where(Cluster.id == request.group_id) 

280 ).scalar_one_or_none() 

281 if not cluster: 281 ↛ 282line 281 didn't jump to line 282 because the condition on line 281 was never true

282 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "group_not_found") 

283 

284 user_in_group = cluster.members.where(User.id == context.user_id).one_or_none() 

285 if user_in_group: 

286 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "already_in_group") 

287 

288 cluster.cluster_subscriptions.append( 

289 ClusterSubscription( 

290 user_id=context.user_id, 

291 cluster_id=cluster.id, 

292 role=ClusterRole.member, 

293 ) 

294 ) 

295 

296 log_event(context, session, "group.joined", {"group_id": cluster.id, "group_name": cluster.name}) 

297 

298 return empty_pb2.Empty() 

299 

300 def LeaveGroup( 

301 self, request: groups_pb2.LeaveGroupReq, context: CouchersContext, session: Session 

302 ) -> empty_pb2.Empty: 

303 cluster = session.execute( 

304 select(Cluster).where(~Cluster.is_official_cluster).where(Cluster.id == request.group_id) 

305 ).scalar_one_or_none() 

306 if not cluster: 306 ↛ 307line 306 didn't jump to line 307 because the condition on line 306 was never true

307 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "group_not_found") 

308 

309 user_in_group = cluster.members.where(User.id == context.user_id).one_or_none() 

310 if not user_in_group: 

311 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "not_in_group") 

312 

313 session.execute( 

314 delete(ClusterSubscription) 

315 .where(ClusterSubscription.cluster_id == request.group_id) 

316 .where(ClusterSubscription.user_id == context.user_id) 

317 ) 

318 

319 log_event(context, session, "group.left", {"group_id": cluster.id, "group_name": cluster.name}) 

320 

321 return empty_pb2.Empty() 

322 

323 def ListUserGroups( 

324 self, request: groups_pb2.ListUserGroupsReq, context: CouchersContext, session: Session 

325 ) -> groups_pb2.ListUserGroupsRes: 

326 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH) 

327 next_cluster_id = int(request.page_token) if request.page_token else 0 

328 user_id = request.user_id or context.user_id 

329 clusters = ( 

330 session.execute( 

331 select(Cluster) 

332 .join(ClusterSubscription, ClusterSubscription.cluster_id == Cluster.id) 

333 .where(ClusterSubscription.user_id == user_id) 

334 .where(~Cluster.is_official_cluster) # not an official group 

335 .where(Cluster.id >= next_cluster_id) 

336 .order_by(Cluster.id) 

337 .limit(page_size + 1) 

338 ) 

339 .scalars() 

340 .all() 

341 ) 

342 return groups_pb2.ListUserGroupsRes( 

343 groups=[group_to_pb(session, cluster, context) for cluster in clusters[:page_size]], 

344 next_page_token=str(clusters[-1].id) if len(clusters) > page_size else None, 

345 )