CKAN-Extensions
/
ckanext-privatedatasets
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Modify the way allowed users are obtained

This commit is contained in:
Aitor Magán 2014-07-16 13:29:53 +02:00
parent 07f6e1e7af
commit 5995a67576
6 changed files with 52 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
ckanext/privatedatasets/actions.py
View File

@ -64,7 +64,7 @@ def package_adquired(context, request_data):
# Some datasets does not allow to introduce the list of allowed users since this property is
# only valid for private datasets outside an organization. In this case, a wanr will return
# but the process will continue
warns.append('Dataset %s: %s' % (dataset_id, e.error_dict[constants.ALLOWED_USERS][0]))
warns.append('%s(%s): %s' % (dataset_id, constants.ALLOWED_USERS, e.error_dict[constants.ALLOWED_USERS][0]))
# Return warnings that inform about non-existing datasets
if len(warns) > 0:

30
ckanext/privatedatasets/converters_validators.py
View File

@ -25,15 +25,29 @@ def private_datasets_metadata_checker(key, data, errors, context):
def allowed_users_convert(key, data, errors, context):
if isinstance(data[key], basestring):
allowed_users = [allowed_user for allowed_user in data[key].split(',')]
# By default, all the fileds are in the data dictionary even if they contains nothing. In this case,
# the value is 'ckan.lib.navl.dictization_functions.Missing' and for this reason the type is checked
# Get the allowed user list
if (constants.ALLOWED_USERS,) in data and isinstance(data[(constants.ALLOWED_USERS,)], list):
allowed_users = data[(constants.ALLOWED_USERS,)]
elif (constants.ALLOWED_USERS_STR,) in data and isinstance(data[(constants.ALLOWED_USERS_STR,)], basestring):
allowed_users_str = data[(constants.ALLOWED_USERS_STR,)].strip()
allowed_users = [allowed_user for allowed_user in allowed_users_str.split(',') if allowed_user.strip() != '']
else:
allowed_users = data[key]
allowed_users = None
current_index = max([int(k[1]) for k in data.keys() if len(k) == 2 and k[0] == constants.ALLOWED_USERS] + [-1])
if allowed_users is not None:
current_index = max([int(k[1]) for k in data.keys() if len(k) == 2 and k[0] == key[0]] + [-1])
for num, allowed_user in zip(count(current_index + 1), allowed_users):
data[(constants.ALLOWED_USERS, num)] = allowed_user.strip()
if len(allowed_users) == 0:
data[(constants.ALLOWED_USERS,)] = []
else:
for num, allowed_user in zip(count(current_index + 1), allowed_users):
allowed_user = allowed_user.strip()
toolkit.get_validator('name_validator')(allowed_user, context) # User name should be validated
data[(key[0], num)] = allowed_user
def get_allowed_users(key, data, errors, context):
@ -47,3 +61,7 @@ def get_allowed_users(key, data, errors, context):
for user in users:
data[(key[0], counter)] = user.user_name
counter += 1
def check_user_names(key, data, errors, context):
print data[key]

9
ckanext/privatedatasets/plugin.py
View File

@ -28,9 +28,9 @@ class PrivateDatasets(p.SingletonPlugin, tk.DefaultDatasetForm):
'private': [tk.get_validator('ignore_missing'),
tk.get_validator('boolean_validator')],
constants.ALLOWED_USERS_STR: [tk.get_validator('ignore_missing'),
conv_val.allowed_users_convert,
conv_val.private_datasets_metadata_checker],
constants.ALLOWED_USERS: [tk.get_validator('ignore_missing'),
constants.ALLOWED_USERS: [conv_val.allowed_users_convert,
tk.get_validator('ignore_missing'),
conv_val.private_datasets_metadata_checker],
constants.ADQUIRE_URL: [tk.get_validator('ignore_missing'),
conv_val.private_datasets_metadata_checker,
@ -156,11 +156,6 @@ class PrivateDatasets(p.SingletonPlugin, tk.DefaultDatasetForm):
# Get the users and the package ID
if constants.ALLOWED_USERS in pkg_dict:
# When the user removes all the users using the UI, we recieve an array with one
# element that is an empty string, so set the value properly
if len(pkg_dict[constants.ALLOWED_USERS]) == 1 and pkg_dict[constants.ALLOWED_USERS][0] == '':
pkg_dict[constants.ALLOWED_USERS] = []
allowed_users = pkg_dict[constants.ALLOWED_USERS]
package_id = pkg_dict['id']

2
ckanext/privatedatasets/tests/test_actions.py
View File

@ -151,7 +151,7 @@ class ActionsTest(unittest.TestCase):
if dataset_id in datasets_not_found:
warns.append('Dataset %s was not found in this instance' % dataset_id)
elif dataset_id in not_updatable_datasets and allowed_users is not None and user_datasets['user'] not in allowed_users:
warns.append('Dataset %s: %s' % (dataset_id, ADD_USERS_ERROR))
warns.append('%s(%s): %s' % (dataset_id, 'allowed_users', ADD_USERS_ERROR))
expected_result = {'warns': warns} if len(warns) > 0 else None

25
ckanext/privatedatasets/tests/test_converters_validators.py
View File

@ -86,24 +86,41 @@ class ConvertersValidatorsTest(unittest.TestCase):
('', 2, []),
('a', 0, ['a']),
('a', 2, ['a']),
(',,, , , ', 0, []),
(',,, , , ', 2, []),
('a,z, d', 0, ['a', 'z', 'd']),
('a,z, d', 2, ['a', 'z', 'd']),
(['a','z', 'd'], 0, ['a', 'z', 'd']),
(['a','z', 'd'], 2, ['a', 'z', 'd'])
(['a','z', 'd'], 2, ['a', 'z', 'd']),
])
def test_allowed_user_convert(self, users_str, previous_users, expected_users):
def test_allowed_user_convert(self, users, previous_users, expected_users):
key_str = 'allowed_users_str'
key = 'allowed_users'
# Configure mock
name_validator = MagicMock()
conv_val.toolkit.get_validator = MagicMock(return_value=name_validator)
data = {(key_str,): users_str}
# Fullfill the data dictionary
# * list should be included in the allowed_users filed
# * strings should be included in the allowed_users_str field
if isinstance(users, basestring):
data_key = key_str
else:
data_key = key
data = {(data_key,): users}
for i in range(0, previous_users):
data[(key, i)] = i
# Call the function
conv_val.allowed_users_convert((key_str,), data, {}, {})
context = {'user': 'test', 'auth_obj_id': {'id': 1}}
conv_val.allowed_users_convert((key,), data, {}, context)
# Check that the users are set properly
for i in range(previous_users, previous_users + len(expected_users)):
name_validator.assert_any_call(expected_users[i - previous_users], context)
self.assertEquals(expected_users[i - previous_users], data[(key, i)])
@parameterized.expand([

10
ckanext/privatedatasets/tests/test_plugin.py
View File

@ -110,9 +110,9 @@ class PluginTest(unittest.TestCase):
plugin.conv_val.private_datasets_metadata_checker],
'searchable': [plugin.tk.get_validator('ignore_missing'), plugin.tk.get_validator('boolean_validator'),
plugin.tk.get_converter('convert_to_extras'), plugin.conv_val.private_datasets_metadata_checker],
'allowed_users_str': [plugin.tk.get_validator('ignore_missing'), plugin.conv_val.allowed_users_convert,
plugin.conv_val.private_datasets_metadata_checker],
'allowed_users': [plugin.tk.get_validator('ignore_missing'), plugin.conv_val.private_datasets_metadata_checker]
'allowed_users_str': [plugin.tk.get_validator('ignore_missing'), plugin.conv_val.private_datasets_metadata_checker],
'allowed_users': [plugin.conv_val.allowed_users_convert, plugin.tk.get_validator('ignore_missing'),
plugin.conv_val.private_datasets_metadata_checker]
}
self._check_fields(returned_schema, fields)
@ -279,14 +279,12 @@ class PluginTest(unittest.TestCase):
(['a'], [], ['a'], []),
(['a'], ['a'], [], []),
([], ['a'], [], ['a']),
([''], ['a'], [], ['a']),
# Two elements
(['a', 'b'], [], ['a', 'b'], []),
(['a', 'b'], ['b'], ['a'], []),
(['a'], ['a', 'b'], [], ['b']),
([], ['a', 'b'], [], ['a', 'b']),
(['a', 'b'], ['a', 'b'], [], []),
([''], ['a', 'b'], [], ['a', 'b']),
# Three or more elements
(['c'], ['a', 'b'], ['c'], ['a', 'b']),
(['a', 'b', 'c'], ['a', 'b'], ['c'], []),
@ -303,14 +301,12 @@ class PluginTest(unittest.TestCase):
(['a'], [], ['a'], []),
(['a'], ['a'], [], []),
([], ['a'], [], ['a']),
([''], ['a'], [], ['a']),
# Two elements
(['a', 'b'], [], ['a', 'b'], []),
(['a', 'b'], ['b'], ['a'], []),
(['a'], ['a', 'b'], [], ['b']),
([], ['a', 'b'], [], ['a', 'b']),
(['a', 'b'], ['a', 'b'], [], []),
([''], ['a', 'b'], [], ['a', 'b']),
# Three or more elements
(['c'], ['a', 'b'], ['c'], ['a', 'b']),
(['a', 'b', 'c'], ['a', 'b'], ['c'], []),