715 lines
26 KiB
Python
715 lines
26 KiB
Python
# 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 json
|
|
import os
|
|
import shutil
|
|
import six
|
|
import sys
|
|
import tempfile
|
|
|
|
from testtools import matchers
|
|
|
|
from muranoclient.common import exceptions as common_exceptions
|
|
from muranoclient.common import utils as mc_utils
|
|
from muranoclient.osc.v1 import package as osc_pkg
|
|
from muranoclient.tests.unit.osc.v1 import fakes
|
|
from muranoclient.tests.unit import test_utils
|
|
from muranoclient.v1 import packages
|
|
|
|
import mock
|
|
from osc_lib import exceptions as exc
|
|
from osc_lib import utils
|
|
import requests_mock
|
|
|
|
make_pkg = test_utils.make_pkg
|
|
|
|
FIXTURE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__),
|
|
'fixture_data'))
|
|
|
|
COLUMNS = ['Id', 'Name', 'Fully_qualified_name', 'Author', 'Active',
|
|
'Is public', 'Type', 'Version']
|
|
|
|
DATA = {
|
|
'class_definitions': ['com.example.apache.ApacheHttpServer'],
|
|
'updated': '2016-09-20T06:23:45.000000',
|
|
'description': 'Test description.\n',
|
|
'created': '2016-09-20T06:23:15.000000',
|
|
'author': 'Mirantis, Inc',
|
|
'enabled': True,
|
|
'owner_id': 'a203405ea871484a940850d6c0b8dfd9',
|
|
'tags': ['Server', 'WebServer', 'Apache', 'HTTP', 'HTML'],
|
|
'is_public': False,
|
|
'fully_qualified_name': 'com.example.apache.ApacheHttpServer',
|
|
'type': 'Application',
|
|
'id': '46860070-5f8a-4936-96e8-d7b89e5187d7',
|
|
'categories': [],
|
|
'name': 'Apache HTTP Server'
|
|
}
|
|
|
|
|
|
class TestPackage(fakes.TestApplicationCatalog):
|
|
def setUp(self):
|
|
super(TestPackage, self).setUp()
|
|
self.package_mock = self.app.client_manager.application_catalog.\
|
|
packages
|
|
self.package_mock.reset_mock()
|
|
|
|
|
|
class TestCreatePackage(TestPackage):
|
|
def setUp(self):
|
|
super(TestCreatePackage, self).setUp()
|
|
|
|
# Command to test
|
|
self.cmd = osc_pkg.CreatePackage(self.app, None)
|
|
|
|
def test_create_package_without_args(self):
|
|
arglist = []
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
error = self.assertRaises(exc.CommandError,
|
|
self.cmd.take_action, parsed_args)
|
|
self.assertEqual('Provide --template for a HOT-based package, OR at '
|
|
'least --classes-dir for a MuranoPL-based package',
|
|
str(error))
|
|
|
|
def test_create_package_template_and_classes_args(self):
|
|
heat_template = os.path.join(FIXTURE_DIR, 'heat-template.yaml')
|
|
classes_dir = os.path.join(FIXTURE_DIR, 'test-app', 'Classes')
|
|
arglist = ['--template', heat_template, '--classes-dir', classes_dir]
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
error = self.assertRaises(exc.CommandError,
|
|
self.cmd.take_action, parsed_args)
|
|
self.assertEqual('Provide --template for a HOT-based package, OR'
|
|
' --classes-dir for a MuranoPL-based package',
|
|
str(error))
|
|
|
|
def test_create_hot_based_package(self):
|
|
with tempfile.NamedTemporaryFile() as f:
|
|
RESULT_PACKAGE = f.name
|
|
heat_template = os.path.join(FIXTURE_DIR, 'heat-template.yaml')
|
|
logo = os.path.join(FIXTURE_DIR, 'logo.png')
|
|
arglist = ['--template', heat_template, '--output', RESULT_PACKAGE,
|
|
'-l', logo]
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
orig = sys.stdout
|
|
try:
|
|
sys.stdout = six.StringIO()
|
|
self.cmd.take_action(parsed_args)
|
|
finally:
|
|
stdout = sys.stdout.getvalue()
|
|
sys.stdout.close()
|
|
sys.stdout = orig
|
|
matchers.MatchesRegex(stdout,
|
|
"Application package "
|
|
"is available at {0}".format(RESULT_PACKAGE))
|
|
|
|
def test_create_mpl_package(self):
|
|
with tempfile.NamedTemporaryFile() as f:
|
|
RESULT_PACKAGE = f.name
|
|
classes_dir = os.path.join(FIXTURE_DIR, 'test-app', 'Classes')
|
|
resources_dir = os.path.join(FIXTURE_DIR, 'test-app', 'Resources')
|
|
ui = os.path.join(FIXTURE_DIR, 'test-app', 'ui.yaml')
|
|
arglist = ['-c', classes_dir, '-r', resources_dir,
|
|
'-u', ui, '-o', RESULT_PACKAGE]
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
orig = sys.stdout
|
|
try:
|
|
sys.stdout = six.StringIO()
|
|
self.cmd.take_action(parsed_args)
|
|
finally:
|
|
stdout = sys.stdout.getvalue()
|
|
sys.stdout.close()
|
|
sys.stdout = orig
|
|
matchers.MatchesRegex(stdout,
|
|
"Application package "
|
|
"is available at {0}".format(RESULT_PACKAGE))
|
|
|
|
|
|
class TestPackageList(TestPackage):
|
|
|
|
def setUp(self):
|
|
super(TestPackageList, self).setUp()
|
|
self.cmd = osc_pkg.ListPackages(self.app, None)
|
|
self.package_mock.filter.return_value = \
|
|
[packages.Package(None, DATA)]
|
|
utils.get_dict_properties = mock.MagicMock(return_value='')
|
|
|
|
def test_package_list_defaults(self):
|
|
arglist = []
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
|
|
columns, data = self.cmd.take_action(parsed_args)
|
|
|
|
self.package_mock.filter.assert_called_with(
|
|
include_disabled=False,
|
|
owned=False)
|
|
self.assertEqual(COLUMNS, columns)
|
|
|
|
def test_package_list_with_limit(self):
|
|
arglist = ['--limit', '10']
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
|
|
columns, data = self.cmd.take_action(parsed_args)
|
|
|
|
self.package_mock.filter.assert_called_with(
|
|
include_disabled=False,
|
|
limit=10,
|
|
owned=False)
|
|
|
|
def test_package_list_with_marker(self):
|
|
arglist = ['--marker', '12345']
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
|
|
columns, data = self.cmd.take_action(parsed_args)
|
|
|
|
self.package_mock.filter.assert_called_with(
|
|
include_disabled=False,
|
|
marker='12345',
|
|
owned=False)
|
|
|
|
def test_package_list_with_name(self):
|
|
arglist = ['--name', 'mysql']
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
|
|
columns, data = self.cmd.take_action(parsed_args)
|
|
|
|
self.package_mock.filter.assert_called_with(
|
|
include_disabled=False,
|
|
name='mysql',
|
|
owned=False)
|
|
|
|
def test_package_list_with_fqn(self):
|
|
arglist = ['--fqn', 'mysql']
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
|
|
columns, data = self.cmd.take_action(parsed_args)
|
|
|
|
self.package_mock.filter.assert_called_with(
|
|
include_disabled=False,
|
|
fqn='mysql',
|
|
owned=False)
|
|
|
|
|
|
class TestPackageDelete(TestPackage):
|
|
def setUp(self):
|
|
super(TestPackageDelete, self).setUp()
|
|
self.package_mock.delete.return_value = None
|
|
self.package_mock.filter.return_value = \
|
|
[packages.Package(None, DATA)]
|
|
|
|
# Command to test
|
|
self.cmd = osc_pkg.DeletePackage(self.app, None)
|
|
|
|
@mock.patch('osc_lib.utils.get_item_properties')
|
|
def test_package_delete(self, mock_util):
|
|
arglist = ['fake1']
|
|
verifylist = [('id', ['fake1'])]
|
|
|
|
mock_util.return_value = ('1234', 'Core library',
|
|
'io.murano', 'murano.io', '',
|
|
'True', 'Library', '0.0.0'
|
|
)
|
|
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
columns, data = self.cmd.take_action(parsed_args)
|
|
|
|
# Check that columns are correct
|
|
self.assertEqual(COLUMNS, columns)
|
|
|
|
# Check that data is correct
|
|
expected_data = [('1234', 'Core library', 'io.murano',
|
|
'murano.io', '', 'True', 'Library', '0.0.0')]
|
|
self.assertEqual(expected_data, data)
|
|
|
|
|
|
class TestPackageImport(TestPackage):
|
|
def setUp(self):
|
|
super(TestPackageImport, self).setUp()
|
|
self.package_mock.filter.return_value = \
|
|
[packages.Package(None, DATA)]
|
|
|
|
# Command to test
|
|
self.cmd = osc_pkg.ImportPackage(self.app, None)
|
|
|
|
@mock.patch('muranoclient.common.utils.Package.from_file')
|
|
def test_package_import(self, from_file):
|
|
with tempfile.NamedTemporaryFile() as f:
|
|
RESULT_PACKAGE = f.name
|
|
categories = ['Cat1', 'Cat2 with space']
|
|
|
|
pkg = make_pkg({'FullName': RESULT_PACKAGE})
|
|
from_file.return_value = mc_utils.Package(mc_utils.File(pkg))
|
|
|
|
arglist = [RESULT_PACKAGE, '--categories',
|
|
categories, '--is-public']
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
self.package_mock.create.assert_called_once_with({
|
|
'categories': [categories],
|
|
'is_public': True
|
|
}, {RESULT_PACKAGE: mock.ANY},)
|
|
|
|
def _test_conflict(self,
|
|
packages, from_file, raw_input_mock,
|
|
input_action, exists_action=''):
|
|
packages.create = mock.MagicMock(
|
|
side_effect=[common_exceptions.HTTPConflict("Conflict"), None])
|
|
|
|
packages.filter.return_value = [mock.Mock(id='test_id')]
|
|
|
|
raw_input_mock.return_value = input_action
|
|
with tempfile.NamedTemporaryFile() as f:
|
|
pkg = make_pkg({'FullName': f.name})
|
|
from_file.return_value = mc_utils.Package(mc_utils.File(pkg))
|
|
if exists_action:
|
|
arglist = [f.name, '--exists-action', exists_action]
|
|
else:
|
|
arglist = [f.name]
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
return f.name
|
|
|
|
@mock.patch('six.moves.input')
|
|
@mock.patch('muranoclient.common.utils.Package.from_file')
|
|
def test_package_import_conflict_skip(self, from_file, raw_input_mock):
|
|
|
|
name = self._test_conflict(
|
|
self.package_mock,
|
|
from_file,
|
|
raw_input_mock,
|
|
's',
|
|
)
|
|
|
|
self.package_mock.create.assert_called_once_with({
|
|
'is_public': False,
|
|
}, {name: mock.ANY},)
|
|
|
|
@mock.patch('six.moves.input')
|
|
@mock.patch('muranoclient.common.utils.Package.from_file')
|
|
def test_package_import_conflict_skip_ea(self, from_file, raw_input_mock):
|
|
|
|
name = self._test_conflict(
|
|
self.package_mock,
|
|
from_file,
|
|
raw_input_mock,
|
|
'',
|
|
exists_action='s',
|
|
)
|
|
|
|
self.package_mock.create.assert_called_once_with({
|
|
'is_public': False,
|
|
}, {name: mock.ANY},)
|
|
self.assertFalse(raw_input_mock.called)
|
|
|
|
@mock.patch('six.moves.input')
|
|
@mock.patch('muranoclient.common.utils.Package.from_file')
|
|
def test_package_import_conflict_abort(self, from_file, raw_input_mock):
|
|
|
|
self.assertRaises(SystemExit, self._test_conflict,
|
|
self.package_mock,
|
|
from_file,
|
|
raw_input_mock,
|
|
'a',
|
|
)
|
|
|
|
self.package_mock.create.assert_called_once_with({
|
|
'is_public': False,
|
|
}, mock.ANY,)
|
|
|
|
@mock.patch('six.moves.input')
|
|
@mock.patch('muranoclient.common.utils.Package.from_file')
|
|
def test_package_import_conflict_abort_ea(self,
|
|
from_file, raw_input_mock):
|
|
|
|
self.assertRaises(SystemExit, self._test_conflict,
|
|
self.package_mock,
|
|
from_file,
|
|
raw_input_mock,
|
|
'',
|
|
exists_action='a',
|
|
)
|
|
|
|
self.package_mock.create.assert_called_once_with({
|
|
'is_public': False,
|
|
}, mock.ANY,)
|
|
self.assertFalse(raw_input_mock.called)
|
|
|
|
@mock.patch('six.moves.input')
|
|
@mock.patch('muranoclient.common.utils.Package.from_file')
|
|
def test_package_import_conflict_update(self, from_file, raw_input_mock):
|
|
|
|
name = self._test_conflict(
|
|
self.package_mock,
|
|
from_file,
|
|
raw_input_mock,
|
|
'u',
|
|
)
|
|
|
|
self.assertEqual(2, self.package_mock.create.call_count)
|
|
self.package_mock.delete.assert_called_once_with('test_id')
|
|
|
|
self.package_mock.create.assert_has_calls(
|
|
[
|
|
mock.call({'is_public': False}, {name: mock.ANY},),
|
|
mock.call({'is_public': False}, {name: mock.ANY},)
|
|
], any_order=True,
|
|
)
|
|
|
|
@mock.patch('six.moves.input')
|
|
@mock.patch('muranoclient.common.utils.Package.from_file')
|
|
def test_package_import_conflict_update_ea(self,
|
|
from_file, raw_input_mock):
|
|
|
|
name = self._test_conflict(
|
|
self.package_mock,
|
|
from_file,
|
|
raw_input_mock,
|
|
'',
|
|
exists_action='u',
|
|
)
|
|
|
|
self.assertEqual(2, self.package_mock.create.call_count)
|
|
self.package_mock.delete.assert_called_once_with('test_id')
|
|
|
|
self.package_mock.create.assert_has_calls(
|
|
[
|
|
mock.call({'is_public': False}, {name: mock.ANY},),
|
|
mock.call({'is_public': False}, {name: mock.ANY},)
|
|
], any_order=True,
|
|
)
|
|
self.assertFalse(raw_input_mock.called)
|
|
|
|
def _test_conflict_dep(self,
|
|
packages, from_file,
|
|
dep_exists_action=''):
|
|
packages.create = mock.MagicMock(
|
|
side_effect=[common_exceptions.HTTPConflict("Conflict"),
|
|
common_exceptions.HTTPConflict("Conflict"),
|
|
None])
|
|
|
|
packages.filter.return_value = [mock.Mock(id='test_id')]
|
|
|
|
pkg1 = make_pkg(
|
|
{'FullName': 'first_app', 'Require': {'second_app': '1.0'}, })
|
|
pkg2 = make_pkg({'FullName': 'second_app', })
|
|
|
|
def side_effect(name):
|
|
if 'first_app' in name:
|
|
return mc_utils.Package(mc_utils.File(pkg1))
|
|
if 'second_app' in name:
|
|
return mc_utils.Package(mc_utils.File(pkg2))
|
|
|
|
from_file.side_effect = side_effect
|
|
|
|
arglist = ['first_app', '--exists-action', 's',
|
|
'--dep-exists-action', dep_exists_action]
|
|
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
@mock.patch('muranoclient.common.utils.Package.from_file')
|
|
def test_package_import_conflict_dep_skip_ea(self, from_file):
|
|
self._test_conflict_dep(
|
|
self.package_mock,
|
|
from_file,
|
|
dep_exists_action='s',
|
|
)
|
|
|
|
self.assertEqual(2, self.package_mock.create.call_count)
|
|
self.package_mock.create.assert_has_calls(
|
|
[
|
|
mock.call({'is_public': False}, {'first_app': mock.ANY}),
|
|
mock.call({'is_public': False}, {'second_app': mock.ANY}),
|
|
], any_order=True,
|
|
)
|
|
|
|
@mock.patch('muranoclient.common.utils.Package.from_file')
|
|
def test_package_import_conflict_dep_abort_ea(self, from_file):
|
|
self.assertRaises(SystemExit, self._test_conflict_dep,
|
|
self.package_mock,
|
|
from_file,
|
|
dep_exists_action='a',
|
|
)
|
|
|
|
self.package_mock.create.assert_called_with({
|
|
'is_public': False,
|
|
}, {'second_app': mock.ANY},)
|
|
|
|
@mock.patch('muranoclient.common.utils.Package.from_file')
|
|
def test_package_import_conflict_dep_update_ea(self, from_file):
|
|
self._test_conflict_dep(
|
|
self.package_mock,
|
|
from_file,
|
|
dep_exists_action='u',
|
|
)
|
|
|
|
self.assertGreater(self.package_mock.create.call_count, 2)
|
|
self.assertLess(self.package_mock.create.call_count, 5)
|
|
|
|
self.assertTrue(self.package_mock.delete.called)
|
|
|
|
self.package_mock.create.assert_has_calls(
|
|
[
|
|
mock.call({'is_public': False}, {'first_app': mock.ANY}),
|
|
mock.call({'is_public': False}, {'second_app': mock.ANY}),
|
|
mock.call({'is_public': False}, {'second_app': mock.ANY}),
|
|
], any_order=True,
|
|
)
|
|
|
|
@mock.patch('muranoclient.common.utils.Package.from_file')
|
|
def test_package_import_no_categories(self, from_file):
|
|
with tempfile.NamedTemporaryFile() as f:
|
|
RESULT_PACKAGE = f.name
|
|
pkg = make_pkg({'FullName': RESULT_PACKAGE})
|
|
from_file.return_value = mc_utils.Package(mc_utils.File(pkg))
|
|
|
|
arglist = [RESULT_PACKAGE]
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
self.package_mock.create.assert_called_once_with(
|
|
{'is_public': False},
|
|
{RESULT_PACKAGE: mock.ANY},
|
|
)
|
|
|
|
@requests_mock.mock()
|
|
@mock.patch('muranoclient.common.utils.Package.from_file')
|
|
def test_package_import_url(self, rm, from_file):
|
|
filename = "http://127.0.0.1/test_package.zip"
|
|
|
|
pkg = make_pkg({'FullName': 'test_package'})
|
|
from_file.return_value = mc_utils.Package(mc_utils.File(pkg))
|
|
|
|
rm.get(filename, body=make_pkg({'FullName': 'test_package'}))
|
|
|
|
arglist = [filename]
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
self.package_mock.create.assert_called_once_with(
|
|
{'is_public': False},
|
|
{'test_package': mock.ANY},
|
|
)
|
|
|
|
@requests_mock.mock()
|
|
@mock.patch('muranoclient.common.utils.Package.from_file')
|
|
def test_package_import_by_name(self, rm, from_file):
|
|
filename = "io.test.apps.test_application"
|
|
murano_repo_url = "http://127.0.0.1"
|
|
|
|
pkg = make_pkg({'FullName': filename})
|
|
from_file.return_value = mc_utils.Package(mc_utils.File(pkg))
|
|
|
|
rm.get(murano_repo_url + '/apps/' + filename + '.zip',
|
|
body=make_pkg({'FullName': 'first_app'}))
|
|
|
|
arglist = [filename, '--murano-repo-url', murano_repo_url]
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
self.assertTrue(self.package_mock.create.called)
|
|
self.package_mock.create.assert_called_once_with(
|
|
{'is_public': False},
|
|
{filename: mock.ANY},
|
|
)
|
|
|
|
@requests_mock.mock()
|
|
def test_package_import_multiple(self, rm):
|
|
filename = ["io.test.apps.test_application",
|
|
"http://127.0.0.1/test_app2.zip", ]
|
|
murano_repo_url = "http://127.0.0.1"
|
|
|
|
rm.get(murano_repo_url + '/apps/' + filename[0] + '.zip',
|
|
body=make_pkg({'FullName': 'first_app'}))
|
|
|
|
rm.get(filename[1],
|
|
body=make_pkg({'FullName': 'second_app'}))
|
|
|
|
arglist = [filename[0], filename[1],
|
|
'--murano-repo-url', murano_repo_url]
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
self.assertEqual(2, self.package_mock.create.call_count)
|
|
|
|
self.package_mock.create.assert_has_calls(
|
|
[
|
|
mock.call({'is_public': False}, {'first_app': mock.ANY}),
|
|
mock.call({'is_public': False}, {'second_app': mock.ANY}),
|
|
], any_order=True,
|
|
)
|
|
|
|
|
|
class TestBundleImport(TestPackage):
|
|
def setUp(self):
|
|
super(TestBundleImport, self).setUp()
|
|
|
|
# Command to test
|
|
self.cmd = osc_pkg.ImportBundle(self.app, None)
|
|
|
|
@requests_mock.mock()
|
|
def test_import_bundle_by_name(self, m):
|
|
"""Asserts bundle import calls packages create once for each pkg."""
|
|
pkg1 = make_pkg({'FullName': 'first_app'})
|
|
pkg2 = make_pkg({'FullName': 'second_app'})
|
|
|
|
murano_repo_url = "http://127.0.0.1"
|
|
|
|
m.get(murano_repo_url + '/apps/first_app.zip', body=pkg1)
|
|
m.get(murano_repo_url + '/apps/second_app.1.0.zip',
|
|
body=pkg2)
|
|
s = six.StringIO()
|
|
bundle_contents = {'Packages': [
|
|
{'Name': 'first_app'},
|
|
{'Name': 'second_app', 'Version': '1.0'}
|
|
]}
|
|
json.dump(bundle_contents, s)
|
|
s = six.BytesIO(s.getvalue().encode('ascii'))
|
|
|
|
m.get(murano_repo_url + '/bundles/test_bundle.bundle',
|
|
body=s)
|
|
|
|
arglist = ["test_bundle", '--murano-repo-url', murano_repo_url]
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
self.package_mock.create.assert_has_calls(
|
|
[
|
|
mock.call({'is_public': False}, {'first_app': mock.ANY}),
|
|
mock.call({'is_public': False}, {'second_app': mock.ANY}),
|
|
], any_order=True,
|
|
)
|
|
|
|
@requests_mock.mock()
|
|
def test_import_bundle_wrong_url(self, m):
|
|
url = 'http://127.0.0.2/test_bundle.bundle'
|
|
m.get(url, status_code=404)
|
|
|
|
arglist = [url]
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
self.assertFalse(self.package_mock.packages.create.called)
|
|
|
|
@requests_mock.mock()
|
|
def test_import_bundle_no_bundle(self, m):
|
|
url = 'http://127.0.0.1/bundles/test_bundle.bundle'
|
|
m.get(url, status_code=404)
|
|
|
|
arglist = ["test_bundle"]
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
self.assertFalse(self.package_mock.packages.create.called)
|
|
|
|
@requests_mock.mock()
|
|
def test_import_bundle_by_url(self, m):
|
|
"""Asserts bundle import calls packages create once for each pkg."""
|
|
pkg1 = make_pkg({'FullName': 'first_app'})
|
|
pkg2 = make_pkg({'FullName': 'second_app'})
|
|
|
|
murano_repo_url = 'http://127.0.0.1'
|
|
m.get(murano_repo_url + '/apps/first_app.zip', body=pkg1)
|
|
m.get(murano_repo_url + '/apps/second_app.1.0.zip',
|
|
body=pkg2)
|
|
s = six.StringIO()
|
|
bundle_contents = {'Packages': [
|
|
{'Name': 'first_app'},
|
|
{'Name': 'second_app', 'Version': '1.0'}
|
|
]}
|
|
json.dump(bundle_contents, s)
|
|
s = six.BytesIO(s.getvalue().encode('ascii'))
|
|
|
|
url = 'http://127.0.0.2/test_bundle.bundle'
|
|
m.get(url, body=s)
|
|
|
|
arglist = [url, '--murano-repo-url', murano_repo_url]
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
self.package_mock.create.assert_has_calls(
|
|
[
|
|
mock.call({'is_public': False}, {'first_app': mock.ANY}),
|
|
mock.call({'is_public': False}, {'second_app': mock.ANY}),
|
|
], any_order=True,
|
|
)
|
|
|
|
@requests_mock.mock()
|
|
def test_import_local_bundle(self, m):
|
|
"""Asserts local bundles are first searched locally."""
|
|
tmp_dir = tempfile.mkdtemp()
|
|
bundle_file = os.path.join(tmp_dir, 'bundle.bundle')
|
|
with open(os.path.join(tmp_dir, 'bundle.bundle'), 'w') as f:
|
|
|
|
bundle_contents = {'Packages': [
|
|
{'Name': 'first_app'},
|
|
{'Name': 'second_app', 'Version': '1.0'}
|
|
]}
|
|
json.dump(bundle_contents, f)
|
|
|
|
pkg1 = make_pkg({'FullName': 'first_app',
|
|
'Require': {'third_app': None}})
|
|
pkg2 = make_pkg({'FullName': 'second_app'})
|
|
pkg3 = make_pkg({'FullName': 'third_app'})
|
|
with open(os.path.join(tmp_dir, 'first_app'), 'wb') as f:
|
|
f.write(pkg1.read())
|
|
with open(os.path.join(tmp_dir, 'third_app'), 'wb') as f:
|
|
f.write(pkg3.read())
|
|
|
|
murano_repo_url = "http://127.0.0.1"
|
|
m.get(murano_repo_url + '/apps/first_app.zip',
|
|
status_code=404)
|
|
m.get(murano_repo_url + '/apps/second_app.1.0.zip',
|
|
body=pkg2)
|
|
m.get(murano_repo_url + '/apps/third_app.zip',
|
|
status_code=404)
|
|
|
|
arglist = [bundle_file, '--murano-repo-url', murano_repo_url]
|
|
parsed_args = self.check_parser(self.cmd, arglist, [])
|
|
self.cmd.take_action(parsed_args)
|
|
|
|
self.package_mock.create.assert_has_calls(
|
|
[
|
|
mock.call({'is_public': False}, {'first_app': mock.ANY}),
|
|
mock.call({'is_public': False}, {'second_app': mock.ANY}),
|
|
mock.call({'is_public': False}, {'third_app': mock.ANY}),
|
|
], any_order=True,
|
|
)
|
|
shutil.rmtree(tmp_dir)
|
|
|
|
|
|
class TestShowPackage(TestPackage):
|
|
def setUp(self):
|
|
super(TestShowPackage, self).setUp()
|
|
|
|
# Command to test
|
|
self.cmd = osc_pkg.ShowPackage(self.app, None)
|
|
|
|
def test_package_show(self):
|
|
arglist = ['fake']
|
|
verifylist = [('id', 'fake')]
|
|
|
|
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
|
|
|
columns, data = self.cmd.take_action(parsed_args)
|
|
|
|
# Check that columns are correct
|
|
expected_columns = ('categories', 'class_definitions', 'description',
|
|
'enabled', 'fully_qualified_name', 'id',
|
|
'is_public', 'name', 'owner_id', 'tags', 'type')
|
|
self.assertEqual(expected_columns, columns)
|
|
|
|
self.package_mock.get.assert_called_with('fake')
|