10:02 a.m.
From: Royce Lv <lvroyce(a)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(a)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