From cccfef890e91d20c2e9fe3a10bafe6d7d045d78d Mon Sep 17 00:00:00 2001 From: Sean Hammond Date: Thu, 9 Feb 2012 12:49:27 +0100 Subject: [PATCH] [#1775,#1776] Add tag_delete() logic action function and authorization and tests. --- ckan/logic/action/delete.py | 23 ++- ckan/logic/auth/delete.py | 4 + .../functional/api/model/test_vocabulary.py | 184 +++++++++++++++++- 3 files changed, 207 insertions(+), 4 deletions(-) diff --git a/ckan/logic/action/delete.py b/ckan/logic/action/delete.py index 2542c44b579..555dd7b04ef 100644 --- a/ckan/logic/action/delete.py +++ b/ckan/logic/action/delete.py @@ -1,4 +1,4 @@ -from ckan.logic import NotFound +from ckan.logic import NotFound, ParameterError, ValidationError from ckan.lib.base import _ from ckan.logic import check_access from ckan.logic.action import rename_keys @@ -121,6 +121,27 @@ def vocabulary_delete(context, data_dict): vocab_obj.delete() model.repo.commit() +def tag_delete(context, data_dict): + model = context['model'] + user = context['user'] + + if not data_dict.has_key('tag_name'): + raise ParameterError(_("Missing 'tag_name' parameter.")) + + if not data_dict.has_key('vocabulary_name'): + raise ParameterError(_("Missing 'vocabulary_name' parameter.")) + + tag_obj = model.tag.Tag.get(data_dict['tag_name'], + data_dict['vocabulary_name']) + + if tag_obj is None: + raise NotFound + + check_access('tag_delete', context, data_dict) + + tag_obj.delete() + model.repo.commit() + def package_relationship_delete_rest(context, data_dict): # rename keys diff --git a/ckan/logic/auth/delete.py b/ckan/logic/auth/delete.py index d20a32bf73c..12937c284d9 100644 --- a/ckan/logic/auth/delete.py +++ b/ckan/logic/auth/delete.py @@ -60,3 +60,7 @@ def task_status_delete(context, data_dict): def vocabulary_delete(context, data_dict): user = context['user'] return {'success': Authorizer.is_sysadmin(user)} + +def tag_delete(context, data_dict): + user = context['user'] + return {'success': Authorizer.is_sysadmin(user)} diff --git a/ckan/tests/functional/api/model/test_vocabulary.py b/ckan/tests/functional/api/model/test_vocabulary.py index 025507ba128..8429788b93c 100644 --- a/ckan/tests/functional/api/model/test_vocabulary.py +++ b/ckan/tests/functional/api/model/test_vocabulary.py @@ -139,8 +139,10 @@ def _delete_vocabulary(self, vocab_id, user=None): status=404) assert response.json['success'] == False - def _list_tags(self, vocabulary, user=None): - params = {'vocabulary_name': vocabulary['name']} + def _list_tags(self, vocabulary=None, user=None): + params = {} + if vocabulary: + params['vocabulary_name'] = vocabulary['name'] if user: extra_environ = {'Authorization' : str(user.apikey)} else: @@ -163,6 +165,18 @@ def _create_tag(self, user, tag_name, vocabulary=None): assert response['success'] == True return response['result'] + def _delete_tag(self, user, tag, vocabulary): + params = {'tag_name': tag['name'], + 'vocabulary_name': vocabulary['name']} + if user: + extra_environ = {'Authorization' : str(user.apikey)} + else: + extra_environ = None + response = self._post('/api/action/tag_delete', params=params, + extra_environ=extra_environ) + assert response['success'] == True + return response['result'] + def test_vocabulary_create(self): '''Test adding a new vocabulary to a CKAN instance via the action API. @@ -348,7 +362,7 @@ def test_add_tag_to_vocab(self): assert tag_created['name'] in new_tag_names def test_add_tag_no_vocab(self): - '''Test the error response when a user tries to create a tab without + '''Test the error response when a user tries to create a tag without specifying a vocab. ''' @@ -475,3 +489,167 @@ def test_add_tag_not_authorized(self): str(self.normal_user.apikey)}, status=403) assert response.json['success'] == False + + def test_delete_tag_from_vocab(self): + '''Test that a tag can be deleted from a vocab.''' + + vocab = self.genre_vocab + + # First add a tag to the vocab. + tag = self._create_tag(self.sysadmin_user, 'noise', vocab) + + # Now delete the tag from the vocab. + tags_before = self._list_tags(vocab) + self._delete_tag(self.sysadmin_user, tag, vocab) + tags_after = self._list_tags(vocab) + + assert len(tags_after) == len(tags_before) - 1 + assert tag['name'] not in tags_after + difference = [tag_name for tag_name in tags_before if tag_name not in + tags_after] + assert len(difference) == 1 + assert tag['name'] in difference + + def test_delete_tag_no_name(self): + '''Test the error response when a user tries to delete a tag without + giving the tag name. + + ''' + vocab = self.genre_vocab + tag = self._create_tag(self.sysadmin_user, 'noise', vocab) + + # Now try to delete the tag from the vocab. + params = {'vocab_name': vocab['name']} + response = self.app.post('/api/action/tag_delete', + params=json.dumps(params), + extra_environ = {'Authorization': + str(self.sysadmin_user.apikey)}, + status=409) + assert response.json['success'] == False + assert (response.json['error']['message'] == + u"Parameter Error: Missing 'tag_name' parameter.") + + def test_delete_tag_no_vocab(self): + '''Test the error response when a user tries to delete a tag without + giving the vocab name. + + ''' + vocab = self.genre_vocab + tag = self._create_tag(self.sysadmin_user, 'noise', vocab) + + # Now try to delete the tag from the vocab. + params = {'tag_name': tag['name']} + response = self.app.post('/api/action/tag_delete', + params=json.dumps(params), + extra_environ = {'Authorization': + str(self.sysadmin_user.apikey)}, + status=409) + assert response.json['success'] == False + assert (response.json['error']['message'] == + u"Parameter Error: Missing 'vocabulary_name' parameter.") + + def test_delete_tag_not_exists(self): + '''Test the error response when a user tries to delete a from a vocab + but there is no tag with that name in the vocab. + + ''' + vocab = self.genre_vocab + tag = self._create_tag(self.sysadmin_user, 'noise', vocab) + + params = {'tag_name': 'nonexistent', + 'vocabulary_name': self.genre_vocab['name']} + response = self.app.post('/api/action/tag_delete', + params=json.dumps(params), + extra_environ = {'Authorization': + str(self.sysadmin_user.apikey)}, + status=404) + assert response.json['success'] == False + assert (response.json['error']['message'] == + u"Not found") + + def test_delete_tag_vocab_not_exists(self): + '''Test the error response when a user tries to delete a from a vocab + but there is no vocab with that name. + + ''' + vocab = self.genre_vocab + tag = self._create_tag(self.sysadmin_user, 'noise', vocab) + + params = {'tag_name': tag['name'], + 'vocabulary_name': 'nonexistent'} + response = self.app.post('/api/action/tag_delete', + params=json.dumps(params), + extra_environ = {'Authorization': + str(self.sysadmin_user.apikey)}, + status=404) + assert response.json['success'] == False + assert (response.json['error']['message'] == + u"Not found") + + def test_delete_tag_invalid_tag(self): + '''Test the error response when a user tries to delete a tag but gives + an invalid tag name. + + ''' + vocab = self.genre_vocab + tag = self._create_tag(self.sysadmin_user, 'noise', vocab) + + for tag_name in ('Invalid!', '', None): + params = {'tag_name': tag_name, + 'vocabulary_name': self.genre_vocab['name']} + response = self.app.post('/api/action/tag_delete', + params=json.dumps(params), + extra_environ = {'Authorization': + str(self.sysadmin_user.apikey)}, + status=404) + assert response.json['success'] == False + assert (response.json['error']['message'] == + u"Not found") + + def test_delete_tag_invalid_vocab(self): + '''Test the error response when a user tries to delete a tag but gives + an invalid vocab name. + + ''' + vocab = self.genre_vocab + tag = self._create_tag(self.sysadmin_user, 'noise', vocab) + + for vocab_name in ('Invalid!', '', None): + params = {'tag_name': tag['name'], + 'vocabulary_name': vocab_name} + response = self.app.post('/api/action/tag_delete', + params=json.dumps(params), + extra_environ = {'Authorization': + str(self.sysadmin_user.apikey)}, + status=404) + assert response.json['success'] == False + assert (response.json['error']['message'] == + u"Not found") + + def test_delete_tag_not_logged_in(self): + vocab = self.genre_vocab + tag = self._create_tag(self.sysadmin_user, 'noise', vocab) + + params = {'tag_name': tag['name'], + 'vocabulary_name': self.genre_vocab['name']} + response = self.app.post('/api/action/tag_delete', + params=json.dumps(params), + status=403) + assert response.json['success'] == False + assert (response.json['error']['message'] == + u"Access denied") + + def test_delete_tag_not_authorized(self): + vocab = self.genre_vocab + tag = self._create_tag(self.sysadmin_user, 'noise', vocab) + + params = {'tag_name': tag['name'], + 'vocabulary_name': self.genre_vocab['name']} + response = self.app.post('/api/action/tag_delete', + params=json.dumps(params), + extra_environ = {'Authorization': + str(self.normal_user.apikey)}, + status=403) + assert response.json['success'] == False + assert (response.json['error']['message'] == + u"Access denied")