From: Royce Lv <lvroyce@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@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) -- 2.1.0