Synnefo » ./kamaki

    $ python test.py Client request

.. code-block:: console

    # Test kamaki.clients.pithos.PithosClient.download

    $ python test.py Pithos download

Mechanism

^^^^^^^^^

Each folder / package contains a test.py file, that represents the test module of this package. All test modules contain a set of classes that extent the TestCase class. They also contain a main method to run the tests.

Adding unit tests

^^^^^^^^^^^^^^^^^

Modifying an existing method

""""""""""""""""""""""""""""

Example 1: to modify the kamaki.clients.utils.filter_in method, the programmer has to also adjust the kamaki.clients.utils.test.Utils.test_filter_in method.

Adding a new method

"""""""""""""""""""

.. code-block:: python

        ...

        def test_list_special(self):

            """Test the list_special method"""

.. code-block:: python

    class Utils(TestCase):

        ...

        def test_get_random_int(self):

Each class or module needs a seperate test sub-module. By convention, each class or module under the kamaki.clients should be located in a separate directory.

* create a new_service package and implement the unit tests in the kamaki.clients.new_service.test module:

.. code-block:: python

    from unittest import TestCase

        def test_method1(self):

            ...

.. code-block:: python

    from kamaki.clients import Client

    class NewService(Client):

..code-block:: python

    from kamaki.clients.new_service.test import NewService

# or implied, of GRNET S.A.

from mock import patch, call

from unittest import TestCase

example = dict(

astakos_pkg = 'kamaki.clients.astakos.AstakosClient'

    cached = False

    def setUp(self):

        self.url = 'https://astakos.example.com'

        self.token = 'ast@k0sT0k3n=='

    def tearDown(self):

        FR.json = example

if __name__ == '__main__':

    from sys import argv

    from kamaki.clients.test import runTestCase

from itertools import product

from json import dumps

rest_pkg = 'kamaki.clients.compute.rest_api.ComputeRestClient'

        pass

    """Set up a ComputesRest thorough test"""

    def setUp(self):

        self.url = 'http://cyclades.example.com'

        self.token = 'cyc14d3s70k3n'

    def tearDown(self):

        FR.json = vm_recv

        self._test_put('images')

    def assert_dicts_are_equal(self, d1, d2):

        for k, v in d1.items():

    def setUp(self):

        self.url = 'http://cyclades.example.com'

        self.token = 'cyc14d3s70k3n'

    def tearDown(self):

        FR.status_code = 200

        return_value=img_recv['image'])

    def test_create_server(self, GID):

        with patch.object(

                side_effect=ClientError(

                    'REQUEST ENTITY TOO LARGE',

                    status=403)):

                vm_name, fid, img_ref)

        with patch.object(

                return_value=FR()) as post:

            r = self.client.create_server(vm_name, fid, img_ref)

            self.assertEqual(r, FR.json['server'])

    from sys import argv

    from kamaki.clients.test import runTestCase

    not_found = True

        not_found = False

    if not argv[1:] or argv[1] == 'ComputeRest':

        not_found = False

    if not_found:

        print('TestCase %s not found' % argv[1])

# documentation are those of the authors and should not be

# interpreted as representing official policies, either expressed

# or implied, of GRNET S.A.

from mock import patch, call

from unittest import TestCase

from itertools import product

img_ref = "1m4g3-r3f3r3nc3"

vm_name = "my new VM"

cyclades_pkg = 'kamaki.clients.cyclades.CycladesClient'

    """Set up a Cyclades thorough test"""

    def setUp(self):

        self.url = 'http://cyclades.example.com'

        self.token = 'cyc14d3s70k3n'

    def tearDown(self):

        FR.json = vm_recv

                **kwargs))

    def assert_dicts_are_equal(self, d1, d2):

        for k, v in d1.items():

    def setUp(self):

        self.url = 'http://cyclades.example.com'

        self.token = 'cyc14d3s70k3n'

    def tearDown(self):

        FR.status_code = 200

        vm_id = vm_recv['server']['id']

        v = firewalls['attachments']['values'][0]['firewallProfile']

        with patch.object(

                return_value=firewalls) as GSD:

            r = self.client.get_firewall_profile(vm_id)

            GSD.assert_called_once_with(vm_id)

            self.assertEqual(r, v)

        with patch.object(

                return_value=dict()):

            self.assertRaises(

                ClientError,

            dict(id='another-nic-id', network_id='another-net-id'),

            dict(id=nic_id * 2, network_id=net_id * 2)]

        with patch.object(

                'list_server_nics',

                return_value=vm_nics) as LSN:

            r = self.client.disconnect_server(vm_id, nic_id)

        net_id = net_recv['network']['id']

        nics = ['nic1', 'nic2', 'nic3']

        with patch.object(

                'list_network_nics',

                return_value=nics) as LNN:

            self.client.disconnect_network_nics(net_id)

    def test_delete_network(self):

        net_id = net_recv['network']['id']

        with patch.object(

                return_value=FR()) as ND:

            self.client.delete_network(net_id)

            ND.assert_called_once_with(net_id)

        with patch.object(

                side_effect=ClientError('A 421 Error', 421)):

            try:

                self.client.delete_network(421)

    from sys import argv

    from kamaki.clients.test import runTestCase

    not_found = True

        not_found = False

        not_found = False

    if not_found:

        print('TestCase %s not found' % argv[1])

image_pkg = 'kamaki.clients.image.ImageClient'

    def assert_dicts_are_equal(self, d1, d2):

        for k, v in d1.items():

    def setUp(self):

        self.url = 'http://image.example.com'

        self.token = 'an1m@g370k3n=='

    def tearDown(self):

        FR.json = example_images

if __name__ == '__main__':

    from sys import argv

    from kamaki.clients.test import runTestCase

except ImportError:

    from kamaki.clients.commisioning.utils.ordereddict import OrderedDict

rest_pkg = 'kamaki.clients.pithos.rest_api.PithosRestClient'

        pass

    def setUp(self):

        self.url = 'https://www.example.com/pithos'

        self.token = 'p17h0570k3n'

        self.client.account = user_id

        self.client.container = 'c0nt@1n3r_i'

                **kwargs))

    files = []

    def setUp(self):

        self.url = 'https://www.example.com/pithos'

        self.token = 'p17h0570k3n'

        self.client.account = user_id

        self.client.container = 'c0nt@1n3r_i'

        FR.headers = object_info

        version = 'v3r510n'

        with patch.object(

                return_value=FR()) as head:

            r = self.client.get_object_info(obj)

            self.assertEqual(r, object_info)

                call(obj, version=None),

                call(obj, version=version)])

        with patch.object(

                side_effect=ClientError('Obj not found', 404)):

            self.assertRaises(

                ClientError,

        FR.json = object_hashmap

        for empty in (304, 412):

            with patch.object(

                    side_effect=ClientError('Empty', status=empty)):

                r = self.client.get_object_hashmap(obj)

                self.assertEqual(r, {})

            if_unmodified_since='some date here',

            data_range='10-20')

        with patch.object(

                return_value=FR()) as get:

            r = self.client.get_object_hashmap(obj)

            self.assertEqual(r, object_hashmap)

    def test_get_account_meta(self):

        key = 'x-account-meta-'

        with patch.object(

                return_value=account_info):

            r = self.client.get_account_meta()

            keys = [k for k in r if k.startswith(key)]

        acc_info['%sk2' % key] = 'v2'

        acc_info['%sk3' % key] = 'v3'

        with patch.object(

                return_value=acc_info):

            r = self.client.get_account_meta()

            for k in [k for k in acc_info if k.startswith(key)]:

    def test_get_account_group(self):

        key = 'x-account-group-'

        with patch.object(

                return_value=account_info):

            r = self.client.get_account_group()

            keys = [k for k in r if k.startswith(key)]

        acc_info['%sk2' % key] = 'g2'

        acc_info['%sk3' % key] = 'g3'

        with patch.object(

                return_value=acc_info):

            r = self.client.get_account_group()

            for k in [k for k in acc_info if k.startswith(key)]:

        container_plus[key] = metaval

        for ret in ((container_info, {}), (container_plus, {key: metaval})):

            with patch.object(

                    'get_container_info',

                    return_value=ret[0]) as GCI:

                for until in (None, somedate):

                (container_info, {key: ''}),

                (container_plus, {key: metaval})):

            with patch.object(

                    'get_container_info',

                    return_value=ret[0]) as GCI:

                for until in (None, somedate):

        val = 'pubL1c'

        oinfo['x-object-public'] = val

        with patch.object(

                return_value=oinfo) as GOF:

            r = self.client.publish_object(obj)

            self.assertEqual(

        info['x-object-sharing'] = '; '.join(

            ['%s=%s' % (k, v) for k, v in expected.items()])

        with patch.object(

                return_value=info) as GOF:

            r = self.client.get_object_sharing(obj)

            self.assertEqual(GOF.mock_calls[-1], call(obj))

        info['content-length'] = file_size

        block_size = container_info['x-container-block-size']

        with patch.object(

                return_value=info) as GOI:

            for start, end in (

                    (0, file_size + 1),

    from sys import argv

    from kamaki.clients.test import runTestCase

    not_found = True

        not_found = False

        not_found = False

    if not_found:

        print('TestCase %s not found' % argv[1])

        pass

    files = []

if __name__ == '__main__':

    from sys import argv

    from kamaki.clients.test import runTestCase

    KamakiResponse,

    KamakiHTTPResponse)

from kamaki.clients.utils.test import Utils

class ClientError(TestCase):