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

1import grpc 

2from sqlalchemy import select 

3from sqlalchemy.orm import Session 

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(select(Page).where(Page.id == request.page_id)).scalar_one_or_none() 

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

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

213 

214 return page_to_pb(session, page, context) 

215 

216 def UpdatePage( 

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

218 ) -> pages_pb2.Page: 

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

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

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

222 

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

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

225 

226 current_version = page.versions[-1] 

227 

228 page_version = PageVersion( 

229 page_id=page.id, 

230 editor_user_id=context.user_id, 

231 title=current_version.title, 

232 content=current_version.content, 

233 photo_key=current_version.photo_key, 

234 address=current_version.address, 

235 geom=current_version.geom, 

236 ) 

237 

238 if request.HasField("title"): 

239 if not request.title.value: 

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

241 page_version.title = request.title.value 

242 

243 if request.HasField("content"): 

244 if not request.content.value: 

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

246 page_version.content = request.content.value 

247 

248 if request.HasField("photo_key"): 

249 if not request.photo_key.value: 

250 page_version.photo_key = None 

251 else: 

252 if not session.execute( 

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

254 ).scalar_one_or_none(): 

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

256 page_version.photo_key = request.photo_key.value 

257 

258 if request.HasField("address"): 

259 if not request.address.value: 

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

261 page_version.address = request.address.value 

262 

263 if request.HasField("location"): 

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

265 

266 session.add(page_version) 

267 session.commit() 

268 return page_to_pb(session, page, context) 

269 

270 def TransferPage( 

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

272 ) -> pages_pb2.Page: 

273 page = session.execute( 

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

275 ).scalar_one_or_none() 

276 

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

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

279 

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

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

282 

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

284 cluster = session.execute( 

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

286 ).scalar_one_or_none() 

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

288 cluster = session.execute( 

289 select(Cluster) 

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

291 .where(Cluster.is_official_cluster) 

292 ).scalar_one_or_none() 

293 else: 

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

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

296 

297 if not cluster: 

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

299 

300 page.owner_user = None 

301 page.owner_cluster = cluster 

302 

303 session.commit() 

304 return page_to_pb(session, page, context) 

305 

306 def ListUserPlaces( 

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

308 ) -> pages_pb2.ListUserPlacesRes: 

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

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

311 user_id = request.user_id or context.user_id 

312 places = ( 

313 session.execute( 

314 select(Page) 

315 .where(Page.owner_user_id == user_id) 

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

317 .where(Page.id >= next_page_id) 

318 .order_by(Page.id) 

319 .limit(page_size + 1) 

320 ) 

321 .scalars() 

322 .all() 

323 ) 

324 return pages_pb2.ListUserPlacesRes( 

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

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

327 ) 

328 

329 def ListUserGuides( 

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

331 ) -> pages_pb2.ListUserGuidesRes: 

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

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

334 user_id = request.user_id or context.user_id 

335 guides = ( 

336 session.execute( 

337 select(Page) 

338 .where(Page.owner_user_id == user_id) 

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

340 .where(Page.id >= next_page_id) 

341 .order_by(Page.id) 

342 .limit(page_size + 1) 

343 ) 

344 .scalars() 

345 .all() 

346 ) 

347 return pages_pb2.ListUserGuidesRes( 

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

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

350 )