[Kimchi-devel] [kimchi-devel][PATCHv4 7/7] A test script against real model server

[Aline Manera]

On 22/04/2015 05:02, lvroyce at linux.vnet.ibm.com wrote:
> From: Royce Lv <lvroyce at linux.vnet.ibm.com>
>
> This is a test script against real model server,
> you can use it to verify the logic of upload.
> NOTE: pls change the username/password/file path
> to a valid one.
>
> Signed-off-by: Royce Lv <lvroyce at linux.vnet.ibm.com>
> ---
>   tests/test_api.py | 108 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
>   1 file changed, 108 insertions(+)
>   create mode 100644 tests/test_api.py
>
> diff --git a/tests/test_api.py b/tests/test_api.py
> new file mode 100644
> index 0000000..953716a
> --- /dev/null
> +++ b/tests/test_api.py
> @@ -0,0 +1,108 @@
> +# -*- coding: utf-8 -*-
> +#
> +# Project Kimchi
> +#
> +# Copyright IBM, Corp. 2013-2014
> +#
> +# This library is free software; you can redistribute it and/or
> +# modify it under the terms of the GNU Lesser General Public
> +# License as published by the Free Software Foundation; either
> +# version 2.1 of the License, or (at your option) any later version.
> +#
> +# This library is distributed in the hope that it will be useful,
> +# but WITHOUT ANY WARRANTY; without even the implied warranty of
> +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
> +# Lesser General Public License for more details.
> +#
> +# You should have received a copy of the GNU Lesser General Public
> +# License along with this library; if not, write to the Free Software
> +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
> +
> +import base64
> +import json
> +import os
> +import requests
> +import unittest
> +import time
> +
> +from functools import partial
> +
> +from utils import request
> +
> +
> +host = None
> +ssl_port = None
> +
> +
> +def setUpModule():
> +    global host, ssl_port
> +
> +    host = '127.0.0.1'
> +    ssl_port = 8001
> +
> +
> +class RestTests(unittest.TestCase):
> +    def setUp(self):
> +        self.request = partial(request, host, ssl_port)
> +
> +    def assertHTTPStatus(self, code, *args):
> +        resp = self.request(*args)
> +        self.assertEquals(code, resp.status)
> +
> +    def assertValidJSON(self, txt):
> +        try:
> +            json.loads(txt)
> +        except ValueError:
> +            self.fail("Invalid JSON: %s" % txt)
> +



> +    def test_upload(self):
> +        # If we use self.request, we may encode multipart formdata by ourselves
> +        # requests lib take care of encode part, so use this lib instead
> +        def fake_auth_header():
> +            headers = {'Accept': 'application/json'}
> +            user, pw = ('this is your user', 'this is your password')
> +            hdr = "Basic " + base64.b64encode("%s:%s" % (user, pw))
> +            headers['AUTHORIZATION'] = hdr
> +            return headers
> +
> +        def do_upload(vol_path, url):
> +            index = 0
> +            chunk_size = 2 * 1000 * 1000
> +            with open(vol_path, 'rb') as fd:
> +                while True:
> +                    with open(vol_path + '.tmp', 'wb') as tmp_fd:
> +                    #  make sure it is truncated
> +                        fd.seek(index*chunk_size)
> +                        data = fd.read(chunk_size)
> +                        tmp_fd.write(data)
> +
> +                    with open(vol_path + '.tmp', 'rb') as tmp_fd:
> +                        print "size of data %s" % len(data)
> +                        r = requests.put(url,
> +                                data={"index": str(index), "chunk_size": str(chunk_size)},
> +                                files={"chunk": tmp_fd},
> +                                verify=False,
> +                                headers=fake_auth_header())
> +                        self.assertEquals(r.status_code, 200)
> +                        index = index + 1
> +                        print "index is %s" % index
> +                    if len(data) < chunk_size:
> +                            return index
> +
> +        def do_thousands(url):
> +            for i in range(1000):
> +                self.request(url, headers=fake_auth_header())
> +
> +        vol_path = os.path.join('/home/royce/upload/vm2.qcow2')
> +        file_len = os.path.getsize(vol_path)
> +        url = "https://%s:%s/storagepools/default/storagevolumes/" % \
> +                (host, ssl_port)
> +        body = json.dumps({'capacity': file_len, 'name': os.path.basename(vol_path), 'format': 'raw'})
> +        r = requests.post(url,
> +                          data=body,
> +                          verify=False,
> +                          headers=fake_auth_header())
> +        time.sleep(1)
> +        print r.text
> +        url = url + os.path.basename(vol_path)
> +        do_upload(vol_path, url)

This test should be added to the Kimchi test cases suite.