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

[lvroyce at linux.vnet.ibm.com]

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)
-- 
2.1.0