Coverage for src/tests/test_blocking.py: 100%

1import grpc 

2import pytest 

3from google.protobuf import empty_pb2 

4 

5from couchers.models import User, UserBlock 

6from couchers.proto import blocking_pb2 

7from couchers.servicers.blocking import is_not_visible 

8from couchers.sql import couchers_select as select 

9from tests.test_fixtures import blocking_session, db, generate_user, make_user_block, session_scope, testconfig # noqa 

10 

11 

12@pytest.fixture(autouse=True) 

13def _(testconfig): 

14 pass 

15 

16 

17def test_BlockUser(db): 

18 user1, token1 = generate_user() 

19 user2, token2 = generate_user() 

20 

21 with session_scope() as session: 

22 blocked_user_list = ( 

23 session.execute(select(UserBlock).where(UserBlock.blocking_user_id == user1.id)).scalars().all() 

24 ) 

25 assert len(blocked_user_list) == 0 

26 

27 with blocking_session(token1) as user_blocks: 

28 with pytest.raises(grpc.RpcError) as e: 

29 user_blocks.BlockUser(blocking_pb2.BlockUserReq(username=user1.username)) 

30 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

31 assert e.value.details() == "You can't block yourself." 

32 

33 user_blocks.BlockUser(blocking_pb2.BlockUserReq(username=user2.username)) 

34 

35 with pytest.raises(grpc.RpcError) as e: 

36 user_blocks.BlockUser(blocking_pb2.BlockUserReq(username=user2.username)) 

37 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

38 assert e.value.details() == "Target user has already been blocked." 

39 

40 with session_scope() as session: 

41 blocked_user_list = ( 

42 session.execute(select(UserBlock).where(UserBlock.blocking_user_id == user1.id)).scalars().all() 

43 ) 

44 assert len(blocked_user_list) == 1 

45 

46 

47def test_make_user_block(db): 

48 user1, token1 = generate_user() 

49 user2, token2 = generate_user() 

50 

51 make_user_block(user1, user2) 

52 

53 with session_scope() as session: 

54 blocked_user_list = ( 

55 session.execute(select(UserBlock).where(UserBlock.blocking_user_id == user1.id)).scalars().all() 

56 ) 

57 assert len(blocked_user_list) == 1 

58 

59 

60def test_UnblockUser(db): 

61 user1, token1 = generate_user() 

62 user2, token2 = generate_user() 

63 make_user_block(user1, user2) 

64 

65 with blocking_session(token1) as user_blocks: 

66 user_blocks.UnblockUser(blocking_pb2.UnblockUserReq(username=user2.username)) 

67 

68 with session_scope() as session: 

69 blocked_users = session.execute(select(UserBlock).where(UserBlock.blocking_user_id == user1.id)).scalars().all() 

70 assert len(blocked_users) == 0 

71 

72 with blocking_session(token1) as user_blocks: 

73 with pytest.raises(grpc.RpcError) as e: 

74 user_blocks.UnblockUser(blocking_pb2.UnblockUserReq(username=user2.username)) 

75 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

76 assert e.value.details() == "Target user is not blocked." 

77 

78 # Test re-blocking 

79 user_blocks.BlockUser(blocking_pb2.BlockUserReq(username=user2.username)) 

80 

81 with session_scope() as session: 

82 blocked_users = session.execute(select(UserBlock).where(UserBlock.blocking_user_id == user1.id)).scalars().all() 

83 assert len(blocked_users) == 1 

84 

85 

86def test_GetBlockedUsers(db): 

87 user1, token1 = generate_user() 

88 user2, token2 = generate_user() 

89 user3, token3 = generate_user() 

90 

91 with blocking_session(token1) as user_blocks: 

92 # Check no blocked users to start 

93 blocked_user_list = user_blocks.GetBlockedUsers(empty_pb2.Empty()) 

94 assert len(blocked_user_list.blocked_users) == 0 

95 

96 make_user_block(user1, user2) 

97 make_user_block(user1, user3) 

98 blocked_user_list = user_blocks.GetBlockedUsers(empty_pb2.Empty()) 

99 assert len(blocked_user_list.blocked_users) == 2 

100 

101 blocked_usernames = [user.username for user in blocked_user_list.blocked_users] 

102 blocked_names = [user.name for user in blocked_user_list.blocked_users] 

103 

104 assert user2.username in blocked_usernames 

105 assert user3.username in blocked_usernames 

106 assert user2.name in blocked_names 

107 assert user3.name in blocked_names 

108 

109 

110def test_relationships_userblock_dot_user(db): 

111 user1, token1 = generate_user() 

112 user2, token2 = generate_user() 

113 

114 make_user_block(user1, user2) 

115 

116 with session_scope() as session: 

117 block = session.execute( 

118 select(UserBlock).where((UserBlock.blocking_user_id == user1.id) & (UserBlock.blocked_user_id == user2.id)) 

119 ).scalar_one_or_none() 

120 

121 blocking_user_username = block.blocking_user.username 

122 blocking_user_name = block.blocking_user.name 

123 

124 blocked_user_username = block.blocked_user.username 

125 blocked_user_name = block.blocked_user.name 

126 

127 assert blocking_user_username == user1.username 

128 assert blocked_user_username == user2.username 

129 assert blocking_user_name == user1.name 

130 assert blocked_user_name == user2.name 

131 

132 

133def test_is_not_visible(db): 

134 """ 

135 Comprehensive tests for is_not_visible function covering: 

136 1. Normal visible users (not blocked, not banned, not deleted) - should be visible to each other 

137 2. User1 blocks User2 - not visible 

138 3. User2 blocks User1 - not visible 

139 4. Mutual blocking - not visible 

140 5. User1 is deleted - not visible 

141 6. User2 is deleted - not visible 

142 7. Both users deleted - not visible 

143 8. User1 is banned - not visible 

144 9. User2 is banned - not visible 

145 10. Both users banned - not visible 

146 11. Mixed: User1 deleted and User2 banned - not visible 

147 12. None user_id cases with visible users 

148 13. None user_id cases with hidden users (banned/deleted) 

149 14. Ensure we have extra users in DB to avoid edge cases with empty database 

150 """ 

151 

152 # Create extra users to ensure the database is not empty - these should remain visible 

153 extra_user1, _ = generate_user() 

154 extra_user2, _ = generate_user() 

155 extra_user3, _ = generate_user() 

156 

157 # Create users for testing - all start as visible 

158 normal_user1, _ = generate_user() 

159 normal_user2, _ = generate_user() 

160 

161 # Users for blocking tests 

162 blocker_user, _ = generate_user() 

163 blockee_user, _ = generate_user() 

164 

165 # Users for reverse blocking tests 

166 reverse_blocker, _ = generate_user() 

167 reverse_blockee, _ = generate_user() 

168 

169 # Users for mutual blocking tests 

170 mutual_blocker1, _ = generate_user() 

171 mutual_blocker2, _ = generate_user() 

172 

173 # Users for deletion tests 

174 deleted_user1, _ = generate_user() 

175 visible_for_deleted, _ = generate_user() 

176 deleted_user2, _ = generate_user() 

177 both_deleted1, _ = generate_user() 

178 both_deleted2, _ = generate_user() 

179 

180 # Users for ban tests 

181 banned_user1, _ = generate_user() 

182 visible_for_banned, _ = generate_user() 

183 banned_user2, _ = generate_user() 

184 both_banned1, _ = generate_user() 

185 both_banned2, _ = generate_user() 

186 

187 # User for mixed test 

188 mixed_deleted, _ = generate_user() 

189 mixed_banned, _ = generate_user() 

190 

191 with session_scope() as session: 

192 # Test 1: Two normal visible users - should be visible to each other 

193 assert not is_not_visible(session, normal_user1.id, normal_user2.id) 

194 assert not is_not_visible(session, normal_user2.id, normal_user1.id) 

195 

196 # Test 2: User1 blocks User2 - should not be visible 

197 make_user_block(blocker_user, blockee_user) 

198 assert is_not_visible(session, blocker_user.id, blockee_user.id) 

199 assert is_not_visible(session, blockee_user.id, blocker_user.id) # symmetric 

200 

201 # Test 3: User2 blocks User1 (reverse block) - should not be visible 

202 make_user_block(reverse_blockee, reverse_blocker) 

203 assert is_not_visible(session, reverse_blocker.id, reverse_blockee.id) 

204 assert is_not_visible(session, reverse_blockee.id, reverse_blocker.id) # symmetric 

205 

206 # Test 4: Mutual blocking - should not be visible 

207 make_user_block(mutual_blocker1, mutual_blocker2) 

208 make_user_block(mutual_blocker2, mutual_blocker1) 

209 assert is_not_visible(session, mutual_blocker1.id, mutual_blocker2.id) 

210 assert is_not_visible(session, mutual_blocker2.id, mutual_blocker1.id) 

211 

212 # Test 5: User1 is deleted - should not be visible 

213 deleted_user1_db = session.get(User, deleted_user1.id) 

214 deleted_user1_db.is_deleted = True 

215 session.commit() 

216 assert is_not_visible(session, deleted_user1.id, visible_for_deleted.id) 

217 assert is_not_visible(session, visible_for_deleted.id, deleted_user1.id) 

218 

219 # Test 6: User2 is deleted - should not be visible 

220 deleted_user2_db = session.get(User, deleted_user2.id) 

221 deleted_user2_db.is_deleted = True 

222 session.commit() 

223 assert is_not_visible(session, normal_user1.id, deleted_user2.id) 

224 assert is_not_visible(session, deleted_user2.id, normal_user1.id) 

225 

226 # Test 7: Both users deleted - should not be visible 

227 both_deleted1_db = session.get(User, both_deleted1.id) 

228 both_deleted2_db = session.get(User, both_deleted2.id) 

229 both_deleted1_db.is_deleted = True 

230 both_deleted2_db.is_deleted = True 

231 session.commit() 

232 assert is_not_visible(session, both_deleted1.id, both_deleted2.id) 

233 assert is_not_visible(session, both_deleted2.id, both_deleted1.id) 

234 

235 # Test 8: User1 is banned - should not be visible 

236 banned_user1_db = session.get(User, banned_user1.id) 

237 banned_user1_db.is_banned = True 

238 session.commit() 

239 assert is_not_visible(session, banned_user1.id, visible_for_banned.id) 

240 assert is_not_visible(session, visible_for_banned.id, banned_user1.id) 

241 

242 # Test 9: User2 is banned - should not be visible 

243 banned_user2_db = session.get(User, banned_user2.id) 

244 banned_user2_db.is_banned = True 

245 session.commit() 

246 assert is_not_visible(session, normal_user2.id, banned_user2.id) 

247 assert is_not_visible(session, banned_user2.id, normal_user2.id) 

248 

249 # Test 10: Both users banned - should not be visible 

250 both_banned1_db = session.get(User, both_banned1.id) 

251 both_banned2_db = session.get(User, both_banned2.id) 

252 both_banned1_db.is_banned = True 

253 both_banned2_db.is_banned = True 

254 session.commit() 

255 assert is_not_visible(session, both_banned1.id, both_banned2.id) 

256 assert is_not_visible(session, both_banned2.id, both_banned1.id) 

257 

258 # Test 11: Mixed - one deleted, one banned - should not be visible 

259 mixed_deleted_db = session.get(User, mixed_deleted.id) 

260 mixed_banned_db = session.get(User, mixed_banned.id) 

261 mixed_deleted_db.is_deleted = True 

262 mixed_banned_db.is_banned = True 

263 session.commit() 

264 assert is_not_visible(session, mixed_deleted.id, mixed_banned.id) 

265 assert is_not_visible(session, mixed_banned.id, mixed_deleted.id) 

266 

267 # Test 12: None user_id cases with visible users - should be visible 

268 assert not is_not_visible(session, None, normal_user1.id) 

269 assert not is_not_visible(session, normal_user1.id, None) 

270 assert not is_not_visible(session, None, None) 

271 

272 # Test 13: None user_id cases with hidden users (deleted/banned) - should not be visible 

273 assert is_not_visible(session, None, deleted_user1.id) 

274 assert is_not_visible(session, deleted_user1.id, None) 

275 assert is_not_visible(session, None, banned_user1.id) 

276 assert is_not_visible(session, banned_user1.id, None) 

277 

278 # Test 14: Verify extra users are still visible (database not empty) 

279 assert not is_not_visible(session, extra_user1.id, extra_user2.id) 

280 assert not is_not_visible(session, extra_user2.id, extra_user3.id) 

281 assert not is_not_visible(session, extra_user1.id, extra_user3.id) 

282 

283 # Additional edge case: Check that normal users are still visible to each other after all the above 

284 assert not is_not_visible(session, normal_user1.id, normal_user2.id) 

285 assert not is_not_visible(session, normal_user1.id, extra_user1.id)