Coverage for ckanext/udc/tests/test_user_actions.py: 100%

1""" 

2Unit tests for user management actions. 

3""" 

4import pytest 

5import ckan.model as model 

6import ckan.plugins.toolkit as tk 

7from ckan.tests import helpers, factories 

8 

9 

10@pytest.mark.usefixtures("clean_db", "with_plugins") 

11class TestDeletedUsersList: 

12 """Tests for the deleted_users_list action.""" 

13 

14 def test_list_deleted_users_as_sysadmin(self): 

15 """Sysadmin can list deleted users.""" 

16 # Create sysadmin 

17 sysadmin = factories.Sysadmin() 

18 

19 # Create and delete a user 

20 user = factories.User() 

21 user_obj = model.User.get(user['id']) 

22 user_obj.delete() 

23 model.Session.commit() 

24 

25 # List deleted users 

26 context = {'user': sysadmin['name'], 'ignore_auth': False} 

27 result = helpers.call_action('deleted_users_list', context=context) 

28 

29 # Verify 

30 assert isinstance(result, dict) 

31 assert result["total"] > 0 

32 assert isinstance(result["results"], list) 

33 

34 # Find our deleted user 

35 deleted_user = next((u for u in result["results"] if u['id'] == user['id']), None) 

36 assert deleted_user is not None 

37 assert deleted_user['name'] == user['name'] 

38 assert deleted_user['state'] == 'deleted' 

39 assert 'email' in deleted_user 

40 assert 'created' in deleted_user 

41 assert 'about' in deleted_user 

42 

43 def test_list_deleted_users_as_normal_user(self): 

44 """Normal users cannot list deleted users.""" 

45 # Create normal user 

46 user = factories.User() 

47 

48 # Try to list deleted users 

49 context = {'user': user['name'], 'ignore_auth': False} 

50 

51 with pytest.raises(tk.NotAuthorized): 

52 helpers.call_action('deleted_users_list', context=context) 

53 

54 def test_list_deleted_users_anonymous(self): 

55 """Anonymous users cannot list deleted users.""" 

56 context = {'user': None, 'ignore_auth': False} 

57 

58 with pytest.raises(tk.NotAuthorized): 

59 helpers.call_action('deleted_users_list', context=context) 

60 

61 def test_list_deleted_users_empty(self): 

62 """Returns empty list when no deleted users exist.""" 

63 # Create sysadmin 

64 sysadmin = factories.Sysadmin() 

65 

66 # Purge any existing deleted users first 

67 context = {'user': sysadmin['name'], 'ignore_auth': True} 

68 helpers.call_action('purge_deleted_users', context=context, ids=[]) 

69 

70 # List deleted users 

71 result = helpers.call_action('deleted_users_list', context=context) 

72 

73 assert isinstance(result, dict) 

74 assert result["total"] == 0 

75 assert result["results"] == [] 

76 

77 def test_list_multiple_deleted_users(self): 

78 """Can list multiple deleted users.""" 

79 # Create sysadmin 

80 sysadmin = factories.Sysadmin() 

81 

82 # Create and delete multiple users 

83 user1 = factories.User() 

84 user2 = factories.User() 

85 user3 = factories.User() 

86 

87 for user in [user1, user2, user3]: 

88 user_obj = model.User.get(user['id']) 

89 user_obj.delete() 

90 model.Session.commit() 

91 

92 # List deleted users 

93 context = {'user': sysadmin['name'], 'ignore_auth': False} 

94 result = helpers.call_action('deleted_users_list', context=context) 

95 

96 # Verify all three are in the list 

97 deleted_ids = [u['id'] for u in result["results"]] 

98 assert user1['id'] in deleted_ids 

99 assert user2['id'] in deleted_ids 

100 assert user3['id'] in deleted_ids 

101 

102 def test_list_does_not_include_active_users(self): 

103 """Active users are not included in deleted users list.""" 

104 # Create sysadmin and active user 

105 sysadmin = factories.Sysadmin() 

106 active_user = factories.User() 

107 

108 # List deleted users 

109 context = {'user': sysadmin['name'], 'ignore_auth': False} 

110 result = helpers.call_action('deleted_users_list', context=context) 

111 

112 # Verify active user is not in the list 

113 deleted_ids = [u['id'] for u in result["results"]] 

114 assert active_user['id'] not in deleted_ids 

115 

116 

117@pytest.mark.usefixtures("clean_db", "with_plugins") 

118class TestPurgeDeletedUsers: 

119 """Tests for the purge_deleted_users action.""" 

120 

121 def test_purge_deleted_users_as_sysadmin(self): 

122 """Sysadmin can purge deleted users.""" 

123 # Create sysadmin 

124 sysadmin = factories.Sysadmin() 

125 

126 # Create and delete a user 

127 user = factories.User() 

128 user_id = user['id'] 

129 user_obj = model.User.get(user_id) 

130 user_obj.delete() 

131 model.Session.commit() 

132 

133 # Purge deleted users 

134 context = {'user': sysadmin['name'], 'ignore_auth': False} 

135 result = helpers.call_action('purge_deleted_users', context=context, ids=[]) 

136 

137 # Verify result 

138 assert result['success'] is True 

139 assert result['count'] >= 1 

140 assert 'message' in result 

141 

142 # Verify user is completely gone 

143 purged_user = model.User.get(user_id) 

144 assert purged_user is None 

145 

146 def test_purge_deleted_users_as_normal_user(self): 

147 """Normal users cannot purge deleted users.""" 

148 # Create normal user 

149 user = factories.User() 

150 

151 # Try to purge deleted users 

152 context = {'user': user['name'], 'ignore_auth': False} 

153 

154 with pytest.raises(tk.NotAuthorized): 

155 helpers.call_action('purge_deleted_users', context=context) 

156 

157 def test_purge_deleted_users_anonymous(self): 

158 """Anonymous users cannot purge deleted users.""" 

159 context = {'user': None, 'ignore_auth': False} 

160 

161 with pytest.raises(tk.NotAuthorized): 

162 helpers.call_action('purge_deleted_users', context=context) 

163 

164 def test_purge_deleted_users_empty(self): 

165 """Purge returns zero count when no deleted users exist.""" 

166 # Create sysadmin 

167 sysadmin = factories.Sysadmin() 

168 

169 # Purge any existing deleted users first 

170 context = {'user': sysadmin['name'], 'ignore_auth': True} 

171 helpers.call_action('purge_deleted_users', context=context, ids=[]) 

172 

173 # Purge again 

174 result = helpers.call_action('purge_deleted_users', context=context, ids=[]) 

175 

176 assert result['success'] is True 

177 assert result['count'] == 0 

178 

179 def test_purge_multiple_deleted_users(self): 

180 """Can purge multiple deleted users at once.""" 

181 # Create sysadmin 

182 sysadmin = factories.Sysadmin() 

183 

184 # Create and delete multiple users 

185 user_ids = [] 

186 for i in range(3): 

187 user = factories.User() 

188 user_ids.append(user['id']) 

189 user_obj = model.User.get(user['id']) 

190 user_obj.delete() 

191 model.Session.commit() 

192 

193 # Purge deleted users 

194 context = {'user': sysadmin['name'], 'ignore_auth': False} 

195 result = helpers.call_action('purge_deleted_users', context=context, ids=user_ids) 

196 

197 # Verify result 

198 assert result['success'] is True 

199 assert result['count'] >= 3 

200 

201 # Verify all users are completely gone 

202 for user_id in user_ids: 

203 purged_user = model.User.get(user_id) 

204 assert purged_user is None 

205 

206 def test_purge_does_not_affect_active_users(self): 

207 """Active users are not affected by purge.""" 

208 # Create sysadmin and active user 

209 sysadmin = factories.Sysadmin() 

210 active_user = factories.User() 

211 

212 # Purge deleted users 

213 context = {'user': sysadmin['name'], 'ignore_auth': False} 

214 helpers.call_action('purge_deleted_users', context=context, ids=[]) 

215 

216 # Verify active user still exists 

217 user_obj = model.User.get(active_user['id']) 

218 assert user_obj is not None 

219 assert user_obj.state == 'active' 

220 

221 def test_purge_removes_user_memberships(self): 

222 """Purge removes user's group and organization memberships.""" 

223 # Create sysadmin 

224 sysadmin = factories.Sysadmin() 

225 

226 # Create user and organization 

227 user = factories.User() 

228 org = factories.Organization( 

229 users=[{'name': user['name'], 'capacity': 'member'}] 

230 ) 

231 

232 # Verify membership exists 

233 member = model.Session.query(model.Member).filter( 

234 model.Member.table_id == user['id'], 

235 model.Member.group_id == org['id'] 

236 ).first() 

237 assert member is not None 

238 

239 # Delete and purge user 

240 user_obj = model.User.get(user['id']) 

241 user_obj.delete() 

242 model.Session.commit() 

243 

244 context = {'user': sysadmin['name'], 'ignore_auth': False} 

245 helpers.call_action('purge_deleted_users', context=context, ids=[]) 

246 

247 # Verify membership is gone 

248 member = model.Session.query(model.Member).filter( 

249 model.Member.table_id == user['id'], 

250 model.Member.group_id == org['id'] 

251 ).first() 

252 assert member is None 

253 

254 def test_purge_workflow(self): 

255 """Complete workflow: list, purge, verify.""" 

256 # Create sysadmin 

257 sysadmin = factories.Sysadmin() 

258 context = {'user': sysadmin['name'], 'ignore_auth': False} 

259 

260 # Create and delete users 

261 user1 = factories.User() 

262 user2 = factories.User() 

263 

264 for user in [user1, user2]: 

265 user_obj = model.User.get(user['id']) 

266 user_obj.delete() 

267 model.Session.commit() 

268 

269 # Step 1: List deleted users 

270 deleted_list = helpers.call_action('deleted_users_list', context=context) 

271 initial_count = deleted_list["total"] 

272 assert initial_count >= 2 

273 

274 deleted_ids = [u['id'] for u in deleted_list["results"]] 

275 assert user1['id'] in deleted_ids 

276 assert user2['id'] in deleted_ids 

277 

278 # Step 2: Purge deleted users 

279 purge_result = helpers.call_action('purge_deleted_users', context=context, ids=deleted_ids) 

280 assert purge_result['success'] is True 

281 assert purge_result['count'] == initial_count 

282 

283 # Step 3: Verify list is now empty 

284 deleted_list_after = helpers.call_action('deleted_users_list', context=context) 

285 assert deleted_list_after["total"] == 0 

286 

287 # Step 4: Verify users are completely gone 

288 assert model.User.get(user1['id']) is None 

289 assert model.User.get(user2['id']) is None 

290 

291 

292@pytest.mark.usefixtures("clean_db", "with_plugins") 

293class TestUserManagementIntegration: 

294 """Integration tests for user management APIs.""" 

295 

296 def test_deleted_user_datasets_remain(self): 

297 """Datasets created by deleted users remain after purge.""" 

298 # Create sysadmin and user 

299 sysadmin = factories.Sysadmin() 

300 user = factories.User() 

301 

302 # Create dataset as user 

303 dataset = factories.Dataset(user=user) 

304 dataset_id = dataset['id'] 

305 

306 # Delete and purge user 

307 user_obj = model.User.get(user['id']) 

308 user_obj.delete() 

309 model.Session.commit() 

310 

311 context = {'user': sysadmin['name'], 'ignore_auth': False} 

312 helpers.call_action('purge_deleted_users', context=context, ids=[]) 

313 

314 # Verify dataset still exists 

315 dataset_obj = model.Package.get(dataset_id) 

316 assert dataset_obj is not None 

317 assert dataset_obj.state == 'active' 

318 

319 def test_cannot_purge_with_api_key_from_deleted_user(self): 

320 """Cannot use API key from a deleted user.""" 

321 # Create user 

322 user = factories.User() 

323 api_key = user['apikey'] 

324 

325 # Delete user 

326 user_obj = model.User.get(user['id']) 

327 user_obj.delete() 

328 model.Session.commit() 

329 

330 # Try to use deleted user's API key 

331 context = {'user': user['name'], 'ignore_auth': False} 

332 

333 # This should fail because the user is deleted 

334 with pytest.raises((tk.NotAuthorized, tk.ObjectNotFound)): 

335 helpers.call_action('purge_deleted_users', context=context, ids=[]) 

336 

337 

338@pytest.mark.usefixtures("clean_db", "with_plugins") 

339class TestUserList: 

340 """Tests for the udc_user_list action.""" 

341 

342 def test_list_users_as_sysadmin(self): 

343 sysadmin = factories.Sysadmin() 

344 user = factories.User() 

345 

346 context = {'user': sysadmin['name'], 'ignore_auth': False} 

347 result = helpers.call_action('udc_user_list', context=context, page=1, page_size=25) 

348 

349 assert isinstance(result, dict) 

350 assert result["total"] >= 1 

351 assert any(u["id"] == user["id"] for u in result["results"]) 

352 

353 def test_list_users_as_normal_user(self): 

354 user = factories.User() 

355 context = {'user': user['name'], 'ignore_auth': False} 

356 

357 with pytest.raises(tk.NotAuthorized): 

358 helpers.call_action('udc_user_list', context=context, page=1, page_size=25)