Coverage for app/backend/src/tests/test_model_constraints.py: 97%
130 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +0000
1import pytest
2from sqlalchemy.exc import IntegrityError
3from sqlalchemy.sql import func
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
27def test_node_constraints(db):
28 # check we can't have two official clusters for a given node
29 with pytest.raises(IntegrityError) as e:
30 with session_scope() as session:
31 node = Node(geom=to_multi(create_1d_polygon(0, 2)), node_type=NodeType.world)
32 session.add(node)
33 session.flush()
34 cluster1 = Cluster(
35 name="Testing community, cluster 1",
36 description="Testing community description",
37 parent_node_id=node.id,
38 is_official_cluster=True,
39 )
40 session.add(cluster1)
41 cluster2 = Cluster(
42 name="Testing community, cluster 2",
43 description="Testing community description",
44 parent_node_id=node.id,
45 is_official_cluster=True,
46 )
47 session.add(cluster2)
48 assert "violates unique constraint" in str(e.value)
49 assert "ix_clusters_owner_parent_node_id_is_official_cluster" in str(e.value)
52def test_page_constraints(db):
53 user, token = generate_user()
55 with session_scope() as session:
56 c_id = create_community(session, 0, 2, "Root node", [user], [], None).id
58 # check we can't create a page without an owner
59 with pytest.raises(IntegrityError) as e:
60 with session_scope() as session:
61 thread = Thread()
62 session.add(thread)
63 session.flush()
64 page = Page(
65 parent_node_id=c_id,
66 # note no owner
67 creator_user_id=user.id,
68 type=PageType.guide,
69 thread_id=thread.id,
70 )
71 session.add(page)
72 session.flush()
73 session.add(
74 PageVersion(
75 page_id=page.id,
76 editor_user_id=user.id,
77 title="Title",
78 content="Content",
79 )
80 )
81 assert "violates check constraint" in str(e.value)
82 assert "one_owner" in str(e.value)
84 with session_scope() as session:
85 node = Node(
86 geom=to_multi(create_polygon_lat_lng([[0, 0], [0, 2], [2, 2], [2, 0], [0, 0]])), node_type=NodeType.world
87 )
88 session.add(node)
89 session.flush()
90 cluster = Cluster(
91 name="Testing Community",
92 description="Description for testing community",
93 parent_node_id=node.id,
94 )
95 session.add(cluster)
96 session.flush()
97 cluster_parent_id = cluster.parent_node_id
98 cluster_id = cluster.id
100 # check we can't create a page with two owners
101 with pytest.raises(IntegrityError) as e:
102 with session_scope() as session:
103 thread = Thread()
104 session.add(thread)
105 session.flush()
106 page = Page(
107 parent_node_id=cluster_parent_id,
108 creator_user_id=user.id,
109 owner_cluster_id=cluster_id,
110 owner_user_id=user.id,
111 type=PageType.guide,
112 thread_id=thread.id,
113 )
114 session.add(page)
115 session.flush()
116 session.add(
117 PageVersion(
118 page_id=page.id,
119 editor_user_id=user.id,
120 title="Title",
121 content="Content",
122 )
123 )
124 assert "violates check constraint" in str(e.value)
125 assert "one_owner" in str(e.value)
127 # main page must be owned by the right cluster
128 with pytest.raises(IntegrityError) as e:
129 with session_scope() as session:
130 thread = Thread()
131 session.add(thread)
132 session.flush()
133 main_page = Page(
134 parent_node_id=cluster_parent_id,
135 # note owner is not cluster
136 creator_user_id=user.id,
137 owner_user_id=user.id,
138 type=PageType.main_page,
139 thread_id=thread.id,
140 )
141 session.add(main_page)
142 session.flush()
143 session.add(
144 PageVersion(
145 page_id=main_page.id,
146 editor_user_id=user.id,
147 title="Main page for the testing community",
148 content="Empty.",
149 )
150 )
151 assert "violates check constraint" in str(e.value)
152 assert "main_page_owned_by_cluster" in str(e.value)
154 # can only have one main page
155 with pytest.raises(IntegrityError) as e:
156 with session_scope() as session:
157 thread1 = Thread()
158 session.add(thread1)
159 session.flush()
160 main_page1 = Page(
161 parent_node_id=cluster_parent_id,
162 creator_user_id=user.id,
163 owner_cluster_id=cluster_id,
164 type=PageType.main_page,
165 thread_id=thread1.id,
166 )
167 session.add(main_page1)
168 session.flush()
169 session.add(
170 PageVersion(
171 page_id=main_page1.id,
172 editor_user_id=user.id,
173 title="Main page 1 for the testing community",
174 content="Empty.",
175 )
176 )
177 thread2 = Thread()
178 session.add(thread2)
179 session.flush()
180 main_page2 = Page(
181 parent_node_id=cluster_parent_id,
182 creator_user_id=user.id,
183 owner_cluster_id=cluster_id,
184 type=PageType.main_page,
185 thread_id=thread2.id,
186 )
187 session.add(main_page2)
188 session.flush()
189 session.add(
190 PageVersion(
191 page_id=main_page2.id,
192 editor_user_id=user.id,
193 title="Main page 2 for the testing community",
194 content="Empty.",
195 )
196 )
197 assert "violates unique constraint" in str(e.value)
198 assert "ix_pages_owner_cluster_id_type" in str(e.value)
201def test_activeness_probes_cant_have_multiple(db):
202 # can't have two active activeness probes for a given user
203 user, token = generate_user()
205 with session_scope() as session:
206 # we can create one
207 first_probe = ActivenessProbe(user_id=user.id)
208 session.add(first_probe)
209 session.commit()
211 # change it to expired
212 first_probe.response = ActivenessProbeStatus.expired
213 first_probe.responded = func.now()
214 session.commit()
216 # can create another one
217 session.add(ActivenessProbe(user_id=user.id))
218 session.commit()
220 # can't create one more
221 with pytest.raises(IntegrityError) as e:
222 with session_scope() as session:
223 session.add(ActivenessProbe(user_id=user.id))
224 assert "violates unique constraint" in str(e.value)
227def _add_friend_relationship(session, from_user_id, to_user_id, status):
228 moderation_state = ModerationState(
229 object_type=ModerationObjectType.friend_request,
230 object_id=0,
231 visibility=ModerationVisibility.visible,
232 )
233 session.add(moderation_state)
234 session.flush()
235 fr = FriendRelationship(
236 from_user_id=from_user_id,
237 to_user_id=to_user_id,
238 status=status,
239 moderation_state_id=moderation_state.id,
240 )
241 session.add(fr)
242 session.flush()
243 moderation_state.object_id = fr.id
244 return fr
247def test_friend_relationship_unique_active_pair(db):
248 # can't have two active (pending/accepted) FriendRelationship rows for the same user pair,
249 # regardless of direction
250 user1, _ = generate_user()
251 user2, _ = generate_user()
252 user3, _ = generate_user()
254 # baseline: one pending relationship is fine
255 with session_scope() as session:
256 _add_friend_relationship(session, user1.id, user2.id, FriendStatus.pending)
258 # can't add a second active row in the same direction
259 with pytest.raises(IntegrityError) as e:
260 with session_scope() as session:
261 _add_friend_relationship(session, user1.id, user2.id, FriendStatus.pending)
262 assert "violates unique constraint" in str(e.value)
263 assert "uq_friend_relationships_active_pair" in str(e.value)
265 # can't add a second active row in the reverse direction either
266 with pytest.raises(IntegrityError) as e:
267 with session_scope() as session:
268 _add_friend_relationship(session, user2.id, user1.id, FriendStatus.accepted)
269 assert "violates unique constraint" in str(e.value)
270 assert "uq_friend_relationships_active_pair" in str(e.value)
272 # a cancelled or rejected row for the same pair is allowed alongside the active one
273 with session_scope() as session:
274 _add_friend_relationship(session, user1.id, user2.id, FriendStatus.cancelled)
275 _add_friend_relationship(session, user2.id, user1.id, FriendStatus.rejected)
277 # and active rows for different pairs are unaffected
278 with session_scope() as session:
279 _add_friend_relationship(session, user1.id, user3.id, FriendStatus.pending)
280 _add_friend_relationship(session, user3.id, user2.id, FriendStatus.accepted)