Coverage for app / backend / src / couchers / servicers / pages.py: 92%

1import grpc 

2from sqlalchemy import select 

3from sqlalchemy.orm import Session, selectinload 

4 

5from couchers.context import CouchersContext 

6from couchers.db import can_moderate_at, can_moderate_node, get_parent_node_at_location 

7from couchers.models import Cluster, Node, Page, PageType, PageVersion, Thread, Upload, User 

8from couchers.proto import pages_pb2, pages_pb2_grpc 

9from couchers.servicers.threads import thread_to_pb 

10from couchers.utils import Timestamp_from_datetime, create_coordinate, not_none, remove_duplicates_retain_order 

11 

12MAX_PAGINATION_LENGTH = 25 

13 

14pagetype2sql = { 

15 pages_pb2.PAGE_TYPE_PLACE: PageType.place, 

16 pages_pb2.PAGE_TYPE_GUIDE: PageType.guide, 

17 pages_pb2.PAGE_TYPE_MAIN_PAGE: PageType.main_page, 

18} 

19 

20pagetype2api = { 

21 PageType.place: pages_pb2.PAGE_TYPE_PLACE, 

22 PageType.guide: pages_pb2.PAGE_TYPE_GUIDE, 

23 PageType.main_page: pages_pb2.PAGE_TYPE_MAIN_PAGE, 

24} 

25 

26 

27def _is_page_owner(page: Page, user_id: int) -> bool: 

28 """ 

29 Checks whether the user can act as an owner of the page 

30 """ 

31 if page.owner_user: 

32 return page.owner_user_id == user_id 

33 # otherwise owned by a cluster 

34 return not_none(page.owner_cluster).admins.where(User.id == user_id).one_or_none() is not None 

35 

36 

37def _can_moderate_page(session: Session, page: Page, user_id: int) -> bool: 

38 """ 

39 Checks if the user is allowed to moderate this page 

40 """ 

41 # checks if either the page is in the exclusive moderation area of a node 

42 latest_version = page.versions[-1] 

43 

44 # if the page has a location, we firstly check if we are the moderator of any node that contains this page 

45 if latest_version.geom is not None and can_moderate_at(session, user_id, latest_version.geom): 

46 return True 

47 

48 # if the page is owned by a cluster, then any moderator of that cluster can moderate this page 

49 if page.owner_cluster is not None and can_moderate_node(session, user_id, page.owner_cluster.parent_node_id): 

50 return True 

51 

52 # finally check if the user can moderate the parent node of the cluster 

53 return can_moderate_node(session, user_id, page.parent_node_id) 

54 

55 

56def page_to_pb(session: Session, page: Page, context: CouchersContext) -> pages_pb2.Page: 

57 first_version = page.versions[0] 

58 current_version = page.versions[-1] 

59 

60 owner_community_id = None 

61 owner_group_id = None 

62 if page.owner_cluster: 

63 if page.owner_cluster.is_official_cluster: 

64 owner_community_id = page.owner_cluster.parent_node_id 

65 else: 

66 owner_group_id = page.owner_cluster.id 

67 

68 can_moderate = _can_moderate_page(session, page, context.user_id) 

69 

70 return pages_pb2.Page( 

71 page_id=page.id, 

72 type=pagetype2api[page.type], 

73 slug=current_version.slug, 

74 created=Timestamp_from_datetime(first_version.created), 

75 last_edited=Timestamp_from_datetime(current_version.created), 

76 last_editor_user_id=current_version.editor_user_id, 

77 creator_user_id=page.creator_user_id, 

78 owner_user_id=page.owner_user_id, 

79 owner_community_id=owner_community_id, 

80 owner_group_id=owner_group_id, 

81 thread=thread_to_pb(session, page.thread_id), 

82 title=current_version.title, 

83 content=current_version.content, 

84 photo_url=current_version.photo.full_url if current_version.photo_key else None, 

85 address=current_version.address, 

86 location=( 

87 pages_pb2.Coordinate( 

88 lat=current_version.coordinates[0], 

89 lng=current_version.coordinates[1], 

90 ) 

91 if current_version.coordinates 

92 else None 

93 ), 

94 editor_user_ids=remove_duplicates_retain_order([version.editor_user_id for version in page.versions]), 

95 can_edit=_is_page_owner(page, context.user_id) or can_moderate, 

96 can_moderate=can_moderate, 

97 ) 

98 

99 

100class Pages(pages_pb2_grpc.PagesServicer): 

101 def CreatePlace( 

102 self, request: pages_pb2.CreatePlaceReq, context: CouchersContext, session: Session 

103 ) -> pages_pb2.Page: 

104 if not request.title: 

105 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_page_title") 

106 if not request.content: 

107 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_page_content") 

108 if not request.address: 

109 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_page_address") 

110 if not request.HasField("location"): 

111 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_page_location") 

112 if request.location.lat == 0 and request.location.lng == 0: 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true

113 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_coordinate") 

114 

115 geom = create_coordinate(request.location.lat, request.location.lng) 

116 

117 if ( 

118 request.photo_key 

119 and not session.execute(select(Upload).where(Upload.key == request.photo_key)).scalar_one_or_none() 

120 ): 

121 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "photo_not_found") 

122 

123 parent_node = get_parent_node_at_location(session, geom) 

124 if not parent_node: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true

125 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "location_not_in_any_community") 

126 thread = Thread() 

127 session.add(thread) 

128 session.flush() 

129 page = Page( 

130 parent_node_id=parent_node.id, 

131 type=PageType.place, 

132 creator_user_id=context.user_id, 

133 owner_user_id=context.user_id, 

134 thread_id=thread.id, 

135 ) 

136 session.add(page) 

137 session.flush() 

138 page_version = PageVersion( 

139 page_id=page.id, 

140 editor_user_id=context.user_id, 

141 title=request.title, 

142 content=request.content, 

143 photo_key=request.photo_key if request.photo_key else None, 

144 address=request.address, 

145 geom=geom, 

146 ) 

147 session.add(page_version) 

148 session.commit() 

149 return page_to_pb(session, page, context) 

150 

151 def CreateGuide( 

152 self, request: pages_pb2.CreateGuideReq, context: CouchersContext, session: Session 

153 ) -> pages_pb2.Page: 

154 if not request.title: 

155 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_page_title") 

156 if not request.content: 

157 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_page_content") 

158 if request.address and request.HasField("location"): 

159 address = request.address 

160 if request.location.lat == 0 and request.location.lng == 0: 160 ↛ 161line 160 didn't jump to line 161 because the condition on line 160 was never true

161 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_coordinate") 

162 geom = create_coordinate(request.location.lat, request.location.lng) 

163 elif not request.address and not request.HasField("location"): 

164 address = None 

165 geom = None 

166 else: 

167 # you have to have both or neither 

168 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_guide_location") 

169 

170 if not request.parent_community_id: 

171 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_page_parent") 

172 

173 parent_node = session.execute(select(Node).where(Node.id == request.parent_community_id)).scalar_one_or_none() 

174 

175 if not parent_node: 

176 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "community_not_found") 

177 

178 if ( 178 ↛ 182line 178 didn't jump to line 182 because the condition on line 178 was never true

179 request.photo_key 

180 and not session.execute(select(Upload).where(Upload.key == request.photo_key)).scalar_one_or_none() 

181 ): 

182 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "photo_not_found") 

183 

184 thread = Thread() 

185 session.add(thread) 

186 session.flush() 

187 page = Page( 

188 parent_node_id=parent_node.id, 

189 type=PageType.guide, 

190 creator_user_id=context.user_id, 

191 owner_user_id=context.user_id, 

192 thread_id=thread.id, 

193 ) 

194 session.add(page) 

195 session.flush() 

196 page_version = PageVersion( 

197 page_id=page.id, 

198 editor_user_id=context.user_id, 

199 title=request.title, 

200 content=request.content, 

201 photo_key=request.photo_key if request.photo_key else None, 

202 address=address, 

203 geom=geom, 

204 ) 

205 session.add(page_version) 

206 session.commit() 

207 return page_to_pb(session, page, context) 

208 

209 def GetPage(self, request: pages_pb2.GetPageReq, context: CouchersContext, session: Session) -> pages_pb2.Page: 

210 page = session.execute( 

211 select(Page) 

212 .where(Page.id == request.page_id) 

213 .options(selectinload(Page.versions), selectinload(Page.owner_cluster)) 

214 ).scalar_one_or_none() 

215 if not page: 215 ↛ 216line 215 didn't jump to line 216 because the condition on line 215 was never true

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

217 

218 return page_to_pb(session, page, context) 

219 

220 def UpdatePage( 

221 self, request: pages_pb2.UpdatePageReq, context: CouchersContext, session: Session 

222 ) -> pages_pb2.Page: 

223 page = session.execute(select(Page).where(Page.id == request.page_id)).scalar_one_or_none() 

224 if not page: 224 ↛ 225line 224 didn't jump to line 225 because the condition on line 224 was never true

225 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "page_not_found") 

226 

227 if not _is_page_owner(page, context.user_id) and not _can_moderate_page(session, page, context.user_id): 227 ↛ 228line 227 didn't jump to line 228 because the condition on line 227 was never true

228 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "page_update_permission_denied") 

229 

230 current_version = page.versions[-1] 

231 

232 page_version = PageVersion( 

233 page_id=page.id, 

234 editor_user_id=context.user_id, 

235 title=current_version.title, 

236 content=current_version.content, 

237 photo_key=current_version.photo_key, 

238 address=current_version.address, 

239 geom=current_version.geom, 

240 ) 

241 

242 if request.HasField("title"): 

243 if not request.title.value: 

244 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_page_title") 

245 page_version.title = request.title.value 

246 

247 if request.HasField("content"): 

248 if not request.content.value: 

249 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_page_content") 

250 page_version.content = request.content.value 

251 

252 if request.HasField("photo_key"): 

253 if not request.photo_key.value: 

254 page_version.photo_key = None 

255 else: 

256 if not session.execute( 

257 select(Upload).where(Upload.key == request.photo_key.value) 

258 ).scalar_one_or_none(): 

259 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "photo_not_found") 

260 page_version.photo_key = request.photo_key.value 

261 

262 if request.HasField("address"): 

263 if not request.address.value: 

264 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_page_address") 

265 page_version.address = request.address.value 

266 

267 if request.HasField("location"): 

268 page_version.geom = create_coordinate(request.location.lat, request.location.lng) 

269 

270 session.add(page_version) 

271 session.commit() 

272 return page_to_pb(session, page, context) 

273 

274 def TransferPage( 

275 self, request: pages_pb2.TransferPageReq, context: CouchersContext, session: Session 

276 ) -> pages_pb2.Page: 

277 page = session.execute( 

278 select(Page).where(Page.id == request.page_id).where(Page.type != PageType.main_page) 

279 ).scalar_one_or_none() 

280 

281 if not page: 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, "page_not_found") 

283 

284 if not _is_page_owner(page, context.user_id) and not _can_moderate_page(session, page, context.user_id): 

285 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "page_transfer_permission_denied") 

286 

287 if request.WhichOneof("new_owner") == "new_owner_group_id": 

288 cluster = session.execute( 

289 select(Cluster).where(~Cluster.is_official_cluster).where(Cluster.id == request.new_owner_group_id) 

290 ).scalar_one_or_none() 

291 elif request.WhichOneof("new_owner") == "new_owner_community_id": 291 ↛ 299line 291 didn't jump to line 299 because the condition on line 291 was always true

292 cluster = session.execute( 

293 select(Cluster) 

294 .where(Cluster.parent_node_id == request.new_owner_community_id) 

295 .where(Cluster.is_official_cluster) 

296 ).scalar_one_or_none() 

297 else: 

298 # i'm not sure if this needs to be checked 

299 context.abort_with_error_code(grpc.StatusCode.UNKNOWN, "unknown_error") 

300 

301 if not cluster: 

302 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "group_or_community_not_found") 

303 

304 page.owner_user = None 

305 page.owner_cluster = cluster 

306 

307 session.commit() 

308 return page_to_pb(session, page, context) 

309 

310 def ListUserPlaces( 

311 self, request: pages_pb2.ListUserPlacesReq, context: CouchersContext, session: Session 

312 ) -> pages_pb2.ListUserPlacesRes: 

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

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

315 user_id = request.user_id or context.user_id 

316 places = ( 

317 session.execute( 

318 select(Page) 

319 .where(Page.owner_user_id == user_id) 

320 .where(Page.type == PageType.place) 

321 .where(Page.id >= next_page_id) 

322 .order_by(Page.id) 

323 .limit(page_size + 1) 

324 .options(selectinload(Page.versions), selectinload(Page.owner_cluster)) 

325 ) 

326 .scalars() 

327 .all() 

328 ) 

329 return pages_pb2.ListUserPlacesRes( 

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

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

332 ) 

333 

334 def ListUserGuides( 

335 self, request: pages_pb2.ListUserGuidesReq, context: CouchersContext, session: Session 

336 ) -> pages_pb2.ListUserGuidesRes: 

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

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

339 user_id = request.user_id or context.user_id 

340 guides = ( 

341 session.execute( 

342 select(Page) 

343 .where(Page.owner_user_id == user_id) 

344 .where(Page.type == PageType.guide) 

345 .where(Page.id >= next_page_id) 

346 .order_by(Page.id) 

347 .limit(page_size + 1) 

348 .options(selectinload(Page.versions), selectinload(Page.owner_cluster)) 

349 ) 

350 .scalars() 

351 .all() 

352 ) 

353 return pages_pb2.ListUserGuidesRes( 

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

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

356 )