Synnefo

# Copyright 2011 GRNET S.A. All rights reserved.

#

# Redistribution and use in source and binary forms, with or

# without modification, are permitted provided that the following

# conditions are met:

#

#   1. Redistributions of source code must retain the above

#      copyright notice, this list of conditions and the following

#      disclaimer.

#

#   2. Redistributions in binary form must reproduce the above

#      copyright notice, this list of conditions and the following

#      disclaimer in the documentation and/or other materials

#      provided with the distribution.

#

# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

# POSSIBILITY OF SUCH DAMAGE.

#

# The views and conclusions contained in the software and

# documentation are those of the authors and should not be

# interpreted as representing official policies, either expressed

# or implied, of GRNET S.A.

#

from django.test import TransactionTestCase

from django.conf import settings

from django.test.client import Client

from django.core.urlresolvers import clear_url_caches

from django.utils import simplejson as json

from django.conf import settings

from django.core.urlresolvers import reverse

from synnefo.userdata.models import *

class AaiClient(Client):

    def request(self, **request):

        request['HTTP_X_AUTH_TOKEN'] = '0000'

        return super(AaiClient, self).request(**request)

class TestRestViews(TransactionTestCase):

    fixtures = ['users']

    def setUp(self):

        def get_user_mock(request, *Args, **kwargs):

            if request.META.get('HTTP_X_AUTH_TOKEN', None) == '0000':

                request.user_uniq = 'test'

                request.user = {'uniq': 'test'}

        # mock the astakos authentication function

        from snf_django.lib import astakos

        astakos.get_user = get_user_mock

        settings.SKIP_SSH_VALIDATION = True

        self.client = AaiClient()

        self.user = 'test'

        self.keys_url = reverse('ui_keys_collection')

    def test_keys_collection_get(self):

        resp = self.client.get(self.keys_url)

        self.assertEqual(resp.content, "[]")

        PublicKeyPair.objects.create(user=self.user, name="key pair 1",

                content="content1")

        resp = self.client.get(self.keys_url)

        resp_list = json.loads(resp.content);

        exp_list = [{"content": "content1", "id": 1,

                    "uri": self.keys_url + "/1", "name": "key pair 1",

                    "fingerprint": "unknown fingerprint"}]

        self.assertEqual(resp_list, exp_list)

        PublicKeyPair.objects.create(user=self.user, name="key pair 2",

                content="content2")

        resp = self.client.get(self.keys_url)

        resp_list = json.loads(resp.content)

        exp_list = [{"content": "content1", "id": 1,

                     "uri": self.keys_url + "/1", "name": "key pair 1",

                     "fingerprint": "unknown fingerprint"},

                    {"content": "content2", "id": 2,

                     "uri": self.keys_url + "/2",

                     "name": "key pair 2",

                     "fingerprint": "unknown fingerprint"}]

        self.assertEqual(resp_list, exp_list)

    def test_keys_resourse_get(self):

        resp = self.client.get(self.keys_url + "/1")

        self.assertEqual(resp.status_code, 404)

        # create a public key

        PublicKeyPair.objects.create(user=self.user, name="key pair 1",

                content="content1")

        resp = self.client.get(self.keys_url + "/1")

        resp_dict = json.loads(resp.content);

        exp_dict = {"content": "content1", "id": 1,

                    "uri": self.keys_url + "/1", "name": "key pair 1",

                    "fingerprint": "unknown fingerprint"}

        self.assertEqual(resp_dict, exp_dict)

        # update

        resp = self.client.put(self.keys_url + "/1",

                               json.dumps({'name':'key pair 1 new name'}),

                               content_type='application/json')

        pk = PublicKeyPair.objects.get(pk=1)

        self.assertEqual(pk.name, "key pair 1 new name")

        # delete

        resp = self.client.delete(self.keys_url + "/1")

        self.assertEqual(PublicKeyPair.objects.count(), 0)

        resp = self.client.get(self.keys_url + "/1")

        self.assertEqual(resp.status_code, 404)

        resp = self.client.get(self.keys_url)

        self.assertEqual(resp.content, "[]")

        # test rest create

        resp = self.client.post(self.keys_url,

                                json.dumps({'name':'key pair 2',

                                            'content':"""key 2 content"""}),

                                content_type='application/json')

        self.assertEqual(PublicKeyPair.objects.count(), 1)

        pk = PublicKeyPair.objects.get(pk=1)

        self.assertEqual(pk.name, "key pair 2")

        self.assertEqual(pk.content, "key 2 content")

    def test_generate_views(self):

        import base64

        # just test that

        resp = self.client.post(self.keys_url + "/generate")

        self.assertNotEqual(resp, "")

        data = json.loads(resp.content)

        self.assertEqual(data.has_key('private'), True)

        self.assertEqual(data.has_key('private'), True)

        # public key is base64 encoded

        base64.b64decode(data['public'].replace("ssh-rsa ",""))

        # remove header/footer

        private = "".join(data['private'].split("\n")[1:-1])

        # private key is base64 encoded

        base64.b64decode(private)

        new_key = PublicKeyPair()

        new_key.content = data['public']

        new_key.name = "new key"

        new_key.user = 'test'

        new_key.full_clean()

        new_key.save()

    def test_invalid_data(self):

        resp = self.client.post(self.keys_url,

                                json.dumps({'content':"""key 2 content"""}),

                                content_type='application/json')

        self.assertEqual(resp.status_code, 500)

        self.assertEqual(resp.content, """{"non_field_key": "__all__", "errors": """

                                       """{"name": ["This field cannot be blank."]}}""")

        settings.USERDATA_MAX_SSH_KEYS_PER_USER = 2

        # test ssh limit

        resp = self.client.post(self.keys_url,

                                json.dumps({'name':'key1',

                                            'content':"""key 1 content"""}),

                                content_type='application/json')

        resp = self.client.post(self.keys_url,

                                json.dumps({'name':'key1',

                                            'content':"""key 1 content"""}),

                                content_type='application/json')

        resp = self.client.post(self.keys_url,

                                json.dumps({'name':'key1',

                                            'content':"""key 1 content"""}),

                                content_type='application/json')

        self.assertEqual(resp.status_code, 500)

        self.assertEqual(resp.content, """{"non_field_key": "__all__", "errors": """

                                       """{"__all__": ["SSH keys limit exceeded."]}}""")