-
Notifications
You must be signed in to change notification settings - Fork 2k
/
test_member.py
124 lines (97 loc) · 5.29 KB
/
test_member.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
from nose.tools import assert_raises
import ckan.model as model
import ckan.logic as logic
import ckan.lib.create_test_data as create_test_data
class TestMemberLogic(object):
@classmethod
def setup_class(cls):
cls.groupname = 'david'
model.repo.new_revision()
create_test_data.CreateTestData.create()
cls.user = model.User.get('testsysadmin')
cls.pkgs = [model.Package.by_name('warandpeace'),
model.Package.by_name('annakarenina')]
@classmethod
def teardown_class(cls):
model.repo.rebuild_db()
def test_member_create(self):
self._member_create(self.pkgs[0].id, 'package', 'public')
res = self._member_list()
assert (self.pkgs[0].id, 'package', 'public') in res, res
def test_member_create_should_update_member_if_it_already_exists(self):
initial = self._member_create(self.pkgs[0].id, 'package', 'public')
final = self._member_create(self.pkgs[0].id, 'package', 'private')
assert initial['id'] == final['id'], [initial, final]
assert initial['capacity'] == u'public'
assert final['capacity'] == u'private'
def test_member_create_raises_if_user_is_unauthorized_to_update_group(self):
ctx, dd = self._build_context(self.pkgs[0].id, 'package', user='unauthorized_user')
assert_raises(logic.NotAuthorized, logic.get_action('member_create'), ctx, dd)
def test_member_create_accepts_group_name_or_id(self):
group = model.Group.get(self.groupname)
by_name = self._member_create_in_group(self.pkgs[0].id, 'package', 'public', group.name)
by_id = self._member_create_in_group(self.pkgs[0].id, 'package', 'public', group.id)
assert by_name['id'] == by_id['id']
def test_member_create_accepts_object_name_or_id(self):
test_cases = ((self.pkgs[0], 'package', 'public'),
(self.user, 'user', 'admin'))
for case in test_cases:
obj = case[0]
by_name = self._member_create(obj.name, case[1], case[2])
by_id = self._member_create(obj.id, case[1], case[2])
assert by_name['id'] == by_id['id']
def test_member_create_raises_if_some_of_the_parameters_isnt_defined(self):
ctx, dd = self._build_context(self.pkgs[0].id, 'package')
for key in dd.keys():
new_dd = dd.copy()
del new_dd[key]
assert_raises(logic.ValidationError, logic.get_action('member_create'), ctx, new_dd)
def test_member_create_raises_if_group_wasnt_found(self):
assert_raises(logic.NotFound, self._member_create_in_group, self.pkgs[0].id, 'package', 'public', 'inexistent_group')
def test_member_create_raises_if_object_wasnt_found(self):
assert_raises(logic.NotFound, self._member_create, 'inexistent_package', 'package', 'public')
def test_member_create_raises_if_object_type_is_invalid(self):
assert_raises(logic.ValidationError, self._member_create, 'obj_id', 'invalid_obj_type', 'public')
def test_member_list(self):
self._member_create(self.pkgs[0].id, 'package', 'public')
self._member_create(self.pkgs[1].id, 'package', 'public')
res = self._member_list('package')
assert (self.pkgs[0].id, 'package', 'public') in res
assert (self.pkgs[1].id, 'package', 'public') in res
res = self._member_list('user', 'admin')
assert len(res) == 0, res
assert_raises(logic.NotFound, self._member_list, 'user', 'admin', 'inexistent_group')
self._member_create(self.user.id, 'user', 'admin')
res = self._member_list('user', 'admin')
assert (self.user.id, 'user', 'Admin') in res, res
def test_member_delete(self):
self._member_create(self.user.id, 'user', 'admin')
res = self._member_list('user', 'admin')
assert (self.user.id, 'user', 'Admin') in res, res
self._member_delete(self.user.id, 'user')
res = self._member_list('user', 'admin')
assert (self.user.id, 'user', 'Admin') not in res, res
def _member_create(self, obj, obj_type, capacity):
ctx, dd = self._build_context(obj, obj_type, capacity)
return logic.get_action('member_create')(ctx, dd)
def _member_create_in_group(self, obj, obj_type, capacity, group_id):
ctx, dd = self._build_context(obj, obj_type, capacity, group_id)
return logic.get_action('member_create')(ctx, dd)
def _member_create_as_user(self, obj, obj_type, capacity, user):
ctx, dd = self._build_context(obj, obj_type, capacity, user=user)
return logic.get_action('member_create')(ctx, dd)
def _member_list(self, obj_type=None, capacity=None, group_id=None):
ctx, dd = self._build_context(None, obj_type, capacity, group_id)
return logic.get_action('member_list')(ctx, dd)
def _member_delete(self, obj, obj_type):
ctx, dd = self._build_context(obj, obj_type)
return logic.get_action('member_delete')(ctx, dd)
def _build_context(self, obj, obj_type, capacity='public', group_id=None, user=None):
ctx = {'model': model,
'session': model.Session,
'user': user or self.user.id}
dd = {'id': group_id or self.groupname,
'object': obj,
'object_type': obj_type,
'capacity': capacity}
return ctx, dd