test_vocabulary.py

import ckan
from pylons.test import pylonsapp
import paste.fixture
from ckan.lib.helpers import json
import sqlalchemy
from nose.tools import raises, assert_raises

class TestVocabulary(object):

    def setup(self):
        self.app = paste.fixture.TestApp(pylonsapp)
        ckan.tests.CreateTestData.create()
        self.sysadmin_user = ckan.model.User.get('testsysadmin')
        self.normal_user = ckan.model.User.get('annafan')
        # Make a couple of test vocabularies needed later.
        self.genre_vocab = self._create_vocabulary(vocab_name="Genre",
                user=self.sysadmin_user)
        self.timeperiod_vocab = self._create_vocabulary(
                vocab_name="Time Period", user=self.sysadmin_user)
        self.composers_vocab = self._create_vocabulary(vocab_name="Composers",
                user=self.sysadmin_user)

    def teardown(self):
        ckan.model.repo.rebuild_db()

    def _post(self, url, params=None, extra_environ=None):
        if params is None:
            params = {}
        param_string = json.dumps(params)
        response = self.app.post(url, params=param_string,
                extra_environ=extra_environ)
        assert not response.errors
        return response.json

    def _create_vocabulary(self, vocab_name=None, user=None):
        # Create a new vocabulary.
        params = {'name': vocab_name}
        if user:
            extra_environ = {'Authorization' : str(user.apikey)}
        else:
            extra_environ = None
        response = self._post('/api/action/vocabulary_create', params=params,
                extra_environ=extra_environ)

        # Check the values of the response.
        assert response['success'] == True
        assert response['result']
        created_vocab = response['result']
        assert created_vocab['name'] == vocab_name
        assert created_vocab['id']

        # Get the list of vocabularies.
        response = self._post('/api/action/vocabulary_list')
        # Check that the vocabulary we created is in the list.
        assert response['success'] == True
        assert response['result']
        assert response['result'].count(created_vocab) == 1

        # Get the created vocabulary.
        params = {'id': created_vocab['id']}
        response = self._post('/api/action/vocabulary_show', params)
        # Check that retrieving the vocab by name gives the same result.
        by_name_params = {'name': created_vocab['name']}
        assert response == self._post('/api/action/vocabulary_show',
                by_name_params)
        # Check that it matches what we created.
        assert response['success'] == True
        assert response['result'] == created_vocab

        return created_vocab

    def _update_vocabulary(self, params, user=None):
        if user:
            extra_environ = {'Authorization' : str(user.apikey)}
        else:
            extra_environ = None
        response = self._post('/api/action/vocabulary_update', params=params,
                extra_environ=extra_environ)

        # Check the values of the response.
        assert response['success'] == True
        assert response['result']
        updated_vocab = response['result']
        if params.has_key('id'):
            assert updated_vocab['id'] == params['id']
        else:
            assert updated_vocab['id']
        if params.has_key('name'):
            assert updated_vocab['name'] == params['name']
        else:
            assert updated_vocab['name']

        # Get the list of vocabularies.
        response = self._post('/api/action/vocabulary_list')
        # Check that the vocabulary we created is in the list.
        assert response['success'] == True
        assert response['result']
        assert response['result'].count(updated_vocab) == 1

        # Get the created vocabulary.
        params = {'id': updated_vocab['id']}
        response = self._post('/api/action/vocabulary_show', params)
        # Check that retrieving the vocab by name gives the same result.
        by_name_params = {'name': updated_vocab['name']}
        assert response == self._post('/api/action/vocabulary_show',
                by_name_params)
        # Check that it matches what we created.
        assert response['success'] == True
        assert response['result'] == updated_vocab

        return updated_vocab

    def _delete_vocabulary(self, vocab_id, user=None):
        if user:
            extra_environ = {'Authorization' : str(user.apikey)}
        else:
            extra_environ = None
        params = {'id': vocab_id}
        response = self._post('/api/action/vocabulary_delete', params=params,
                extra_environ=extra_environ)

        # Check the values of the response.
        assert response['success'] == True
        assert response['result'] is None
        response['result']

        # Get the list of vocabularies.
        response = self._post('/api/action/vocabulary_list')
        assert response['success'] == True
        assert response['result']
        # Check that the vocabulary we deleted is not in the list.
        assert vocab_id not in [vocab['id'] for vocab in response['result']]

        # Check that the deleted vocabulary can no longer be retrieved.
        response = self.app.post('/api/action/vocabulary_show',
                params=json.dumps(params),
                extra_environ = {'Authorization':
                    str(self.sysadmin_user.apikey)},
                status=404)
        assert response.json['success'] == False

    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:
            extra_environ = None
        response = self._post('/api/action/tag_list', params=params,
                extra_environ=extra_environ)
        assert response['success'] == True
        return response['result']

    def _create_tag(self, user, tag_name, vocabulary=None):
        tag_dict = {'name': tag_name}
        if vocabulary:
            tag_dict['vocabulary_id'] = vocabulary['id']
        if user:
            extra_environ = {'Authorization' : str(user.apikey)}
        else:
            extra_environ = None
        response = self._post('/api/action/tag_create', params=tag_dict,
                extra_environ=extra_environ)
        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.

        '''
        self._create_vocabulary(vocab_name="My cool vocab",
                user=self.sysadmin_user)

    def test_vocabulary_create_id(self):
        '''Test error response when user tries to supply their own ID when
        creating a vocabulary.

        '''
        params = {'id': 'xxx', 'name': 'foobar'}
        param_string = json.dumps(params)
        response = self.app.post('/api/action/vocabulary_create',
                params=param_string,
                extra_environ = {'Authorization':
                    str(self.sysadmin_user.apikey)},
                status=409)
        assert response.json['success'] == False

    def test_vocabulary_create_no_name(self):
        '''Test error response when user tries to create a vocab without a
        name.

        '''
        params = {}
        param_string = json.dumps(params)
        response = self.app.post('/api/action/vocabulary_create',
                params=param_string,
                extra_environ = {'Authorization':
                    str(self.sysadmin_user.apikey)},
                status=409)
        assert response.json['success'] == False

    def test_vocabulary_create_invalid_name(self):
        '''Test error response when user tries to create a vocab with an
        invalid name.

        '''
        for name in (None, '', 'a', 'foobar'*100):
            params = {'name': name}
            param_string = json.dumps(params)
            response = self.app.post('/api/action/vocabulary_create',
                    params=param_string,
                    extra_environ = {'Authorization':
                        str(self.sysadmin_user.apikey)},
                    status=409)
            assert response.json['success'] == False

    def test_vocabulary_create_exists(self):
        '''Test error response when user tries to create a vocab that already
        exists.

        '''
        params = {'name': self.genre_vocab['name']}
        param_string = json.dumps(params)
        response = self.app.post('/api/action/vocabulary_create',
                params=param_string,
                extra_environ = {'Authorization':
                    str(self.sysadmin_user.apikey)},
                status=409)
        assert response.json['success'] == False

    def test_vocabulary_create_not_logged_in(self):
        '''Test that users who are not logged in cannot create vocabularies.'''

        params = {'name':
            "Spam Vocabulary: SpamCo Duck Rental: Rent Your Ducks From Us!"}
        param_string = json.dumps(params)
        response = self.app.post('/api/action/vocabulary_create',
                params=param_string,
                status=403)
        assert response.json['success'] == False

    def test_vocabulary_create_not_authorized(self):
        '''Test that users who are not authorized cannot create vocabs.'''

        params = {'name': 'My Unauthorised Vocabulary'}
        param_string = json.dumps(params)
        response = self.app.post('/api/action/vocabulary_create',
                params=param_string,
                extra_environ = {'Authorization':
                    str(self.normal_user.apikey)},
                status=403)
        assert response.json['success'] == False

    def test_vocabulary_update(self):
        self._update_vocabulary({'id': self.genre_vocab['id'],
            'name': 'updated_name'}, self.sysadmin_user)

    def test_vocabulary_update_not_exists(self):
        '''Test the error response given when a user tries to update a
        vocabulary that doesn't exist.

        '''
        params = {'id': 'xxxxxxx', 'name': 'updated_name'}
        param_string = json.dumps(params)
        response = self.app.post('/api/action/vocabulary_update',
                params=param_string,
                extra_environ = {'Authorization':
                    str(self.sysadmin_user.apikey)},
                status=404)
        assert response.json['success'] == False

    def test_vocabulary_update_no_name(self):
        params = {'id': self.genre_vocab['id']}
        param_string = json.dumps(params)
        response = self.app.post('/api/action/vocabulary_update',
                params=param_string,
                extra_environ = {'Authorization':
                    str(self.sysadmin_user.apikey)},
                status=409)
        assert response.json['success'] == False

    def test_vocabulary_update_not_logged_in(self):
        '''Test that users who are not logged in cannot update vocabularies.'''
        params = {'id': self.genre_vocab['id']}
        param_string = json.dumps(params)
        response = self.app.post('/api/action/vocabulary_update',
                params=param_string,
                status=403)
        assert response.json['success'] == False

    def test_vocabulary_update_not_authorized(self):
        '''Test that users who are not authorized cannot update vocabs.'''
        params = {'id': self.genre_vocab['id']}
        param_string = json.dumps(params)
        response = self.app.post('/api/action/vocabulary_update',
                params=param_string,
                extra_environ = {'Authorization':
                    str(self.normal_user.apikey)},
                status=403)
        assert response.json['success'] == False

    def test_vocabulary_delete(self):
        self._delete_vocabulary(self.genre_vocab['id'], self.sysadmin_user)

    def test_vocabulary_delete_not_exists(self):
        '''Test the error response given when a user tries to delete a
        vocabulary that doesn't exist.

        '''
        params = {'id': 'xxxxxxx'}
        param_string = json.dumps(params)
        response = self.app.post('/api/action/vocabulary_delete',
                params=param_string,
                extra_environ = {'Authorization':
                    str(self.sysadmin_user.apikey)},
                status=404)
        assert response.json['success'] == False

    def test_vocabulary_delete_not_logged_in(self):
        '''Test that users who are not logged in cannot delete vocabularies.'''
        params = {'id': self.genre_vocab['id']}
        param_string = json.dumps(params)
        response = self.app.post('/api/action/vocabulary_delete',
                params=param_string,
                status=403)
        assert response.json['success'] == False

    def test_vocabulary_delete_not_authorized(self):
        '''Test that users who are not authorized cannot delete vocabs.'''
        params = {'id': self.genre_vocab['id']}
        param_string = json.dumps(params)
        response = self.app.post('/api/action/vocabulary_delete',
                params=param_string,
                extra_environ = {'Authorization':
                    str(self.normal_user.apikey)},
                status=403)
        assert response.json['success'] == False

    def test_add_tag_to_vocab(self):
        '''Test that a tag can be added to and then retrieved from a vocab.'''
        vocab = self.genre_vocab
        tags_before = self._list_tags(vocab)
        tag_created = self._create_tag(self.sysadmin_user, 'noise', vocab)
        tags_after = self._list_tags(vocab)
        new_tag_names = [tag_name for tag_name in tags_after if tag_name not in 
                tags_before]
        assert len(new_tag_names) == 1
        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 tag without
        specifying a vocab.

        '''
        tag_dict = {'name': 'noise'}
        tag_string = json.dumps(tag_dict)
        response = self.app.post('/api/action/tag_create',
                params=tag_string,
                extra_environ = {'Authorization':
                    str(self.sysadmin_user.apikey)},
                status=409)
        assert response.json['success'] == False

    def test_add_tag_vocab_not_exists(self):
        '''Test the error response when a user tries to add a tag to a vocab
        that doesn't exist.

        '''
        tag_dict = {'name': 'noise', 'vocabulary_id': 'does not exist'}
        tag_string = json.dumps(tag_dict)
        response = self.app.post('/api/action/tag_create',
                params=tag_string,
                extra_environ = {'Authorization':
                    str(self.sysadmin_user.apikey)},
                status=409)
        assert response.json['success'] == False

    def test_add_tag_already_added(self):
        '''Test the error response when a user tries to add a tag to a vocab 
        that already has a tag with the same name.

        '''
        self.test_add_tag_to_vocab()
        vocab = self.genre_vocab
        tag_dict = {'name': 'noise', 'vocabulary_id': vocab['id']}
        tag_string = json.dumps(tag_dict)
        response = self.app.post('/api/action/tag_create',
                params=tag_string,
                extra_environ = {'Authorization':
                    str(self.sysadmin_user.apikey)},
                status=409)
        assert response.json['success'] == False

    def test_add_tag_with_id(self):
        '''Test the error response when a user tries to specify the tag ID when
        adding a tag to a vocab.

        '''
        tag_dict = {
                'id': 'dsagdsgsgsd',
                'name': 'noise',
                'vocabulary_id': self.genre_vocab['id']
                }
        tag_string = json.dumps(tag_dict)
        response = self.app.post('/api/action/tag_create',
                params=tag_string,
                extra_environ = {'Authorization':
                    str(self.sysadmin_user.apikey)},
                status=409)
        assert response.json['success'] == False

    def test_add_tag_without_name(self):
        '''Test the error response when a user tries to create a tag without a
        name.

        '''
        tag_dict = {
                'vocabulary_id': self.genre_vocab['id']
                }
        tag_string = json.dumps(tag_dict)
        response = self.app.post('/api/action/tag_create',
                params=tag_string,
                extra_environ = {'Authorization':
                    str(self.sysadmin_user.apikey)},
                status=409)
        assert response.json['success'] == False

    def test_add_tag_invalid_name(self):
        for name in ('Not a valid tag name!', '', None):
            tag_dict = {
                    'name': name,
                    'vocabulary_id': self.genre_vocab['id']
                    }
            tag_string = json.dumps(tag_dict)
            response = self.app.post('/api/action/tag_create',
                    params=tag_string,
                    extra_environ = {'Authorization':
                        str(self.sysadmin_user.apikey)},
                    status=409)
            assert response.json['success'] == False

    def test_add_tag_invalid_vocab_id(self):
        tag_dict = {
                'name': 'noise',
                'vocabulary_id': 'xxcxzczxczxc',
                }
        tag_string = json.dumps(tag_dict)
        response = self.app.post('/api/action/tag_create',
                params=tag_string,
                extra_environ = {'Authorization':
                    str(self.sysadmin_user.apikey)},
                status=409)
        assert response.json['success'] == False

    def test_add_tag_not_logged_in(self):
        tag_dict = {
                'name': 'noise',
                'vocabulary_id': self.genre_vocab['id']
                }
        tag_string = json.dumps(tag_dict)
        response = self.app.post('/api/action/tag_create',
                params=tag_string,
                status=403)
        assert response.json['success'] == False

    def test_add_tag_not_authorized(self):
        tag_dict = {
                'name': 'noise',
                'vocabulary_id': self.genre_vocab['id']
                }
        tag_string = json.dumps(tag_dict)
        response = self.app.post('/api/action/tag_create',
                params=tag_string,
                    extra_environ = {'Authorization':
                        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")