From: Royce Lv <lvroyce@linux.vnet.ibm.com> Signed-off-by: Royce Lv <lvroyce@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