Mercurial > kallithea

comparison rhodecode/tests/api/api_base.py @ 2526:473794943022 beta

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Refactored API - added ~100% line coverage tests for api - test against hg and git
author Marcin Kuzminski <marcin@python-works.com>
date Sun, 01 Jul 2012 16:22:38 +0200
parents
children 40b3a54391f9
comparison
equal deleted inserted replaced
2525:c35980ae7958 2526:473794943022
1 import random
2 import mock
3
4 from rhodecode.tests import *
5 from rhodecode.model.meta import Session
6 from rhodecode.lib.compat import json
7 from rhodecode.lib.auth import AuthUser
8 from rhodecode.model.user import UserModel
9 from rhodecode.model.users_group import UsersGroupModel
10 from rhodecode.model.repo import RepoModel
11 from nose.tools import with_setup
12
13 API_URL = '/_admin/api'
14
15
16 def _build_data(apikey, method, **kw):
17 """
18 Builds API data with given random ID
19
20 :param random_id:
21 :type random_id:
22 """
23 random_id = random.randrange(1, 9999)
24 return random_id, json.dumps({
25 "id": random_id,
26 "api_key": apikey,
27 "method": method,
28 "args": kw
29 })
30
31 jsonify = lambda obj: json.loads(json.dumps(obj))
32
33
34 def crash(*args, **kwargs):
35 raise Exception('Total Crash !')
36
37
38 TEST_USERS_GROUP = 'test_users_group'
39
40
41 def make_users_group(name=TEST_USERS_GROUP):
42 gr = UsersGroupModel().create(name=name)
43 UsersGroupModel().add_user_to_group(users_group=gr,
44 user=TEST_USER_ADMIN_LOGIN)
45 Session().commit()
46 return gr
47
48
49 def destroy_users_group(name=TEST_USERS_GROUP):
50 UsersGroupModel().delete(users_group=name, force=True)
51 Session().commit()
52
53
54 def create_repo(repo_name, repo_type):
55 # create new repo
56 form_data = dict(repo_name=repo_name,
57 repo_name_full=repo_name,
58 fork_name=None,
59 description='description %s' % repo_name,
60 repo_group=None,
61 private=False,
62 repo_type=repo_type,
63 clone_uri=None,
64 landing_rev='tip')
65 cur_user = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
66 r = RepoModel().create(form_data, cur_user)
67 Session().commit()
68 return r
69
70
71 def destroy_repo(repo_name):
72 RepoModel().delete(repo_name)
73 Session().commit()
74
75
76 class BaseTestApi(object):
77 REPO = None
78 REPO_TYPE = None
79
80 @classmethod
81 def setUpClass(self):
82 self.usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
83 self.apikey = self.usr.api_key
84 self.TEST_USER = UserModel().create_or_update(
85 username='test-api',
86 password='test',
87 email='test@api.rhodecode.org',
88 firstname='first',
89 lastname='last'
90 )
91 Session().commit()
92 self.TEST_USER_LOGIN = self.TEST_USER.username
93
94 @classmethod
95 def teardownClass(self):
96 pass
97
98 def setUp(self):
99 self.maxDiff = None
100 make_users_group()
101
102 def tearDown(self):
103 destroy_users_group()
104
105 def _compare_ok(self, id_, expected, given):
106 expected = jsonify({
107 'id': id_,
108 'error': None,
109 'result': expected
110 })
111 given = json.loads(given)
112 self.assertEqual(expected, given)
113
114 def _compare_error(self, id_, expected, given):
115 expected = jsonify({
116 'id': id_,
117 'error': expected,
118 'result': None
119 })
120 given = json.loads(given)
121 self.assertEqual(expected, given)
122
123 # def test_Optional(self):
124 # from rhodecode.controllers.api.api import Optional
125 # option1 = Optional(None)
126 # self.assertEqual('<Optional:%s>' % None, repr(option1))
127 #
128 # self.assertEqual(1, Optional.extract(Optional(1)))
129 # self.assertEqual('trololo', Optional.extract('trololo'))
130
131 def test_api_wrong_key(self):
132 id_, params = _build_data('trololo', 'get_user')
133 response = self.app.post(API_URL, content_type='application/json',
134 params=params)
135
136 expected = 'Invalid API KEY'
137 self._compare_error(id_, expected, given=response.body)
138
139 def test_api_missing_non_optional_param(self):
140 id_, params = _build_data(self.apikey, 'get_user')
141 response = self.app.post(API_URL, content_type='application/json',
142 params=params)
143
144 expected = 'Missing non optional `userid` arg in JSON DATA'
145 self._compare_error(id_, expected, given=response.body)
146
147 def test_api_get_users(self):
148 id_, params = _build_data(self.apikey, 'get_users',)
149 response = self.app.post(API_URL, content_type='application/json',
150 params=params)
151 ret_all = []
152 for usr in UserModel().get_all():
153 ret = usr.get_api_data()
154 ret_all.append(jsonify(ret))
155 expected = ret_all
156 self._compare_ok(id_, expected, given=response.body)
157
158 def test_api_get_user(self):
159 id_, params = _build_data(self.apikey, 'get_user',
160 userid=TEST_USER_ADMIN_LOGIN)
161 response = self.app.post(API_URL, content_type='application/json',
162 params=params)
163
164 usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
165 ret = usr.get_api_data()
166 ret['permissions'] = AuthUser(usr.user_id).permissions
167
168 expected = ret
169 self._compare_ok(id_, expected, given=response.body)
170
171 def test_api_get_user_that_does_not_exist(self):
172 id_, params = _build_data(self.apikey, 'get_user',
173 userid='trololo')
174 response = self.app.post(API_URL, content_type='application/json',
175 params=params)
176
177 expected = "user `%s` does not exist" % 'trololo'
178 self._compare_error(id_, expected, given=response.body)
179
180 def test_api_pull(self):
181 #TODO: issues with rhodecode_extras here.. not sure why !
182 pass
183
184 # repo_name = 'test_pull'
185 # r = create_repo(repo_name, self.REPO_TYPE)
186 # r.clone_uri = TEST_self.REPO
187 # Session.add(r)
188 # Session.commit()
189 #
190 # id_, params = _build_data(self.apikey, 'pull',
191 # repoid=repo_name,)
192 # response = self.app.post(API_URL, content_type='application/json',
193 # params=params)
194 #
195 # expected = 'Pulled from `%s`' % repo_name
196 # self._compare_ok(id_, expected, given=response.body)
197 #
198 # destroy_repo(repo_name)
199
200 def test_api_pull_error(self):
201 id_, params = _build_data(self.apikey, 'pull',
202 repoid=self.REPO,)
203 response = self.app.post(API_URL, content_type='application/json',
204 params=params)
205
206 expected = 'Unable to pull changes from `%s`' % self.REPO
207 self._compare_error(id_, expected, given=response.body)
208
209 def test_api_create_existing_user(self):
210 id_, params = _build_data(self.apikey, 'create_user',
211 username=TEST_USER_ADMIN_LOGIN,
212 email='test@foo.com',
213 password='trololo')
214 response = self.app.post(API_URL, content_type='application/json',
215 params=params)
216
217 expected = "user `%s` already exist" % TEST_USER_ADMIN_LOGIN
218 self._compare_error(id_, expected, given=response.body)
219
220 def test_api_create_user_with_existing_email(self):
221 id_, params = _build_data(self.apikey, 'create_user',
222 username=TEST_USER_ADMIN_LOGIN + 'new',
223 email=TEST_USER_REGULAR_EMAIL,
224 password='trololo')
225 response = self.app.post(API_URL, content_type='application/json',
226 params=params)
227
228 expected = "email `%s` already exist" % TEST_USER_REGULAR_EMAIL
229 self._compare_error(id_, expected, given=response.body)
230
231 def test_api_create_user(self):
232 username = 'test_new_api_user'
233 email = username + "@foo.com"
234
235 id_, params = _build_data(self.apikey, 'create_user',
236 username=username,
237 email=email,
238 password='trololo')
239 response = self.app.post(API_URL, content_type='application/json',
240 params=params)
241
242 usr = UserModel().get_by_username(username)
243 ret = dict(
244 msg='created new user `%s`' % username,
245 user=jsonify(usr.get_api_data())
246 )
247
248 expected = ret
249 self._compare_ok(id_, expected, given=response.body)
250
251 UserModel().delete(usr.user_id)
252 Session().commit()
253
254 @mock.patch.object(UserModel, 'create_or_update', crash)
255 def test_api_create_user_when_exception_happened(self):
256
257 username = 'test_new_api_user'
258 email = username + "@foo.com"
259
260 id_, params = _build_data(self.apikey, 'create_user',
261 username=username,
262 email=email,
263 password='trololo')
264 response = self.app.post(API_URL, content_type='application/json',
265 params=params)
266 expected = 'failed to create user `%s`' % username
267 self._compare_error(id_, expected, given=response.body)
268
269 def test_api_delete_user(self):
270 usr = UserModel().create_or_update(username=u'test_user',
271 password=u'qweqwe',
272 email=u'u232@rhodecode.org',
273 firstname=u'u1', lastname=u'u1')
274 Session().commit()
275 username = usr.username
276 email = usr.email
277 usr_id = usr.user_id
278 ## DELETE THIS USER NOW
279
280 id_, params = _build_data(self.apikey, 'delete_user',
281 userid=username,)
282 response = self.app.post(API_URL, content_type='application/json',
283 params=params)
284
285 ret = {'msg': 'deleted user ID:%s %s' % (usr_id, username),
286 'user': None}
287 expected = ret
288 self._compare_ok(id_, expected, given=response.body)
289
290 @mock.patch.object(UserModel, 'delete', crash)
291 def test_api_delete_user_when_exception_happened(self):
292 usr = UserModel().create_or_update(username=u'test_user',
293 password=u'qweqwe',
294 email=u'u232@rhodecode.org',
295 firstname=u'u1', lastname=u'u1')
296 Session().commit()
297 username = usr.username
298
299 id_, params = _build_data(self.apikey, 'delete_user',
300 userid=username,)
301 response = self.app.post(API_URL, content_type='application/json',
302 params=params)
303 ret = 'failed to delete ID:%s %s' % (usr.user_id,
304 usr.username)
305 expected = ret
306 self._compare_error(id_, expected, given=response.body)
307
308 @parameterized.expand([('firstname', 'new_username'),
309 ('lastname', 'new_username'),
310 ('email', 'new_username'),
311 ('admin', True),
312 ('admin', False),
313 ('ldap_dn', 'test'),
314 ('ldap_dn', None),
315 ('active', False),
316 ('active', True),
317 ('password', 'newpass')
318 ])
319 def test_api_update_user(self, name, expected):
320 usr = UserModel().get_by_username(self.TEST_USER_LOGIN)
321 kw = {name: expected,
322 'userid': usr.user_id}
323 id_, params = _build_data(self.apikey, 'update_user', **kw)
324 response = self.app.post(API_URL, content_type='application/json',
325 params=params)
326
327 ret = {
328 'msg': 'updated user ID:%s %s' % (usr.user_id, self.TEST_USER_LOGIN),
329 'user': jsonify(UserModel()\
330 .get_by_username(self.TEST_USER_LOGIN)\
331 .get_api_data())
332 }
333
334 expected = ret
335 self._compare_ok(id_, expected, given=response.body)
336
337 def test_api_update_user_no_changed_params(self):
338 usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
339 ret = jsonify(usr.get_api_data())
340 id_, params = _build_data(self.apikey, 'update_user',
341 userid=TEST_USER_ADMIN_LOGIN)
342
343 response = self.app.post(API_URL, content_type='application/json',
344 params=params)
345 ret = {
346 'msg': 'updated user ID:%s %s' % (usr.user_id, TEST_USER_ADMIN_LOGIN),
347 'user': ret
348 }
349 expected = ret
350 self._compare_ok(id_, expected, given=response.body)
351
352 def test_api_update_user_by_user_id(self):
353 usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
354 ret = jsonify(usr.get_api_data())
355 id_, params = _build_data(self.apikey, 'update_user',
356 userid=usr.user_id)
357
358 response = self.app.post(API_URL, content_type='application/json',
359 params=params)
360 ret = {
361 'msg': 'updated user ID:%s %s' % (usr.user_id, TEST_USER_ADMIN_LOGIN),
362 'user': ret
363 }
364 expected = ret
365 self._compare_ok(id_, expected, given=response.body)
366
367 @mock.patch.object(UserModel, 'create_or_update', crash)
368 def test_api_update_user_when_exception_happens(self):
369 usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
370 ret = jsonify(usr.get_api_data())
371 id_, params = _build_data(self.apikey, 'update_user',
372 userid=usr.user_id)
373
374 response = self.app.post(API_URL, content_type='application/json',
375 params=params)
376 ret = 'failed to update user `%s`' % usr.user_id
377
378 expected = ret
379 self._compare_error(id_, expected, given=response.body)
380
381 def test_api_get_repo(self):
382 new_group = 'some_new_group'
383 make_users_group(new_group)
384 RepoModel().grant_users_group_permission(repo=self.REPO,
385 group_name=new_group,
386 perm='repository.read')
387 Session().commit()
388 id_, params = _build_data(self.apikey, 'get_repo',
389 repoid=self.REPO)
390 response = self.app.post(API_URL, content_type='application/json',
391 params=params)
392
393 repo = RepoModel().get_by_repo_name(self.REPO)
394 ret = repo.get_api_data()
395
396 members = []
397 for user in repo.repo_to_perm:
398 perm = user.permission.permission_name
399 user = user.user
400 user_data = user.get_api_data()
401 user_data['type'] = "user"
402 user_data['permission'] = perm
403 members.append(user_data)
404
405 for users_group in repo.users_group_to_perm:
406 perm = users_group.permission.permission_name
407 users_group = users_group.users_group
408 users_group_data = users_group.get_api_data()
409 users_group_data['type'] = "users_group"
410 users_group_data['permission'] = perm
411 members.append(users_group_data)
412
413 ret['members'] = members
414
415 expected = ret
416 self._compare_ok(id_, expected, given=response.body)
417 destroy_users_group(new_group)
418
419 def test_api_get_repo_that_doesn_not_exist(self):
420 id_, params = _build_data(self.apikey, 'get_repo',
421 repoid='no-such-repo')
422 response = self.app.post(API_URL, content_type='application/json',
423 params=params)
424
425 ret = 'repository `%s` does not exist' % 'no-such-repo'
426 expected = ret
427 self._compare_error(id_, expected, given=response.body)
428
429 def test_api_get_repos(self):
430 id_, params = _build_data(self.apikey, 'get_repos')
431 response = self.app.post(API_URL, content_type='application/json',
432 params=params)
433
434 result = []
435 for repo in RepoModel().get_all():
436 result.append(repo.get_api_data())
437 ret = jsonify(result)
438
439 expected = ret
440 self._compare_ok(id_, expected, given=response.body)
441
442 @parameterized.expand([('all', 'all'),
443 ('dirs', 'dirs'),
444 ('files', 'files'), ])
445 def test_api_get_repo_nodes(self, name, ret_type):
446 rev = 'tip'
447 path = '/'
448 id_, params = _build_data(self.apikey, 'get_repo_nodes',
449 repoid=self.REPO, revision=rev,
450 root_path=path,
451 ret_type=ret_type)
452 response = self.app.post(API_URL, content_type='application/json',
453 params=params)
454
455 # we don't the actual return types here since it's tested somewhere
456 # else
457 expected = json.loads(response.body)['result']
458 self._compare_ok(id_, expected, given=response.body)
459
460 def test_api_get_repo_nodes_bad_revisions(self):
461 rev = 'i-dont-exist'
462 path = '/'
463 id_, params = _build_data(self.apikey, 'get_repo_nodes',
464 repoid=self.REPO, revision=rev,
465 root_path=path,)
466 response = self.app.post(API_URL, content_type='application/json',
467 params=params)
468
469 expected = 'failed to get repo: `%s` nodes' % self.REPO
470 self._compare_error(id_, expected, given=response.body)
471
472 def test_api_get_repo_nodes_bad_path(self):
473 rev = 'tip'
474 path = '/idontexits'
475 id_, params = _build_data(self.apikey, 'get_repo_nodes',
476 repoid=self.REPO, revision=rev,
477 root_path=path,)
478 response = self.app.post(API_URL, content_type='application/json',
479 params=params)
480
481 expected = 'failed to get repo: `%s` nodes' % self.REPO
482 self._compare_error(id_, expected, given=response.body)
483
484 def test_api_get_repo_nodes_bad_ret_type(self):
485 rev = 'tip'
486 path = '/'
487 ret_type = 'error'
488 id_, params = _build_data(self.apikey, 'get_repo_nodes',
489 repoid=self.REPO, revision=rev,
490 root_path=path,
491 ret_type=ret_type)
492 response = self.app.post(API_URL, content_type='application/json',
493 params=params)
494
495 expected = 'ret_type must be one of %s' % (['files', 'dirs', 'all'])
496 self._compare_error(id_, expected, given=response.body)
497
498 def test_api_create_repo(self):
499 repo_name = 'api-repo'
500 id_, params = _build_data(self.apikey, 'create_repo',
501 repo_name=repo_name,
502 owner=TEST_USER_ADMIN_LOGIN,
503 repo_type='hg',
504 )
505 response = self.app.post(API_URL, content_type='application/json',
506 params=params)
507
508 repo = RepoModel().get_by_repo_name(repo_name)
509 ret = {
510 'msg': 'Created new repository `%s`' % repo_name,
511 'repo': jsonify(repo.get_api_data())
512 }
513 expected = ret
514 self._compare_ok(id_, expected, given=response.body)
515 destroy_repo(repo_name)
516
517 def test_api_create_repo_unknown_owner(self):
518 repo_name = 'api-repo'
519 owner = 'i-dont-exist'
520 id_, params = _build_data(self.apikey, 'create_repo',
521 repo_name=repo_name,
522 owner=owner,
523 repo_type='hg',
524 )
525 response = self.app.post(API_URL, content_type='application/json',
526 params=params)
527 expected = 'user `%s` does not exist' % owner
528 self._compare_error(id_, expected, given=response.body)
529
530 def test_api_create_repo_exists(self):
531 repo_name = self.REPO
532 id_, params = _build_data(self.apikey, 'create_repo',
533 repo_name=repo_name,
534 owner=TEST_USER_ADMIN_LOGIN,
535 repo_type='hg',
536 )
537 response = self.app.post(API_URL, content_type='application/json',
538 params=params)
539 expected = "repo `%s` already exist" % repo_name
540 self._compare_error(id_, expected, given=response.body)
541
542 @mock.patch.object(RepoModel, 'create_repo', crash)
543 def test_api_create_repo_exception_occurred(self):
544 repo_name = 'api-repo'
545 id_, params = _build_data(self.apikey, 'create_repo',
546 repo_name=repo_name,
547 owner=TEST_USER_ADMIN_LOGIN,
548 repo_type='hg',
549 )
550 response = self.app.post(API_URL, content_type='application/json',
551 params=params)
552 expected = 'failed to create repository `%s`' % repo_name
553 self._compare_error(id_, expected, given=response.body)
554
555 def test_api_delete_repo(self):
556 repo_name = 'api_delete_me'
557 create_repo(repo_name, self.REPO_TYPE)
558
559 id_, params = _build_data(self.apikey, 'delete_repo',
560 repoid=repo_name,)
561 response = self.app.post(API_URL, content_type='application/json',
562 params=params)
563
564 ret = {
565 'msg': 'Deleted repository `%s`' % repo_name,
566 'success': True
567 }
568 expected = ret
569 self._compare_ok(id_, expected, given=response.body)
570
571 def test_api_delete_repo_exception_occurred(self):
572 repo_name = 'api_delete_me'
573 create_repo(repo_name, self.REPO_TYPE)
574 with mock.patch.object(RepoModel, 'delete', crash):
575 id_, params = _build_data(self.apikey, 'delete_repo',
576 repoid=repo_name,)
577 response = self.app.post(API_URL, content_type='application/json',
578 params=params)
579
580 expected = 'failed to delete repository `%s`' % repo_name
581 self._compare_error(id_, expected, given=response.body)
582 destroy_repo(repo_name)
583
584 def test_api_fork_repo(self):
585 self.failIf(False, 'TODO:')
586
587 def test_api_get_users_group(self):
588 id_, params = _build_data(self.apikey, 'get_users_group',
589 usersgroupid=TEST_USERS_GROUP)
590 response = self.app.post(API_URL, content_type='application/json',
591 params=params)
592
593 users_group = UsersGroupModel().get_group(TEST_USERS_GROUP)
594 members = []
595 for user in users_group.members:
596 user = user.user
597 members.append(user.get_api_data())
598
599 ret = users_group.get_api_data()
600 ret['members'] = members
601 expected = ret
602 self._compare_ok(id_, expected, given=response.body)
603
604 def test_api_get_users_groups(self):
605
606 make_users_group('test_users_group2')
607
608 id_, params = _build_data(self.apikey, 'get_users_groups',)
609 response = self.app.post(API_URL, content_type='application/json',
610 params=params)
611
612 expected = []
613 for gr_name in [TEST_USERS_GROUP, 'test_users_group2']:
614 users_group = UsersGroupModel().get_group(gr_name)
615 ret = users_group.get_api_data()
616 expected.append(ret)
617 self._compare_ok(id_, expected, given=response.body)
618
619 UsersGroupModel().delete(users_group='test_users_group2')
620 Session().commit()
621
622 def test_api_create_users_group(self):
623 group_name = 'some_new_group'
624 id_, params = _build_data(self.apikey, 'create_users_group',
625 group_name=group_name)
626 response = self.app.post(API_URL, content_type='application/json',
627 params=params)
628
629 ret = {
630 'msg': 'created new users group `%s`' % group_name,
631 'users_group': jsonify(UsersGroupModel()\
632 .get_by_name(group_name)\
633 .get_api_data())
634 }
635 expected = ret
636 self._compare_ok(id_, expected, given=response.body)
637
638 destroy_users_group(group_name)
639
640 def test_api_get_users_group_that_exist(self):
641 id_, params = _build_data(self.apikey, 'create_users_group',
642 group_name=TEST_USERS_GROUP)
643 response = self.app.post(API_URL, content_type='application/json',
644 params=params)
645
646 expected = "users group `%s` already exist" % TEST_USERS_GROUP
647 self._compare_error(id_, expected, given=response.body)
648
649 @mock.patch.object(UsersGroupModel, 'create', crash)
650 def test_api_get_users_group_exception_occurred(self):
651 group_name = 'exception_happens'
652 id_, params = _build_data(self.apikey, 'create_users_group',
653 group_name=group_name)
654 response = self.app.post(API_URL, content_type='application/json',
655 params=params)
656
657 expected = 'failed to create group `%s`' % group_name
658 self._compare_error(id_, expected, given=response.body)
659
660 def test_api_add_user_to_users_group(self):
661 gr_name = 'test_group'
662 UsersGroupModel().create(gr_name)
663 Session().commit()
664 id_, params = _build_data(self.apikey, 'add_user_to_users_group',
665 usersgroupid=gr_name,
666 userid=TEST_USER_ADMIN_LOGIN)
667 response = self.app.post(API_URL, content_type='application/json',
668 params=params)
669
670 expected = {
671 'msg': 'added member `%s` to users group `%s`' % (
672 TEST_USER_ADMIN_LOGIN, gr_name
673 ),
674 'success': True}
675 self._compare_ok(id_, expected, given=response.body)
676
677 UsersGroupModel().delete(users_group=gr_name)
678 Session().commit()
679
680 def test_api_add_user_to_users_group_that_doesnt_exist(self):
681 id_, params = _build_data(self.apikey, 'add_user_to_users_group',
682 usersgroupid='false-group',
683 userid=TEST_USER_ADMIN_LOGIN)
684 response = self.app.post(API_URL, content_type='application/json',
685 params=params)
686
687 expected = 'users group `%s` does not exist' % 'false-group'
688 self._compare_error(id_, expected, given=response.body)
689
690 @mock.patch.object(UsersGroupModel, 'add_user_to_group', crash)
691 def test_api_add_user_to_users_group_exception_occurred(self):
692 gr_name = 'test_group'
693 UsersGroupModel().create(gr_name)
694 Session().commit()
695 id_, params = _build_data(self.apikey, 'add_user_to_users_group',
696 usersgroupid=gr_name,
697 userid=TEST_USER_ADMIN_LOGIN)
698 response = self.app.post(API_URL, content_type='application/json',
699 params=params)
700
701 expected = 'failed to add member to users group `%s`' % gr_name
702 self._compare_error(id_, expected, given=response.body)
703
704 UsersGroupModel().delete(users_group=gr_name)
705 Session().commit()
706
707 def test_api_remove_user_from_users_group(self):
708 gr_name = 'test_group_3'
709 gr = UsersGroupModel().create(gr_name)
710 UsersGroupModel().add_user_to_group(gr, user=TEST_USER_ADMIN_LOGIN)
711 Session().commit()
712 id_, params = _build_data(self.apikey, 'remove_user_from_users_group',
713 usersgroupid=gr_name,
714 userid=TEST_USER_ADMIN_LOGIN)
715 response = self.app.post(API_URL, content_type='application/json',
716 params=params)
717
718 expected = {
719 'msg': 'removed member `%s` from users group `%s`' % (
720 TEST_USER_ADMIN_LOGIN, gr_name
721 ),
722 'success': True}
723 self._compare_ok(id_, expected, given=response.body)
724
725 UsersGroupModel().delete(users_group=gr_name)
726 Session().commit()
727
728 @mock.patch.object(UsersGroupModel, 'remove_user_from_group', crash)
729 def test_api_remove_user_from_users_group_exception_occurred(self):
730 gr_name = 'test_group_3'
731 gr = UsersGroupModel().create(gr_name)
732 UsersGroupModel().add_user_to_group(gr, user=TEST_USER_ADMIN_LOGIN)
733 Session().commit()
734 id_, params = _build_data(self.apikey, 'remove_user_from_users_group',
735 usersgroupid=gr_name,
736 userid=TEST_USER_ADMIN_LOGIN)
737 response = self.app.post(API_URL, content_type='application/json',
738 params=params)
739
740 expected = 'failed to remove member from users group `%s`' % gr_name
741 self._compare_error(id_, expected, given=response.body)
742
743 UsersGroupModel().delete(users_group=gr_name)
744 Session().commit()
745
746 @parameterized.expand([('none', 'repository.none'),
747 ('read', 'repository.read'),
748 ('write', 'repository.write'),
749 ('admin', 'repository.admin')])
750 def test_api_grant_user_permission(self, name, perm):
751 id_, params = _build_data(self.apikey, 'grant_user_permission',
752 repoid=self.REPO,
753 userid=TEST_USER_ADMIN_LOGIN,
754 perm=perm)
755 response = self.app.post(API_URL, content_type='application/json',
756 params=params)
757
758 ret = {
759 'msg': 'Granted perm: `%s` for user: `%s` in repo: `%s`' % (
760 perm, TEST_USER_ADMIN_LOGIN, self.REPO
761 ),
762 'success': True
763 }
764 expected = ret
765 self._compare_ok(id_, expected, given=response.body)
766
767 def test_api_grant_user_permission_wrong_permission(self):
768 perm = 'haha.no.permission'
769 id_, params = _build_data(self.apikey, 'grant_user_permission',
770 repoid=self.REPO,
771 userid=TEST_USER_ADMIN_LOGIN,
772 perm=perm)
773 response = self.app.post(API_URL, content_type='application/json',
774 params=params)
775
776 expected = 'permission `%s` does not exist' % perm
777 self._compare_error(id_, expected, given=response.body)
778
779 @mock.patch.object(RepoModel, 'grant_user_permission', crash)
780 def test_api_grant_user_permission_exception_when_adding(self):
781 perm = 'repository.read'
782 id_, params = _build_data(self.apikey, 'grant_user_permission',
783 repoid=self.REPO,
784 userid=TEST_USER_ADMIN_LOGIN,
785 perm=perm)
786 response = self.app.post(API_URL, content_type='application/json',
787 params=params)
788
789 expected = 'failed to edit permission for user: `%s` in repo: `%s`' % (
790 TEST_USER_ADMIN_LOGIN, self.REPO
791 )
792 self._compare_error(id_, expected, given=response.body)
793
794 def test_api_revoke_user_permission(self):
795 id_, params = _build_data(self.apikey, 'revoke_user_permission',
796 repoid=self.REPO,
797 userid=TEST_USER_ADMIN_LOGIN,)
798 response = self.app.post(API_URL, content_type='application/json',
799 params=params)
800
801 expected = {
802 'msg': 'Revoked perm for user: `%s` in repo: `%s`' % (
803 TEST_USER_ADMIN_LOGIN, self.REPO
804 ),
805 'success': True
806 }
807 self._compare_ok(id_, expected, given=response.body)
808
809 @mock.patch.object(RepoModel, 'revoke_user_permission', crash)
810 def test_api_revoke_user_permission_exception_when_adding(self):
811 id_, params = _build_data(self.apikey, 'revoke_user_permission',
812 repoid=self.REPO,
813 userid=TEST_USER_ADMIN_LOGIN,)
814 response = self.app.post(API_URL, content_type='application/json',
815 params=params)
816
817 expected = 'failed to edit permission for user: `%s` in repo: `%s`' % (
818 TEST_USER_ADMIN_LOGIN, self.REPO
819 )
820 self._compare_error(id_, expected, given=response.body)
821
822 @parameterized.expand([('none', 'repository.none'),
823 ('read', 'repository.read'),
824 ('write', 'repository.write'),
825 ('admin', 'repository.admin')])
826 def test_api_grant_users_group_permission(self, name, perm):
827 id_, params = _build_data(self.apikey, 'grant_users_group_permission',
828 repoid=self.REPO,
829 usersgroupid=TEST_USERS_GROUP,
830 perm=perm)
831 response = self.app.post(API_URL, content_type='application/json',
832 params=params)
833
834 ret = {
835 'msg': 'Granted perm: `%s` for users group: `%s` in repo: `%s`' % (
836 perm, TEST_USERS_GROUP, self.REPO
837 ),
838 'success': True
839 }
840 expected = ret
841 self._compare_ok(id_, expected, given=response.body)
842
843 def test_api_grant_users_group_permission_wrong_permission(self):
844 perm = 'haha.no.permission'
845 id_, params = _build_data(self.apikey, 'grant_users_group_permission',
846 repoid=self.REPO,
847 usersgroupid=TEST_USERS_GROUP,
848 perm=perm)
849 response = self.app.post(API_URL, content_type='application/json',
850 params=params)
851
852 expected = 'permission `%s` does not exist' % perm
853 self._compare_error(id_, expected, given=response.body)
854
855 @mock.patch.object(RepoModel, 'grant_users_group_permission', crash)
856 def test_api_grant_users_group_permission_exception_when_adding(self):
857 perm = 'repository.read'
858 id_, params = _build_data(self.apikey, 'grant_users_group_permission',
859 repoid=self.REPO,
860 usersgroupid=TEST_USERS_GROUP,
861 perm=perm)
862 response = self.app.post(API_URL, content_type='application/json',
863 params=params)
864
865 expected = 'failed to edit permission for users group: `%s` in repo: `%s`' % (
866 TEST_USERS_GROUP, self.REPO
867 )
868 self._compare_error(id_, expected, given=response.body)
869
870 def test_api_revoke_users_group_permission(self):
871 RepoModel().grant_users_group_permission(repo=self.REPO,
872 group_name=TEST_USERS_GROUP,
873 perm='repository.read')
874 Session().commit()
875 id_, params = _build_data(self.apikey, 'revoke_users_group_permission',
876 repoid=self.REPO,
877 usersgroupid=TEST_USERS_GROUP,)
878 response = self.app.post(API_URL, content_type='application/json',
879 params=params)
880
881 expected = {
882 'msg': 'Revoked perm for users group: `%s` in repo: `%s`' % (
883 TEST_USERS_GROUP, self.REPO
884 ),
885 'success': True
886 }
887 self._compare_ok(id_, expected, given=response.body)
888
889 @mock.patch.object(RepoModel, 'revoke_users_group_permission', crash)
890 def test_api_revoke_users_group_permission_exception_when_adding(self):
891
892 id_, params = _build_data(self.apikey, 'revoke_users_group_permission',
893 repoid=self.REPO,
894 usersgroupid=TEST_USERS_GROUP,)
895 response = self.app.post(API_URL, content_type='application/json',
896 params=params)
897
898 expected = 'failed to edit permission for users group: `%s` in repo: `%s`' % (
899 TEST_USERS_GROUP, self.REPO
900 )
901 self._compare_error(id_, expected, given=response.body)
902