1:02 a.m.
From: Royce Lv <lvroyce(a)linux.vnet.ibm.com>
Signed-off-by: Royce Lv <lvroyce(a)linux.vnet.ibm.com>
---
tests/test_model_storagevolume.py | 95 +++++++++++++++++++++++++++++----------
1 file changed, 72 insertions(+), 23 deletions(-)
diff --git a/tests/test_model_storagevolume.py b/tests/test_model_storagevolume.py
index a3c3ce3..6433a1f 100644
--- a/tests/test_model_storagevolume.py
+++ b/tests/test_model_storagevolume.py
@@ -20,17 +20,18 @@
import json
import os
-import requests
+import tempfile
import unittest
+from cherrypy._cpreqbody import Entity
from functools import partial
-from kimchi.config import paths, READONLY_POOL_TYPE
+from kimchi.config import READONLY_POOL_TYPE
from kimchi.model.model import Model
from kimchi.mockmodel import MockModel
from kimchi.rollbackcontext import RollbackContext
-from utils import fake_auth_header, get_free_port, patch_auth, request
-from utils import rollback_wrapper, run_server, wait_task
+from utils import get_free_port, patch_auth, request
+from utils import rollback_wrapper, run_server, wait_task, running_as_root
model = None
@@ -149,25 +150,6 @@ def _do_volume_test(self, model, host, ssl_port, pool_name):
resp = self.request(vol_uri)
self.assertEquals(404, resp.status)
- # Create storage volume with 'file'
- filepath = os.path.join(paths.get_prefix(), 'COPYING.LGPL')
- url = 'https://%s:%s' % (host, ssl_port) + uri
- with open(filepath, 'rb') as fd:
- r = requests.post(url, files={'file': fd},
- verify=False,
- headers=fake_auth_header())
-
- if pool_info['type'] in READONLY_POOL_TYPE:
- self.assertEquals(r.status_code, 400)
- else:
- rollback.prependDefer(model.storagevolume_delete, pool_name,
- 'COPYING.LGPL')
- self.assertEquals(r.status_code, 202)
- task = r.json()
- wait_task(_task_lookup, task['id'])
- resp = self.request(uri + '/COPYING.LGPL')
- self.assertEquals(200, resp.status)
-
# Create storage volume with 'url'
url = 'https://github.com/kimchi-project/kimchi/raw/master/COPYING'
req = json.dumps({'url': url})
@@ -189,6 +171,73 @@ class StorageVolumeTests(unittest.TestCase):
def setUp(self):
self.request = partial(request, host, ssl_port)
+ @unittest.skipUnless(running_as_root(), 'Must be run as root')
+ def test_volume_upload(self):
+ class fake_header(object):
+ def elements(self, key):
+ return None
+
+ def __getattr__(self, key):
+ try:
+ return self.__getattribute__(key)
+ except AttributeError:
+ return lambda x, y: None
+
+ #inst = Model(objstore_loc=self.tmp_store)
+ vol_path = os.path.abspath('./run_tests.sh')
+
+ def do_upload(tmp_file, vol_path, url):
+ index = 0
+ chunk_size = 2 * 1000
+ with open(vol_path, 'rb') as fd:
+ while True:
+ with open(tmp_file, 'wb') as tmp_fd:
+ fd.seek(index * chunk_size)
+ data = fd.read(chunk_size)
+ tmp_fd.write(data)
+
+ # only file open for read can be handled by cherrypy
+ with open(tmp_file, 'rb') as tmp_fd:
+ # Hack cherrypy entity object and pass it to volume upload
+ entity = Entity(None, fake_header())
+ entity.file = tmp_fd
+ param = {'index': str(index),
+ 'chunk_size': str(chunk_size),
+ 'chunk': entity}
+ model.storagevolume_update(pool, params['name'], param)
+ vol_info = model.storagevolume_lookup(pool,
params['name'])
+ index = index + 1
+ if len(data) < chunk_size:
+ return vol_info
+
+ # Create a volume with raw format first, following upload will override it.
+ params = {'capacity': os.path.getsize(vol_path),
+ 'format': 'raw',
+ 'name': os.path.basename(vol_path)}
+ pool = 'default'
+
+ with RollbackContext() as rollback:
+ task_response = model.storagevolumes_create(pool, params)
+ rollback.prependDefer(model.storagevolume_delete, pool,
+ params['name'])
+ taskid = task_response['id']
+ vol_uri = task_response['target_uri']
+ model.task_wait(taskid)
+ self.assertEquals('finished',
model.task_lookup(taskid)['status'])
+
+ f = tempfile.NamedTemporaryFile(delete=False)
+ rollback.prependDefer(os.remove, f.name)
+ resp = do_upload(f.name, vol_path, vol_uri)
+
+ with open(vol_path) as vol_file:
+ vol_content = vol_file.read()
+
+ with open(resp['path']) as copy_file:
+ cp_content = copy_file.read()
+
+ self.assertEquals(vol_content, cp_content)
+
+
def test_get_storagevolume(self):
uri = '/storagepools/default/storagevolumes'
resp = self.request(uri)
--
2.1.0