Revision 5ed87051

b/kamaki/clients/compute/test.py
35 35
from unittest import TestCase
36 36
from itertools import product
37 37
from json import dumps
38
from sys import stdout
38 39

  
39 40
from kamaki.clients import ClientError, compute
40 41

  
......
119 120
        FR.json = vm_recv
120 121

  
121 122
    @patch('%s.get' % rest_pkg, return_value=FR())
122
    def _test_get(self, service, get):
123
        for args in product(
123
    def test_limits_get(self, get):
124
        self.client.limits_get(success='some_val')
125
        get.assert_called_once_with('/limits', success='some_val')
126

  
127
    @patch('%s.set_param' % rest_pkg)
128
    @patch('%s.get' % rest_pkg, return_value=FR())
129
    def _test_get(self, service, params, get, set_param):
130
        param_args = [({}, {k: k}, {k: v[1]}) for k, v in params.items()]
131
        num_of_its = ''
132
        stdout.write('# of iterations: ')
133
        i = 0
134
        for i, args in enumerate(product(
124 135
                ('', '%s_id' % service),
125
                ('', 'cmd'),
136
                (None, False, True),
126 137
                (200, 204),
127
                ({}, {'k': 'v'})):
128
            (srv_id, command, success, kwargs) = args
138
                ({}, {'k': 'v'}),
139
                *param_args)):
140
            (srv_id, detail, success, kwargs) = args[:4]
141
            kwargs['success'] = success
142
            srv_kwargs = dict()
143
            for param in args[4:]:
144
                srv_kwargs.update(param)
145
            srv_kwargs.update(kwargs)
129 146
            method = getattr(self.client, '%s_get' % service)
130
            method(*args[:3], **kwargs)
131
            srv_str = '/%s' % srv_id if srv_id else ''
132
            cmd_str = '/%s' % command if command else ''
133
            self.assertEqual(get.mock_calls[-1], call(
134
                '/%s%s%s' % (service, srv_str, cmd_str),
135
                success=success,
136
                **kwargs))
147
            method(*args[:2], **srv_kwargs)
148
            srv_str = '/detail' if detail else (
149
                '/%s' % srv_id) if srv_id else ''
150
            self.assertEqual(
151
                get.mock_calls[-1],
152
                call('/%s%s' % (service, srv_str), **kwargs))
153
            param_calls = []
154
            for k, v in params.items():
155
                real_v = srv_kwargs.get(k, v[1]) if not srv_id else v[1]
156
                param_calls.append(call(v[0], real_v, iff=real_v))
157
            actual = set_param.mock_calls[- len(param_calls):]
158
            self.assertEqual(sorted(actual), sorted(param_calls))
159

  
160
            if not i % 1000:
161
                stdout.write('\b' * len(num_of_its))
162
                num_of_its = '%s' % i
163
                stdout.write(num_of_its)
164
                stdout.flush()
165
        print ('\b' * len(num_of_its)) + ('%s' % i)
137 166

  
138 167
    def test_servers_get(self):
139
        self._test_get('servers')
168
        params = dict(
169
            changes_since=('changes-since', None),
170
            image=('image', None),
171
            flavor=('flavor', None),
172
            name=('name', None),
173
            marker=('marker', None),
174
            limit=('limit', None),
175
            status=('status', None),
176
            host=('host', None))
177
        self._test_get('servers', params)
140 178

  
141 179
    def test_flavors_get(self):
142
        self._test_get('flavors')
180
        params = dict(
181
            changes_since=('changes-since', None),
182
            minDisk=('minDisk', None),
183
            minRam=('minRam', None),
184
            marker=('marker', None),
185
            limit=('limit', None))
186
        self._test_get('flavors', params)
143 187

  
144 188
    def test_images_get(self):
145
        self._test_get('images')
189
        param = dict(
190
            changes_since=('changes-since', None),
191
            server_name=('server', None),
192
            name=('name', None),
193
            status=('status', None),
194
            marker=('marker', None),
195
            limit=('limit', None),
196
            type=('type', None))
197
        self._test_get('images', param)
146 198

  
147 199
    @patch('%s.delete' % rest_pkg, return_value=FR())
148 200
    def _test_delete(self, service, delete):

Also available in: Unified diff