stackforge-attic/murano-repository
Code Issues Proposed changes

Add some unit tests

Browse Source 
Refactor v1 api and archiver
Change-Id: I193023dd8eca4f8cc665a29a6b86698bb38edc83
This commit is contained in:
Ekaterina Fedorova 2013-10-21 11:45:08 +04:00
parent de388f5e64
commit 367d90f116
13 changed files with 647 additions and 165 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
AUTHORS
View File

@ -1,2 +1,3 @@
efedorova <efedorova@mirantis.com>
Ekaterina Fedorova <efedorova@mirantis.com>
EkaterinaFedorova <efedorova@mirantis.com>

10
ChangeLog
View File

@ -1,4 +1,12 @@
commit d259b6c33a4379c12dd673cb8cabc769545739d0
commit 5fbab4294125e6258e7e5c961522170214d5558d
Author: Ekaterina Fedorova <efedorova@mirantis.com>
Date: Mon Oct 21 11:45:08 2013 +0400
Add some unit tests
Change-Id: I193023dd8eca4f8cc665a29a6b86698bb38edc83
commit de388f5e64d887ff6b20790fb540c05ab43dacf7
Author: efedorova <efedorova@mirantis.com>
Date: Thu Oct 17 19:18:08 2013 +0400

67
muranorepository/api/v1.py
View File

@ -27,7 +27,7 @@ CONF = cfg.CONF
v1_api = Blueprint('v1', __name__)
def get_archive(client):
def _get_archive(client):
parser = ManifestParser(CONF.manifests)
manifests = parser.parse()
if client == 'conductor':
@ -35,11 +35,13 @@ def get_archive(client):
'heat',
'agent',
'scripts')
else:
elif client == 'ui':
return Archiver().create(manifests, client)
else:
abort(404)
def get_locations(data_type, result_path):
def _get_locations(data_type, result_path):
locations = []
if data_type == MANIFEST:
for item in os.listdir(result_path):
@ -57,7 +59,7 @@ def get_locations(data_type, result_path):
return jsonify({data_type: locations})
def save_file(request, result_path):
def _save_file(request, result_path):
file_to_upload = request.files.get('file')
if file_to_upload:
filename = secure_filename(file_to_upload.filename)
@ -67,66 +69,67 @@ def save_file(request, result_path):
abort(400)
def compose_path(data_type, path=None):
def _compose_path(data_type, path=None):
if path:
return os.path.join(CONF.manifests, getattr(CONF, data_type), path)
else:
return os.path.join(CONF.manifests, getattr(CONF, data_type))
def check_data_type(data_type):
def _check_data_type(data_type):
if data_type not in DATA_TYPES:
abort(404)
@v1_api.route('/client/ui')
def get_ui_data():
return send_file(get_archive('ui'))
@v1_api.route('/client/conductor')
def get_conductor_data():
return send_file(get_archive('conductor'))
@v1_api.route('/client/<path:type>')
def get_archive_data(type):
return send_file(_get_archive(type),
mimetype='application/octet-stream')
@v1_api.route('/admin/<data_type>')
def get_data_type_locations(data_type):
check_data_type(data_type)
result_path = compose_path(data_type)
return get_locations(data_type, result_path)
_check_data_type(data_type)
result_path = _compose_path(data_type)
return _get_locations(data_type, result_path)
@v1_api.route('/admin/<data_type>', methods=['POST'])
def upload_file(data_type):
check_data_type(data_type)
result_path = compose_path(data_type)
_check_data_type(data_type)
result_path = _compose_path(data_type)
try:
return save_file(request, result_path)
return _save_file(request, result_path)
except:
abort(403)
@v1_api.route('/admin/<data_type>/<path:path>')
def get_locations_in_nested_path_or_get_file(data_type, path):
check_data_type(data_type)
result_path = compose_path(data_type, path)
def _get_locations_in_nested_path_or_get_file(data_type, path):
_check_data_type(data_type)
result_path = _compose_path(data_type, path)
if os.path.isfile(result_path):
return send_file(result_path)
return send_file(result_path, mimetype='application/octet-stream')
else:
return get_locations(data_type, result_path)
return _get_locations(data_type, result_path)
@v1_api.route('/admin/<data_type>/<path:path>', methods=['POST'])
def upload_file_in_nested_path(data_type, path):
check_data_type(data_type)
result_path = compose_path(data_type, path)
return save_file(request, result_path)
_check_data_type(data_type)
result_path = _compose_path(data_type, path)
# it's forbidden to upload manifests to subfolder
if data_type == MANIFEST:
abort(403)
if not os.path.exists(result_path):
abort(404)
return _save_file(request, result_path)
@v1_api.route('/admin/<data_type>/<path:path>', methods=['PUT'])
def create_dirs(data_type, path):
check_data_type(data_type)
result_path = compose_path(data_type, path)
_check_data_type(data_type)
result_path = _compose_path(data_type, path)
resp = jsonify(result='success')
if os.path.exists(result_path):
return resp
@ -141,8 +144,8 @@ def create_dirs(data_type, path):
@v1_api.route('/admin/<data_type>/<path:path>', methods=['DELETE'])
def delete_dirictory_or_file(data_type, path):
check_data_type(data_type)
result_path = compose_path(data_type, path)
_check_data_type(data_type)
result_path = _compose_path(data_type, path)
if not os.path.exists(result_path):
abort(404)
if os.path.isfile(result_path):

57
muranorepository/tests/_test_client_api.py Normal file
View File

@ -0,0 +1,57 @@
# Copyright (c) 2013 Mirantis Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import sys
import os
from flask.ext.testing import TestCase as FlaskTestCase
import mockfs
possible_topdir = os.path.normpath(os.path.join(os.path.abspath(__file__),
os.pardir,
os.pardir,
os.pardir))
if os.path.exists(os.path.join(possible_topdir,
'muranorepository',
'__init__.py')):
sys.path.insert(0, possible_topdir)
from muranorepository.tests.fixtures import FIXTURE
from muranorepository import config
from muranorepository.main import make_app
class TestClientApi(FlaskTestCase):
url = "/v1/client/{0}"
def create_app(self):
test_app = make_app()
test_app.config['TESTING'] = True
return test_app
def setUp(self):
config_files = [os.path.join(possible_topdir,
'muranorepository',
'tests',
'test.conf')]
config.parse_configs(None, config_files)
self.mfs = mockfs.replace_builtins()
self.mfs.add_entries(FIXTURE)
def tearDown(self):
mockfs.restore_builtins()
def test_create_ui_archive(self):
# copy files in not implemented in mockfs
response = self.client.get(self.url.format('ui'))
self.assert200(response)

63
muranorepository/tests/_test_get_files.py Normal file
View File

@ -0,0 +1,63 @@
# Copyright (c) 2013 Mirantis Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import sys
import os
from flask.ext.testing import TestCase as FlaskTestCase
import mockfs
possible_topdir = os.path.normpath(os.path.join(os.path.abspath(__file__),
os.pardir,
os.pardir,
os.pardir))
if os.path.exists(os.path.join(possible_topdir,
'muranorepository',
'__init__.py')):
sys.path.insert(0, possible_topdir)
from muranorepository.consts import MANIFEST
from muranorepository.tests.fixtures import FIXTURE
from muranorepository import config
from muranorepository.main import make_app
class TestGetFiles(FlaskTestCase):
url = "/v1/admin/{0}/{1}"
def create_app(self):
test_app = make_app()
test_app.config['TESTING'] = True
return test_app
def setUp(self):
config_files = [os.path.join(possible_topdir,
'muranorepository',
'tests',
'test.conf')]
config.parse_configs(None, config_files)
self.mfs = mockfs.replace_builtins()
self.mfs.add_entries(FIXTURE)
def tearDown(self):
mockfs.restore_builtins()
def test_get_manifest_file(self):
data_type = MANIFEST
file_name = 'test-get-file-manifest.yaml'
content = 'content-manifest'
response = self.client.get(self.url.format(data_type,
file_name))
#os.path.getmtime is not implemented in Mockfs
self.assert200(response)
self.assertEqual(response, content)

63
muranorepository/tests/fixtures.py Normal file
View File

@ -0,0 +1,63 @@
# Copyright (c) 2013 Mirantis Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
MANIFEST_FILE = """
version: 0.1
service_display_name: Test
description: >-
<strong> This goes a description
full_service_name: test_service
author: Mirantis Inc.
service_version: 1.0
enabled: True
ui:
- test1.yaml
workflows:
- test1.xml
heat:
- Windows.template
agents:
- test1.template
scripts:
- test1.sh
"""
FIXTURE = {
'/bin/server': {
'test-manifest.yaml': MANIFEST_FILE,
'ui': {'test1.yaml': 'content_ui'},
'heat': {
'Windows.template': 'content_heat',
'folder_to_delete': {}
},
'agent': {
'test1.template': 'content_agent1',
'subfolder': {'test11.template': 'content_agent2',
'test12.template': '',
'subsubfolder': {'test21.template': 'content_agent3'}
},
},
'scripts': {'test1.sh': 'content_script'},
'workflows': {'test1.xml': 'content_workflow'},
'subfolder': {}
}
}

0
muranorepository/tests/fixtures/__init__.py vendored
View File

42
muranorepository/tests/fixtures/consts.py vendored
View File

@ -1,42 +0,0 @@
# Copyright (c) 2013 Mirantis Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
MANIFEST_FILE = ''
'version: 0.1'
'service_display_name: Test'
'description: >-'
' <strong> This goes a description'
'full_service_name: test_service'
'author: Mirantis Inc.'
'service_version: 1.0'
'enabled: True'
'ui:'
' - test1.yaml'
'workflows:'
' - test1.xml'
'heat:'
' - Windows.template'
'agents:'
' - test1.template'
'scripts:'
' - test1.sh'

109
muranorepository/tests/test_data_type_listing.py Normal file
View File

@ -0,0 +1,109 @@
# Copyright (c) 2013 Mirantis Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import sys
import os
from flask.ext.testing import TestCase as FlaskTestCase
import mockfs
possible_topdir = os.path.normpath(os.path.join(os.path.abspath(__file__),
os.pardir,
os.pardir,
os.pardir))
if os.path.exists(os.path.join(possible_topdir,
'muranorepository',
'__init__.py')):
sys.path.insert(0, possible_topdir)
from muranorepository.consts import MANIFEST
from muranorepository.consts import UI, WORKFLOW, AGENT, SCRIPTS, HEAT
from muranorepository.tests.fixtures import FIXTURE
from muranorepository import config
from muranorepository.main import make_app
class TestDataTypeListing(FlaskTestCase):
url = "/v1/admin/{0}"
url_with_path = "/v1/admin/{0}/{1}"
def create_app(self):
test_app = make_app()
test_app.config['TESTING'] = True
return test_app
def setUp(self):
config_files = [os.path.join(possible_topdir,
'muranorepository',
'tests',
'test.conf')]
config.parse_configs(None, config_files)
self.mfs = mockfs.replace_builtins()
self.mfs.add_entries(FIXTURE)
def tearDown(self):
mockfs.restore_builtins()
def test_list_manifests_files(self):
response = self.client.get(self.url.format(MANIFEST))
self.assert200(response)
expected_result = {MANIFEST: ['test-manifest.yaml']}
self.assertEquals(response.json, expected_result)
def test_list_ui_files(self):
response = self.client.get(self.url.format(UI))
self.assert200(response)
expected_result = {UI: ['test1.yaml']}
self.assertEquals(response.json, expected_result)
def test_list_workflows(self):
response = self.client.get(self.url.format(WORKFLOW))
self.assert200(response)
expected_result = {WORKFLOW: ['test1.xml']}
self.assertEquals(response.json, expected_result)
def test_list_heat_templates(self):
response = self.client.get(self.url.format(HEAT))
expected_result = {HEAT: ['Windows.template']}
self.assert200(response)
self.assertEquals(response.json, expected_result)
def test_list_agent_templates(self):
response = self.client.get(self.url.format(AGENT))
self.assert200(response)
expected_result = {AGENT: ['test1.template',
'subfolder/test12.template',
'subfolder/test11.template',
'subfolder/subsubfolder/test21.template']}
self.assertEquals(response.json, expected_result)
def test_list_scripts(self):
response = self.client.get(self.url.format(SCRIPTS))
self.assert200(response)
expected_result = {SCRIPTS: ['test1.sh']}
self.assertEquals(response.json, expected_result)
def test_list_agent_templates_from_subfolder(self):
url = self.url_with_path.format(AGENT, 'subfolder')
response = self.client.get(url)
self.assert200(response)
expected_result = {AGENT: ['test12.template',
'test11.template',
'subsubfolder/test21.template']}
self.assertEquals(response.json, expected_result)
def test_list_agent_templates_from_subsubfolder(self):
url = self.url_with_path.format(AGENT, 'subfolder/subsubfolder')
response = self.client.get(url)
self.assert200(response)
expected_result = {AGENT: ['test21.template']}
self.assertEquals(response.json, expected_result)

115
muranorepository/tests/test_dir_operations.py Normal file
View File

@ -0,0 +1,115 @@
# Copyright (c) 2013 Mirantis Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import sys
import shutil
import os
from flask.ext.testing import TestCase as FlaskTestCase
import mockfs
possible_topdir = os.path.normpath(os.path.join(os.path.abspath(__file__),
os.pardir,
os.pardir,
os.pardir))
if os.path.exists(os.path.join(possible_topdir,
'muranorepository',
'__init__.py')):
sys.path.insert(0, possible_topdir)
from muranorepository.consts import MANIFEST
from muranorepository.consts import UI, AGENT, HEAT, WORKFLOW, SCRIPTS
from muranorepository.tests.fixtures import FIXTURE
from muranorepository import config
from muranorepository.main import make_app
class TestDirOperations(FlaskTestCase):
url = "/v1/admin/{0}"
url_with_path = "/v1/admin/{0}/{1}"
expected_result = {'result': 'success'}
def create_app(self):
test_app = make_app()
test_app.config['TESTING'] = True
return test_app
def setUp(self):
config_files = [os.path.join(possible_topdir,
'muranorepository',
'tests',
'test.conf')]
config.parse_configs(None, config_files)
self.mfs = mockfs.replace_builtins()
self.mfs.add_entries(FIXTURE)
def tearDown(self):
mockfs.restore_builtins()
def test_create_ui_subfolder(self):
response = self.client.put(self.url_with_path.format(UI, 'test'))
self.assert200(response)
created_folder = 'bin/server/{0}/test'.format(UI)
self.assertTrue(os.path.exists(created_folder))
self.assertEquals(response.json, self.expected_result)
shutil.rmtree(created_folder)
def test_create_workflow_subfolder(self):
response = self.client.put(self.url_with_path.format(WORKFLOW,
'test'))
self.assert200(response)
created_folder = 'bin/server/{0}/test'.format(WORKFLOW)
self.assertTrue(os.path.exists(created_folder))
self.assertEquals(response.json, self.expected_result)
shutil.rmtree(created_folder)
def test_create_heat_subfolder(self):
response = self.client.put(self.url_with_path.format(HEAT,
'test_heat'))
self.assert200(response)
created_folder = 'bin/server/{0}/test_heat'.format(HEAT)
self.assertTrue(os.path.exists(created_folder))
self.assertEquals(response.json, self.expected_result)
shutil.rmtree(created_folder)
def test_create_scripts_subfolder(self):
response = self.client.put(self.url_with_path.format(SCRIPTS, 'test'))
self.assert200(response)
created_folder = 'bin/server/{0}/test'.format(SCRIPTS)
self.assertTrue(os.path.exists(created_folder))
self.assertEquals(response.json, self.expected_result)
shutil.rmtree(created_folder)
def test_create_nested_agent_folders(self):
response = self.client.put(self.url_with_path.format(AGENT,
'sub1/sub2'))
self.assert200(response)
self.assertTrue(
os.path.exists('bin/server/{0}/sub1/sub2'.format(AGENT)))
self.assertEquals(response.json, self.expected_result)
shutil.rmtree('bin/server/{0}/sub1'.format(AGENT))
def test_delete_heat_subfolder(self):
url = self.url_with_path.format(HEAT, 'folder_to_delete')
response = self.client.delete(url)
self.assert200(response)
self.assertEquals(response.json, self.expected_result)
#negative tests
def test_create_root_subfolder(self):
response = self.client.put(self.url_with_path.format(MANIFEST,
'test'))
self.assert403(response)
created_folder = 'bin/server/{0}/test'.format(MANIFEST)
self.assertFalse(os.path.exists(created_folder))

87
muranorepository/tests/test_main.py
View File

@ -1,87 +0,0 @@
import sys
import os
from flask.ext.testing import TestCase as FlaskTestCase
import shutil
from StringIO import StringIO
import mockfs
possible_topdir = os.path.normpath(os.path.join(os.path.abspath(__file__),
os.pardir,
os.pardir,
os.pardir))
if os.path.exists(os.path.join(possible_topdir,
'muranorepository',
'__init__.py')):
sys.path.insert(0, possible_topdir)
from muranorepository.consts import MANIFEST
from muranorepository.tests.fixtures.consts import MANIFEST_FILE
from muranorepository import config
from muranorepository.main import make_app
class TestAdminAPI(FlaskTestCase):
url = "/v1/admin/{0}"
url_with_path = "/v1/admin/{0}/{1}"
def create_app(self):
test_app = make_app()
test_app.config['TESTING'] = True
return test_app
def setUp(self):
config_files = [os.path.join(possible_topdir,
'muranorepository',
'tests',
'test.conf')]
config.parse_configs(None, config_files)
self.mfs = mockfs.replace_builtins()
self.mfs.add_entries(
{
'/bin/server': {
'test-manifest.yaml': MANIFEST_FILE,
'ui': {'test1.yaml': ''},
'heat':
{'Windows.template': '',
'folder_to_delete': {}
}
}
})
def tearDown(self):
mockfs.restore_builtins()
def test_list_manifests(self):
response = self.client.get(self.url.format(MANIFEST))
expected_result = {MANIFEST: ['test-manifest.yaml']}
self.assert200(response)
self.assertEquals(response.json, expected_result)
def test_list_ui(self):
response = self.client.get(self.url.format('ui'))
expected_result = {'ui': ['test1.yaml']}
self.assert200(response)
self.assertEquals(response.json, expected_result)
def test_create_ui_subfolder(self):
response = self.client.put(self.url_with_path.format('ui', 'test'))
expected_result = {'result': 'success'}
self.assert200(response)
self.assertEquals(response.json, expected_result)
shutil.rmtree('bin/server/ui/test')
def test_delete_heat_subfolder(self):
url = self.url_with_path.format('heat',
'folder_to_delete')
response = self.client.delete(url)
self.assert200(response)
expected_result = {'result': 'success'}
self.assertEquals(response.json, expected_result)
def test_upload_ui_file(self):
upload_data = {'file': (StringIO('content'), 'test.yaml')}
response = self.client.post(self.url.format('ui'),
data=upload_data)
'test.yaml' in os.listdir('bin/server/ui')
self.assert200(response)
expected_result = {'result': 'success'}
self.assertEquals(response.json, expected_result)

188
muranorepository/tests/test_upload.py Normal file
View File

@ -0,0 +1,188 @@
# Copyright (c) 2013 Mirantis Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import sys
from StringIO import StringIO
import os
from flask.ext.testing import TestCase as FlaskTestCase
import mockfs
possible_topdir = os.path.normpath(os.path.join(os.path.abspath(__file__),
os.pardir,
os.pardir,
os.pardir))
if os.path.exists(os.path.join(possible_topdir,
'muranorepository',
'__init__.py')):
sys.path.insert(0, possible_topdir)
from muranorepository.consts import MANIFEST
from muranorepository.consts import UI, WORKFLOW, AGENT, HEAT, SCRIPTS
from muranorepository.tests.fixtures import FIXTURE
from muranorepository import config
from muranorepository.main import make_app
class TestUploadFiles(FlaskTestCase):
url = '/v1/admin/{0}'
url_with_path = '/v1/admin/{0}/{1}'
path_to_upload = 'bin/server/{0}/{1}'
expected_result = {'result': 'success'}
def create_app(self):
test_app = make_app()
test_app.config['TESTING'] = True
return test_app
def setUp(self):
config_files = [os.path.join(possible_topdir,
'muranorepository',
'tests',
'test.conf')]
config.parse_configs(None, config_files)
self.mfs = mockfs.replace_builtins()
self.mfs.add_entries(FIXTURE)
def tearDown(self):
mockfs.restore_builtins()
def test_upload_manifest_file(self):
data_type = MANIFEST
file_name = 'new_manifest.yaml'
upload_data = {'file': (StringIO('content'), file_name)}
response = self.client.post(self.url.format(data_type),
data=upload_data)
self.assert200(response)
self.assertTrue(os.path.exists('bin/server/{0}'.format(file_name)))
self.assertEquals(response.json, self.expected_result)
os.remove('bin/server/{0}'.format(file_name))
def test_upload_ui_file(self):
data_type = UI
file_name = 'new.yaml'
upload_data = {'file': (StringIO('content'), file_name)}
response = self.client.post(self.url.format(data_type),
data=upload_data)
self.assert200(response)
self.assertTrue(os.path.exists(self.path_to_upload.format(data_type,
file_name)))
self.assertEquals(response.json, self.expected_result)
os.remove(self.path_to_upload.format(data_type, file_name))
def test_upload_workflow_file(self):
data_type = WORKFLOW
file_name = 'new.xml'
upload_data = {'file': (StringIO('content'), file_name)}
response = self.client.post(self.url.format(data_type),
data=upload_data)
self.assert200(response)
self.assertTrue(os.path.exists(self.path_to_upload.format(data_type,
file_name)))
self.assertEquals(response.json, self.expected_result)
os.remove(self.path_to_upload.format(data_type, file_name))
def test_upload_agent_file(self):
data_type = AGENT
file_name = 'new.ps1'
upload_data = {'file': (StringIO('content'), file_name)}
response = self.client.post(self.url.format(data_type),
data=upload_data)
self.assert200(response)
self.assertTrue(os.path.exists(self.path_to_upload.format(data_type,
file_name)))
self.assertEquals(response.json, self.expected_result)
os.remove(self.path_to_upload.format(data_type, file_name))
def test_upload_script(self):
data_type = SCRIPTS
file_name = 'new.sh'
upload_data = {'file': (StringIO('content'), file_name)}
response = self.client.post(self.url.format(data_type),
data=upload_data)
self.assert200(response)
self.assertTrue(os.path.exists(self.path_to_upload.format(data_type,
file_name)))
self.assertEquals(response.json, self.expected_result)
os.remove(self.path_to_upload.format(data_type, file_name))
def test_upload_heat_file(self):
data_type = HEAT
file_name = 'new_heat_template.yaml'
upload_data = {'file': (StringIO('content'), file_name)}
response = self.client.post(self.url.format(data_type),
data=upload_data)
self.assert200(response)
self.assertTrue(os.path.exists(self.path_to_upload.format(data_type,
file_name)))
self.assertEquals(response.json, self.expected_result)
os.remove(self.path_to_upload.format(data_type, file_name))
def test_upload_agent_template_to_subfolder(self):
data_type = AGENT
file_name = 'new_agent1.template'
folder_name = 'subfolder'
path_to_upload = 'bin/server/{0}/{1}/{2}'.format(data_type,
folder_name,
file_name)
upload_data = {'file': (StringIO('content'), file_name)}
response = self.client.post(
self.url_with_path.format(data_type,
folder_name),
data=upload_data)
self.assert200(response)
self.assertTrue(os.path.exists(path_to_upload))
os.remove(path_to_upload)
def test_upload_agent_template_to_subsubfolder(self):
data_type = AGENT
file_name = 'new_agent2.template'
folders = 'subfolder/subsubfolder'
upload_data = {'file': (StringIO('content'), file_name)}
path_to_upload = 'bin/server/{0}/{1}/{2}'.format(data_type,
folders,
file_name)
response = self.client.post(
self.url_with_path.format(data_type, folders),
data=upload_data)
self.assert200(response)
self.assertTrue(
os.path.exists(path_to_upload))
os.remove(path_to_upload)
# Negative tests
def test_upload_manifest_to_subfolder(self):
data_type = MANIFEST
file_name = 'new_manifest1.yaml'
upload_data = {'file': (StringIO('content'), file_name)}
response = self.client.post(
self.url_with_path.format(data_type,
'subfolder'),
data=upload_data)
self.assert403(response)
self.assertFalse(os.path.exists(self.path_to_upload.format('subfolder',
file_name)))
def test_upload_ui_file_to_non_existent_folder(self):
data_type = UI
file_name = 'new_ui_desc.yaml'
folder_name = 'subfolder'
upload_data = {'file': (StringIO('content'), file_name)}
response = self.client.post(
self.url_with_path.format(data_type,
folder_name),
data=upload_data)
self.assert404(response)

10
muranorepository/utils/archiver.py
View File

@ -40,8 +40,8 @@ class Archiver(object):
"{0}".format(file))
def _compose_archive(self, path):
target_archive = "service_metadata.tar"
with tarfile.open(target_archive, "w") as tar:
target_archive = "data.tar.gz"
with tarfile.open(target_archive, "w:gz") as tar:
for item in os.listdir(path):
tar.add(os.path.join(path, item), item)
try:
@ -57,7 +57,11 @@ class Archiver(object):
return: absolute path to created archive
"""
temp_dir = tempfile.mkdtemp()
#TODO: temporary hack for mockfs
try:
temp_dir = tempfile.mkdtemp()
except:
temp_dir = '/tmp'
for data_type in types:
if data_type not in DATA_TYPES:
raise Exception("Please, specify one of the supported data "