Coverage for app / backend / src / tests / test_model_constraints.py: 97%

1import pytest 

2from sqlalchemy.exc import IntegrityError 

3from sqlalchemy.sql import func 

4 

5from couchers.db import session_scope 

6from couchers.models import ( 

7 ActivenessProbe, 

8 ActivenessProbeStatus, 

9 Cluster, 

10 FriendRelationship, 

11 FriendStatus, 

12 ModerationObjectType, 

13 ModerationState, 

14 ModerationVisibility, 

15 Node, 

16 NodeType, 

17 Page, 

18 PageType, 

19 PageVersion, 

20 Thread, 

21) 

22from couchers.utils import create_polygon_lat_lng, to_multi 

23from tests.fixtures.db import generate_user 

24from tests.test_communities import create_1d_polygon, create_community 

25 

26 

27@pytest.fixture(autouse=True) 

28def _(testconfig): 

29 pass 

30 

31 

32def test_node_constraints(db): 

33 # check we can't have two official clusters for a given node 

34 with pytest.raises(IntegrityError) as e: 

35 with session_scope() as session: 

36 node = Node(geom=to_multi(create_1d_polygon(0, 2)), node_type=NodeType.world) 

37 session.add(node) 

38 session.flush() 

39 cluster1 = Cluster( 

40 name="Testing community, cluster 1", 

41 description="Testing community description", 

42 parent_node_id=node.id, 

43 is_official_cluster=True, 

44 ) 

45 session.add(cluster1) 

46 cluster2 = Cluster( 

47 name="Testing community, cluster 2", 

48 description="Testing community description", 

49 parent_node_id=node.id, 

50 is_official_cluster=True, 

51 ) 

52 session.add(cluster2) 

53 assert "violates unique constraint" in str(e.value) 

54 assert "ix_clusters_owner_parent_node_id_is_official_cluster" in str(e.value) 

55 

56 

57def test_page_constraints(db): 

58 user, token = generate_user() 

59 

60 with session_scope() as session: 

61 c_id = create_community(session, 0, 2, "Root node", [user], [], None).id 

62 

63 # check we can't create a page without an owner 

64 with pytest.raises(IntegrityError) as e: 

65 with session_scope() as session: 

66 thread = Thread() 

67 session.add(thread) 

68 session.flush() 

69 page = Page( 

70 parent_node_id=c_id, 

71 # note no owner 

72 creator_user_id=user.id, 

73 type=PageType.guide, 

74 thread_id=thread.id, 

75 ) 

76 session.add(page) 

77 session.flush() 

78 session.add( 

79 PageVersion( 

80 page_id=page.id, 

81 editor_user_id=user.id, 

82 title="Title", 

83 content="Content", 

84 ) 

85 ) 

86 assert "violates check constraint" in str(e.value) 

87 assert "one_owner" in str(e.value) 

88 

89 with session_scope() as session: 

90 node = Node( 

91 geom=to_multi(create_polygon_lat_lng([[0, 0], [0, 2], [2, 2], [2, 0], [0, 0]])), node_type=NodeType.world 

92 ) 

93 session.add(node) 

94 session.flush() 

95 cluster = Cluster( 

96 name="Testing Community", 

97 description="Description for testing community", 

98 parent_node_id=node.id, 

99 ) 

100 session.add(cluster) 

101 session.flush() 

102 cluster_parent_id = cluster.parent_node_id 

103 cluster_id = cluster.id 

104 

105 # check we can't create a page with two owners 

106 with pytest.raises(IntegrityError) as e: 

107 with session_scope() as session: 

108 thread = Thread() 

109 session.add(thread) 

110 session.flush() 

111 page = Page( 

112 parent_node_id=cluster_parent_id, 

113 creator_user_id=user.id, 

114 owner_cluster_id=cluster_id, 

115 owner_user_id=user.id, 

116 type=PageType.guide, 

117 thread_id=thread.id, 

118 ) 

119 session.add(page) 

120 session.flush() 

121 session.add( 

122 PageVersion( 

123 page_id=page.id, 

124 editor_user_id=user.id, 

125 title="Title", 

126 content="Content", 

127 ) 

128 ) 

129 assert "violates check constraint" in str(e.value) 

130 assert "one_owner" in str(e.value) 

131 

132 # main page must be owned by the right cluster 

133 with pytest.raises(IntegrityError) as e: 

134 with session_scope() as session: 

135 thread = Thread() 

136 session.add(thread) 

137 session.flush() 

138 main_page = Page( 

139 parent_node_id=cluster_parent_id, 

140 # note owner is not cluster 

141 creator_user_id=user.id, 

142 owner_user_id=user.id, 

143 type=PageType.main_page, 

144 thread_id=thread.id, 

145 ) 

146 session.add(main_page) 

147 session.flush() 

148 session.add( 

149 PageVersion( 

150 page_id=main_page.id, 

151 editor_user_id=user.id, 

152 title="Main page for the testing community", 

153 content="Empty.", 

154 ) 

155 ) 

156 assert "violates check constraint" in str(e.value) 

157 assert "main_page_owned_by_cluster" in str(e.value) 

158 

159 # can only have one main page 

160 with pytest.raises(IntegrityError) as e: 

161 with session_scope() as session: 

162 thread1 = Thread() 

163 session.add(thread1) 

164 session.flush() 

165 main_page1 = Page( 

166 parent_node_id=cluster_parent_id, 

167 creator_user_id=user.id, 

168 owner_cluster_id=cluster_id, 

169 type=PageType.main_page, 

170 thread_id=thread1.id, 

171 ) 

172 session.add(main_page1) 

173 session.flush() 

174 session.add( 

175 PageVersion( 

176 page_id=main_page1.id, 

177 editor_user_id=user.id, 

178 title="Main page 1 for the testing community", 

179 content="Empty.", 

180 ) 

181 ) 

182 thread2 = Thread() 

183 session.add(thread2) 

184 session.flush() 

185 main_page2 = Page( 

186 parent_node_id=cluster_parent_id, 

187 creator_user_id=user.id, 

188 owner_cluster_id=cluster_id, 

189 type=PageType.main_page, 

190 thread_id=thread2.id, 

191 ) 

192 session.add(main_page2) 

193 session.flush() 

194 session.add( 

195 PageVersion( 

196 page_id=main_page2.id, 

197 editor_user_id=user.id, 

198 title="Main page 2 for the testing community", 

199 content="Empty.", 

200 ) 

201 ) 

202 assert "violates unique constraint" in str(e.value) 

203 assert "ix_pages_owner_cluster_id_type" in str(e.value) 

204 

205 

206def test_activeness_probes_cant_have_multiple(db): 

207 # can't have two active activeness probes for a given user 

208 user, token = generate_user() 

209 

210 with session_scope() as session: 

211 # we can create one 

212 first_probe = ActivenessProbe(user_id=user.id) 

213 session.add(first_probe) 

214 session.commit() 

215 

216 # change it to expired 

217 first_probe.response = ActivenessProbeStatus.expired 

218 first_probe.responded = func.now() 

219 session.commit() 

220 

221 # can create another one 

222 session.add(ActivenessProbe(user_id=user.id)) 

223 session.commit() 

224 

225 # can't create one more 

226 with pytest.raises(IntegrityError) as e: 

227 with session_scope() as session: 

228 session.add(ActivenessProbe(user_id=user.id)) 

229 assert "violates unique constraint" in str(e.value) 

230 

231 

232def _add_friend_relationship(session, from_user_id, to_user_id, status): 

233 moderation_state = ModerationState( 

234 object_type=ModerationObjectType.friend_request, 

235 object_id=0, 

236 visibility=ModerationVisibility.visible, 

237 ) 

238 session.add(moderation_state) 

239 session.flush() 

240 fr = FriendRelationship( 

241 from_user_id=from_user_id, 

242 to_user_id=to_user_id, 

243 status=status, 

244 moderation_state_id=moderation_state.id, 

245 ) 

246 session.add(fr) 

247 session.flush() 

248 moderation_state.object_id = fr.id 

249 return fr 

250 

251 

252def test_friend_relationship_unique_active_pair(db): 

253 # can't have two active (pending/accepted) FriendRelationship rows for the same user pair, 

254 # regardless of direction 

255 user1, _ = generate_user() 

256 user2, _ = generate_user() 

257 user3, _ = generate_user() 

258 

259 # baseline: one pending relationship is fine 

260 with session_scope() as session: 

261 _add_friend_relationship(session, user1.id, user2.id, FriendStatus.pending) 

262 

263 # can't add a second active row in the same direction 

264 with pytest.raises(IntegrityError) as e: 

265 with session_scope() as session: 

266 _add_friend_relationship(session, user1.id, user2.id, FriendStatus.pending) 

267 assert "violates unique constraint" in str(e.value) 

268 assert "uq_friend_relationships_active_pair" in str(e.value) 

269 

270 # can't add a second active row in the reverse direction either 

271 with pytest.raises(IntegrityError) as e: 

272 with session_scope() as session: 

273 _add_friend_relationship(session, user2.id, user1.id, FriendStatus.accepted) 

274 assert "violates unique constraint" in str(e.value) 

275 assert "uq_friend_relationships_active_pair" in str(e.value) 

276 

277 # a cancelled or rejected row for the same pair is allowed alongside the active one 

278 with session_scope() as session: 

279 _add_friend_relationship(session, user1.id, user2.id, FriendStatus.cancelled) 

280 _add_friend_relationship(session, user2.id, user1.id, FriendStatus.rejected) 

281 

282 # and active rows for different pairs are unaffected 

283 with session_scope() as session: 

284 _add_friend_relationship(session, user1.id, user3.id, FriendStatus.pending) 

285 _add_friend_relationship(session, user3.id, user2.id, FriendStatus.accepted)