}
-def _create_external_network(network_file):
+def _create_external_network(network_file, openrc=None):
network_conf = NetworkConfig(network_file=network_file)
ext_name = network_conf.ext_network_name
physnet = network_conf.ext_mapping if hasattr(network_conf, 'ext_mapping') else 'physnet1'
- neutronclient = neutron.Neutron()
+ neutronclient = neutron.Neutron(openrc=openrc)
ext_id = neutronclient.create_network(ext_name,
_config_external_network(ext_name, physnet))
if ext_id:
neutronclient.create_subnet(_config_external_subnet(ext_id, network_conf))
-def _create_flavor_m1_micro():
+def _create_flavor_m1_micro(openrc=None):
name = 'm1.micro'
- novaclient = nova.Nova()
+ novaclient = nova.Nova(openrc=openrc)
if not novaclient.get_flavor_by_name(name):
try:
return novaclient.create_flavor(name, ram=64, vcpus=1, disk=0)
return img
-def _create_image_TestVM():
- glanceclient = glance.Glance()
+def _create_image_TestVM(openrc=None):
+ glanceclient = glance.Glance(openrc=openrc)
image = 'TestVM'
if not glanceclient.get_by_name(image):
try:
return body
-def _create_security_group_rules():
- neutronclient = neutron.Neutron()
+def _create_security_group_rules(openrc=None):
+ neutronclient = neutron.Neutron(openrc=openrc)
try:
security_group_name = 'default'
security_group = neutronclient.get_security_group_by_name(security_group_name)
--- /dev/null
+##############################################################################
+# Copyright (c) 2017 ZTE Corp and others.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+##############################################################################
+import os
+
+import pytest
+import mock
+
+from deploy.post.glance import Glance
+from deploy.post import glance
+
+
+@pytest.fixture(scope="module")
+def openrc_conf_file_dir(data_root):
+ return os.path.join(data_root, 'openrc_conf')
+
+
+def test_create_Glance_instance(openrc_conf_file_dir):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ glance = Glance(openrc=openrc_file)
+ assert glance.controller == glance.client.images
+
+
+@mock.patch('glanceclient.v2.images.Controller.create')
+@mock.patch('glanceclient.v2.images.Controller.upload')
+def test_create_in_Glance(mock_upload, mock_create,
+ openrc_conf_file_dir, tmpdir):
+ class Test_image():
+ def __init__(self, id):
+ self.id = id
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ file_name = 'test_image.qcow2'
+ file_path = os.path.join(tmpdir.dirname, tmpdir.basename, file_name)
+ with open(file_path, 'w') as f:
+ f.write('test_data')
+ id = 0x1234
+ glance = Glance(openrc=openrc_file)
+ mock_create.return_value = Test_image(id)
+ ret = glance.create(file_name, file_path)
+ assert ret == id
+ tmpdir.remove()
+
+
+@mock.patch.object(glance.Glance, 'list')
+def test_get_by_name_in_Glance(mock_list, openrc_conf_file_dir):
+ class Test_image():
+ def __init__(self, name):
+ self.name = name
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ image_inst1 = Test_image('test_image1.qcow2')
+ image_inst2 = Test_image('test_image2.qcow2')
+ images_list = [image_inst1, image_inst2]
+ mock_list.return_value = images_list
+ glance = Glance(openrc=openrc_file)
+ ret = glance.get_by_name('test_image1.qcow2')
+ assert ret == image_inst1
+
+
+@mock.patch('glanceclient.v2.images.Controller.list')
+def test_list_in_Glance(mock_list, openrc_conf_file_dir):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ glance_list = ['test1']
+ mock_list.return_value = glance_list
+ glance = Glance(openrc=openrc_file)
+ ret = glance.list()
+ assert ret == glance_list
KeystoneClient = Keystoneauth(openrc)
ret_openrc_dict = KeystoneClient._parse_openrc()
assert expected == ret_openrc_dict
+
+
+@pytest.mark.parametrize('openrc_file_name', [
+ (
+ 'admin-openrc.sh'
+ )])
+def test__get_auth(openrc_conf_file_dir, openrc_file_name,):
+ openrc = os.path.join(openrc_conf_file_dir, openrc_file_name)
+ KeystoneClient = Keystoneauth(openrc)
+ assert KeystoneClient._get_auth()
--- /dev/null
+##############################################################################
+# Copyright (c) 2017 ZTE Corp and others.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+##############################################################################
+import os
+
+import pytest
+import mock
+
+from deploy.post import neutron
+from deploy.post.neutron import Neutron
+
+
+@pytest.fixture(scope="module")
+def openrc_conf_file_dir(data_root):
+ return os.path.join(data_root, 'openrc_conf')
+
+
+def test_create_Neutron_instance(openrc_conf_file_dir):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ neutron = Neutron(openrc=openrc_file)
+ assert neutron.openrc == openrc_file
+
+
+@pytest.mark.parametrize('ret_get_network_by_name', [
+ (None),
+ ({'name': 'network_test2'})])
+@mock.patch.object(neutron.Neutron, 'get_network_by_name')
+@mock.patch.object(neutron.Neutron, '_create_network')
+def test_create_network_in_Neutron(mock__create_network, mock_get_network_by_name,
+ ret_get_network_by_name, openrc_conf_file_dir):
+ net_name = 'network_test1'
+ net_body = {'name': 'network_test1'}
+ net_id = 0xabcd
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ mock_get_network_by_name.return_value = ret_get_network_by_name
+ mock__create_network.return_value = net_id
+ neutron = Neutron(openrc=openrc_file)
+ ret = neutron.create_network(net_name, net_body)
+ if ret_get_network_by_name is None:
+ assert ret == net_id
+ mock__create_network.asset_called_once_with(net_body)
+ else:
+ assert ret is None
+ mock__create_network.assert_not_called()
+
+
+@pytest.mark.parametrize('ret_get_subnet_by_name', [
+ (None),
+ ({'name': 'subnet_test2'})])
+@mock.patch.object(neutron.Neutron, 'get_subnet_by_name')
+@mock.patch.object(neutron.Neutron, '_create_subnet')
+def test_create_subnet_in_Neutron_no_exist(mock__create_subnet, mock_get_subnet_by_name,
+ ret_get_subnet_by_name, openrc_conf_file_dir):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ subnet_body = {'name': 'subnet_test1'}
+ subnet_id = 0xabcd
+ mock_get_subnet_by_name.return_value = ret_get_subnet_by_name
+ mock__create_subnet.return_value = subnet_id
+ neutron = Neutron(openrc=openrc_file)
+ ret = neutron.create_subnet(subnet_body)
+ if ret_get_subnet_by_name is None:
+ assert ret == subnet_id
+ mock__create_subnet.asset_called_once_with(subnet_body)
+ else:
+ assert ret is None
+ mock__create_subnet.assert_not_called()
+
+
+@mock.patch('neutronclient.v2_0.client.Client.list_networks')
+def test_list_networks_in_Neutron(mock_list_networks, openrc_conf_file_dir):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ network_list = [{'name': 'network_test1'}]
+ mock_list_networks.return_value = {'networks': network_list}
+ neutron = Neutron(openrc=openrc_file)
+ ret = neutron.list_networks()
+ assert ret == network_list
+ mock_list_networks.assert_called_once()
+
+
+@mock.patch('neutronclient.v2_0.client.Client.list_subnets')
+def test_list_subnets_in_Neutron(mock_list_subnets, openrc_conf_file_dir):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ subnet_list = [{'name': 'subnet_test1'}]
+ mock_list_subnets.return_value = {'subnets': subnet_list}
+ neutron = Neutron(openrc=openrc_file)
+ ret = neutron.list_subnets()
+ assert ret == subnet_list
+ mock_list_subnets.assert_called_once()
+
+
+@mock.patch.object(neutron.Neutron, 'list_networks')
+def test_get_network_by_name_in_Neutron(mock_list_networks, openrc_conf_file_dir):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ network_list = [{'name': 'network_test1'}, {'name': 'network_test2'}]
+ network_name = 'network_test1'
+ mock_list_networks.return_value = network_list
+ neutron = Neutron(openrc=openrc_file)
+ ret = neutron.get_network_by_name(network_name)
+ assert ret == {'name': 'network_test1'}
+ mock_list_networks.assert_called_once()
+
+
+@pytest.mark.parametrize('body, ret_list_subnets, expeced', [
+ (
+ {
+ 'subnets': [
+ {
+ 'name': 'ext_subnet',
+ 'ip_version': 4,
+ 'network_id': 0x1234
+ }
+ ]
+ },
+ [
+ {'name': 'ext_subnet', 'network_id': 0x1234},
+ {'name': 'inter_subnet', 'network_id': 0x2345}
+ ],
+ {'name': 'ext_subnet', 'network_id': 0x1234}
+ ),
+ (
+ {
+ 'subnets': [
+ {
+ 'name': 'ext_subnet',
+ 'ip_version': 4,
+ 'network_id': 0x1234
+ }
+ ]
+ },
+ [
+ {'name': 'admin_subnet', 'network_id': 0x1234},
+ {'name': 'inter_subnet', 'network_id': 0x2345}
+ ], None
+ )])
+@mock.patch.object(neutron.Neutron, 'list_subnets')
+def test_get_subnet_by_name_in_Neutron(mock_list_subnets, body,
+ ret_list_subnets, expeced,
+ openrc_conf_file_dir):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ mock_list_subnets.return_value = ret_list_subnets
+ neutron = Neutron(openrc=openrc_file)
+ ret = neutron.get_subnet_by_name(body)
+ assert ret == expeced
+ mock_list_subnets.assert_called_once()
+
+
+@mock.patch('neutronclient.v2_0.client.Client.create_network')
+def test__create_network_in_Neutron_no_exist(mock_create_network, openrc_conf_file_dir):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ nid = 0x1234
+ name = 'ext'
+ body = {
+ 'network': {
+ 'name': 'ext',
+ 'admin_state_up': True,
+ 'shared': False,
+ 'provider:network_type': 'flat',
+ 'provider:physical_network': 'physnet1',
+ 'router:external': True
+ }
+ }
+ ret_net_info = {
+ 'network':
+ {
+ 'id': 0x1234
+ }
+ }
+ mock_create_network.return_value = ret_net_info
+ neutron = Neutron(openrc=openrc_file)
+ ret = neutron._create_network(name, body)
+ assert ret == nid
+ mock_create_network.assert_called_once_with(body=body)
+
+
+@mock.patch('neutronclient.v2_0.client.Client.create_subnet')
+def test__create_subnet_in_Neutron_no_exist(mock_create_subnet, openrc_conf_file_dir):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ snid = 0xabcd
+ body = {
+ 'subnets': {
+ 'name': 'admin_external_subnet',
+ 'cidr': '172.70.0.0/24',
+ 'gateway_ip': '172.70.0.1',
+ 'allocation_pools': [{'start': '172.70.0.2', 'end': '172.70.0.100'}],
+ 'enable_dhcp': False
+ }
+ }
+ ret_subnet_info = {
+ 'subnets': [
+ {
+ 'name': 'admin_external_subnet',
+ 'cidr': '172.70.0.0/24',
+ 'ip_version': 4,
+ 'network_id': 0x1234,
+ 'gateway_ip': '172.70.0.1',
+ 'allocation_pools': [{'start': '172.70.0.2', 'end': '172.70.0.100'}],
+ 'enable_dhcp': False,
+ 'id': 0xabcd
+ }
+ ]
+ }
+ mock_create_subnet.return_value = ret_subnet_info
+ neutron = Neutron(openrc=openrc_file)
+ ret = neutron._create_subnet(body)
+ assert ret == snid
+ mock_create_subnet.assert_called_once_with(body)
+
+
+@mock.patch('neutronclient.v2_0.client.Client.list_security_groups')
+def test__list_security_groups_in_Neutron(mock_list_security_groups, openrc_conf_file_dir):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ ret_security_groups = {
+ 'security_groups':
+ [
+ {
+ 'name': 'default'
+ }
+ ]
+ }
+ security_groups_list = [
+ {
+ 'name': 'default'
+ }
+ ]
+ mock_list_security_groups.return_value = ret_security_groups
+ neutron = Neutron(openrc=openrc_file)
+ ret = neutron._list_security_groups()
+ assert ret == security_groups_list
+ mock_list_security_groups.assert_called_once_with()
+
+
+@mock.patch.object(neutron.Neutron, '_list_security_groups')
+def test_get_security_group_by_name_in_Neutron(mock__list_security_groups, openrc_conf_file_dir):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ security_group_1 = {'name': 'default', 'security_group_id': 0x1234}
+ security_group_2 = {'name': 'security_1', 'security_group_id': 0x2345}
+ security_group_list = [security_group_1, security_group_2]
+ mock__list_security_groups.return_value = security_group_list
+ neutron = Neutron(openrc=openrc_file)
+ ret = neutron.get_security_group_by_name(security_group_1['name'])
+ assert ret == security_group_1
+ mock__list_security_groups.assert_called_once_with()
+
+
+@pytest.mark.parametrize('security_group, body, expected', [
+ (
+ {
+ 'security_group_rules':
+ [
+ {
+ 'direction': 'ingress',
+ 'ethertype': 'IPv4',
+ 'protocol': 'tcp',
+ 'port_range_min': 22,
+ 'port_range_max': 22,
+ 'remote_ip_prefix': '0.0.0.0/0',
+ },
+ {
+ 'direction': 'ingress',
+ 'ethertype': 'IPv4',
+ 'protocol': 'icmp',
+ 'remote_ip_prefix': '0.0.0.0/0',
+ }
+ ],
+ 'id': 0x1234
+ },
+ {
+ 'security_group_rule':
+ {
+ 'direction': 'ingress',
+ 'ethertype': 'IPv4',
+ 'protocol': 'tcp',
+ 'port_range_min': 22,
+ 'port_range_max': 22,
+ 'remote_ip_prefix': '0.0.0.0/0',
+ }
+ }, True
+ ),
+ (
+ {
+ 'security_group_rules':
+ [
+ {
+ 'direction': 'ingress',
+ 'ethertype': 'IPv4',
+ 'protocol': 'icmp',
+ 'remote_ip_prefix': '0.0.0.0/0',
+ }
+ ],
+ 'id': 0x1234
+ },
+ {
+ 'security_group_rule':
+ {
+ 'direction': 'ingress',
+ 'ethertype': 'IPv4',
+ 'protocol': 'tcp',
+ 'port_range_min': 22,
+ 'port_range_max': 22,
+ 'remote_ip_prefix': '0.0.0.0/0',
+ }
+ }, False
+ )])
+def test__check_security_group_rule_conflict(security_group, body,
+ expected, openrc_conf_file_dir):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ neutron = Neutron(openrc=openrc_file)
+ ret = neutron._check_security_group_rule_conflict(security_group, body)
+ assert ret == expected
+
+
+@pytest.mark.parametrize('ret__check_security_group_rule_conflict', [
+ (True),
+ (False)])
+@mock.patch('neutronclient.v2_0.client.Client.create_security_group_rule')
+@mock.patch.object(neutron.Neutron, '_check_security_group_rule_conflict')
+def test_create_security_group_rule_in_Neutron(mock__check_security_group_rule_conflict,
+ mock_create_security_group_rule,
+ ret__check_security_group_rule_conflict,
+ openrc_conf_file_dir):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ security_group = {
+ 'security_group_rules':
+ [
+ {
+ 'direction': 'ingress',
+ 'ethertype': 'IPv4',
+ 'protocol': 'tcp',
+ 'port_range_min': 22,
+ 'port_range_max': 22,
+ 'remote_ip_prefix': '0.0.0.0/0',
+ },
+ {
+ 'direction': 'ingress',
+ 'ethertype': 'IPv4',
+ 'protocol': 'icmp',
+ 'remote_ip_prefix': '0.0.0.0/0',
+ }
+ ],
+ 'id': 0x1234
+ }
+ body = {
+ 'security_group_rule':
+ {
+ 'direction': 'ingress',
+ 'ethertype': 'IPv4',
+ 'protocol': 'tcp',
+ 'port_range_min': 22,
+ 'port_range_max': 22,
+ 'remote_ip_prefix': '0.0.0.0/0',
+ }
+ }
+ rule = {
+ 'security_group_rule':
+ {
+ 'id': 0x1234
+ }
+ }
+ mock__check_security_group_rule_conflict.return_value = ret__check_security_group_rule_conflict
+ mock_create_security_group_rule.return_value = rule
+ neutron = Neutron(openrc=openrc_file)
+ neutron.create_security_group_rule(security_group, body)
+ if ret__check_security_group_rule_conflict is False:
+ mock_create_security_group_rule.assert_called_once_with(body=body)
+ else:
+ mock_create_security_group_rule.assert_not_called()
--- /dev/null
+##############################################################################
+# Copyright (c) 2017 ZTE Corp and others.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+##############################################################################
+import os
+
+import pytest
+import mock
+
+from deploy.post import nova
+from deploy.post.nova import Nova
+
+
+@pytest.fixture(scope="module")
+def openrc_conf_file_dir(data_root):
+ return os.path.join(data_root, 'openrc_conf')
+
+
+def test_create_Nova_instance(openrc_conf_file_dir):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ nova = Nova(openrc=openrc_file)
+ assert nova.flavors == nova.client.flavors
+
+
+@mock.patch('novaclient.v2.flavors.FlavorManager.create')
+def test_create_flavor_in_Glance(mock_create, openrc_conf_file_dir):
+ class Test_flavor():
+ def __init__(self, id):
+ self.id = id
+ flavor_conf = {
+ 'name': 'flavor_test',
+ 'ram': 64,
+ 'vcpus': 1,
+ 'disk': 1,
+ 'is_public': True
+ }
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ id = 0x1234
+ nova = Nova(openrc=openrc_file)
+ mock_create.return_value = Test_flavor(id)
+ ret = nova.create_flavor(flavor_conf['name'], flavor_conf['ram'],
+ flavor_conf['vcpus'], flavor_conf['disk'],
+ is_public=flavor_conf['is_public'])
+ assert ret == id
+ mock_create.assert_called_once_with(flavor_conf['name'], flavor_conf['ram'],
+ flavor_conf['vcpus'], flavor_conf['disk'],
+ is_public=flavor_conf['is_public'])
+
+
+@mock.patch.object(nova.Nova, 'list_flavors')
+def test_get_flavor_by_name_in_Nova(mock_list_flavors, openrc_conf_file_dir):
+ class Test_flavor():
+ def __init__(self, name):
+ self.name = name
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ flavor_inst1 = Test_flavor('test_flavor1')
+ flavor_inst2 = Test_flavor('test_flavor2')
+ flavors_list = [flavor_inst1, flavor_inst2]
+ flavor_name = 'test_flavor2'
+ mock_list_flavors.return_value = flavors_list
+ nova = Nova(openrc=openrc_file)
+ ret = nova.get_flavor_by_name(flavor_name)
+ assert ret == flavor_inst2
+ mock_list_flavors.assert_called_once_with()
+
+
+@mock.patch('novaclient.v2.flavors.FlavorManager.list')
+def test_list_flavors_in_Nova(mock_list, openrc_conf_file_dir):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ flavor_list = ['test1']
+ mock_list.return_value = flavor_list
+ nova = Nova(openrc=openrc_file)
+ ret = nova.list_flavors()
+ assert ret == flavor_list
+ mock_list.assert_called_once_with(detailed=True)
from deepdiff import DeepDiff
import pytest
+import mock
+from deploy.post import execute
from deploy.post.execute import (
_config_external_network,
+ _config_external_subnet,
_config_icmp_security_group_rule,
_config_ssh_security_group_rule,
- _config_kolla_admin_openrc
+ _config_kolla_admin_openrc,
+ _create_external_network,
+ _create_flavor_m1_micro,
+ _prepare_cirros,
+ _create_image_TestVM,
+ _create_security_group_rules
)
assert _config_external_network(ext_name, physnet) == expected
+@pytest.mark.parametrize('ext_id, network_conf, expected', [
+ (0x1234,
+ {
+ 'ext_network_name': 'admin_external',
+ 'ext_cidr': '172.70.0.0/24',
+ 'ext_gateway': '172.70.0.1',
+ 'ext_ip_ranges': [{'start': '172.70.0.2', 'end': '172.70.0.100'}]
+ },
+ {
+ 'subnets': [
+ {
+ 'name': 'admin_external_subnet',
+ 'cidr': '172.70.0.0/24',
+ 'ip_version': 4,
+ 'network_id': 0x1234,
+ 'gateway_ip': '172.70.0.1',
+ 'allocation_pools': [{'start': '172.70.0.2', 'end': '172.70.0.100'}],
+ 'enable_dhcp': False
+ }
+ ]
+ })])
+def test__config_external_subnet(ext_id, network_conf, expected):
+ class Network_conf():
+ def __init__(self, network_name, cidr, gateway, ip_ranges):
+ self.ext_network_name = network_name
+ self.ext_cidr = cidr
+ self.ext_gateway = gateway
+ self.ext_ip_ranges = ip_ranges
+
+ network_conf_inst = Network_conf(network_conf['ext_network_name'],
+ network_conf['ext_cidr'],
+ network_conf['ext_gateway'],
+ network_conf['ext_ip_ranges'])
+ assert _config_external_subnet(ext_id, network_conf_inst) == expected
+
+
@pytest.mark.parametrize('security_group_id, expected', [
('0x1111',
{
for val in diff['iterable_item_added'].values():
assert 'export SDN_CONTROLLER_IP' in val
tmpdir.remove()
+
+
+@pytest.fixture(scope="module")
+def net_conf_file_dir(data_root):
+ return os.path.join(data_root, 'lab_conf')
+
+
+@mock.patch.object(execute.neutron.Neutron, 'create_network')
+@mock.patch.object(execute.neutron.Neutron, 'create_subnet')
+def test__create_external_network_with_openrc(mock_create_subnet, mock_create_network,
+ net_conf_file_dir, openrc_conf_file_dir):
+ external_network_info = {
+ 'network': {
+ 'name': 'admin_external',
+ 'admin_state_up': True,
+ 'shared': False,
+ 'provider:network_type': 'flat',
+ 'provider:physical_network': 'physnet1',
+ 'router:external': True
+ }
+ }
+ external_subnet_info = {
+ 'subnets': [
+ {
+ 'name': 'admin_external_subnet',
+ 'cidr': '172.70.0.0/24',
+ 'ip_version': 4,
+ 'network_id': 0x1234,
+ 'gateway_ip': '172.70.0.1',
+ 'allocation_pools': [{'start': '172.70.0.2', 'end': '172.70.0.100'}],
+ 'enable_dhcp': False
+ }
+ ]
+ }
+ network_file = os.path.join(net_conf_file_dir, 'network_baremetal.yml')
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ mock_create_network.return_value = 0x1234
+ mock_create_subnet.return_value = 0xabcd
+ _create_external_network(network_file, openrc_file)
+ mock_create_network.assert_called_once_with('admin_external', external_network_info)
+ mock_create_subnet.assert_called_once_with(external_subnet_info)
+
+
+@mock.patch.object(execute.neutron.Neutron, 'create_network')
+@mock.patch.object(execute.neutron.Neutron, 'create_subnet')
+@mock.patch('deploy.post.execute.neutron.Neutron')
+def test__create_external_network_without_openrc(mock_Neutron, mock_create_subnet, mock_create_network,
+ net_conf_file_dir):
+ network_file = os.path.join(net_conf_file_dir, 'network_baremetal.yml')
+ mock_create_network.return_value = 0x1234
+ mock_create_subnet.return_value = 0xabcd
+ _create_external_network(network_file)
+ mock_create_network.assert_not_called()
+ mock_create_subnet.assert_not_called()
+
+
+@pytest.mark.parametrize('flavor,', [
+ (None),
+ ({'name': 'm1.micro', 'ram': 64})])
+@mock.patch.object(execute.nova.Nova, 'get_flavor_by_name')
+@mock.patch.object(execute.nova.Nova, 'create_flavor')
+def test__create_flavor_m1_micro(mock_create_flavor, mock_get_flavor_by_name,
+ openrc_conf_file_dir, flavor):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ mock_create_flavor.return_value = True
+ mock_get_flavor_by_name.return_value = flavor
+ _create_flavor_m1_micro(openrc_file)
+ mock_get_flavor_by_name.assert_called_once_with('m1.micro')
+ if flavor is None:
+ mock_create_flavor.assert_called_once_with('m1.micro', ram=64, vcpus=1, disk=0)
+ else:
+ mock_create_flavor.assert_not_called()
+
+
+@mock.patch('deploy.post.execute.os.system')
+def test__prepare_cirros(mock_system):
+ mock_system.return_value = 0
+ _prepare_cirros()
+ assert mock_system.call_count == 3
+
+
+@pytest.mark.parametrize('image_path, image_info', [
+ (None, None),
+ ('/var/lib/daisy/images/cirros-0.3.5-x86_64-disk.img', None),
+ (None, {'name': 'TestVM'}),
+ ('/var/lib/daisy/images/cirros-0.3.5-x86_64-disk.img', {'name': 'TestVM'})])
+@mock.patch('deploy.post.execute._prepare_cirros')
+@mock.patch('deploy.post.execute.err_exit')
+@mock.patch.object(execute.glance.Glance, 'create')
+@mock.patch.object(execute.glance.Glance, 'get_by_name')
+def test__create_image_TestVM(mock_get_by_name, mock_create,
+ mock_err_exit, mock___prepare_cirros,
+ image_path, image_info,
+ openrc_conf_file_dir):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ mock_get_by_name.return_value = image_info
+ mock_create.return_value = 0x1234
+ mock___prepare_cirros.return_value = image_path
+ _create_image_TestVM(openrc_file)
+ mock_get_by_name.assert_called_once_with('TestVM')
+ if image_info:
+ mock___prepare_cirros.assert_not_called()
+ else:
+ mock___prepare_cirros.assert_called_once_with()
+ if image_path:
+ mock_create.assert_called_once_with('TestVM', image_path)
+ else:
+ mock_err_exit.assert_called_once_with("Test image preparation failed")
+
+
+@mock.patch.object(execute.neutron.Neutron, 'create_security_group_rule')
+@mock.patch.object(execute.neutron.Neutron, 'get_security_group_by_name')
+def test__create_security_group_rules(mock_get_security_group_by_name, mock_create_security_group_rule,
+ openrc_conf_file_dir):
+ openrc_file = os.path.join(openrc_conf_file_dir, 'admin-openrc.sh')
+ mock_get_security_group_by_name.return_value = {'id': 0xaaaa}
+ _create_security_group_rules(openrc_file)
+ mock_get_security_group_by_name.assert_called_once_with('default')
+ assert mock_create_security_group_rule.call_count == 2
import os
import pytest
-
+import mock
import deploy.prepare.execute
from deploy.prepare.execute import (
_set_qemu_compute,
_set_default_floating_pool,
- _set_trusts_auth
+ _set_trusts_auth,
+ _make_dirs,
+ _write_conf_file
)
deploy.prepare.execute.KOLLA_CONF_PATH = '/tmp'
exp_conf_file_2 = os.path.join(kolla_conf_file_heat_dir, 'heat-engine.conf')
assert (os.path.isfile(exp_conf_file_1) and os.path.isfile(exp_conf_file_2))
clear_tmp_dir(kolla_conf_file_heat_dir)
+
+
+@pytest.mark.parametrize('ret_isdir', [
+ (True),
+ (False)])
+@mock.patch('os.path.isdir')
+@mock.patch('os.makedirs')
+def test__make_dirs(mock_makedirs, mock_isdir, ret_isdir):
+ path = '/tmp/test'
+ mock_makedirs.return_value = True
+ mock_isdir.return_value = ret_isdir
+ _make_dirs(path)
+ if ret_isdir is False:
+ mock_makedirs.assert_called_once_with(path, mode=0744)
+ else:
+ mock_makedirs.assert_not_called()
+
+
+def test__write_conf_file(tmpdir):
+ conf_file = os.path.join(tmpdir.dirname, tmpdir.basename, 'test_conf')
+ conf = 'conf_key1 = value1'
+ _write_conf_file(conf_file, conf)
+ with open(conf_file) as f:
+ data = f.readline().rstrip('\n')
+ assert data == conf
+ tmpdir.remove()
import argparse
import os
import sys
+import shutil
import yaml
+from deploy.utils import WORKSPACE
import mock
sys.modules['libvirt'] = mock.Mock()
+from deploy import environment # noqa: ignore=E402
from deploy.deploy import (
config_arg_parser,
- DaisyDeployment
+ DaisyDeployment,
+ parse_arguments,
+ main
) # noqa: ignore=E402
return os.path.join(data_root, 'lab_conf')
-@pytest.mark.parametrize('kwargs, expect_dasiy_info', [
+@pytest.mark.parametrize('kwargs, check_err, expect_dasiy_info', [
({'lab_name': 'zte',
'pod_name': 'virtual1',
'deploy_file': 'deploy_virtual1.yml',
'pxe_bridge': 'pxebr',
'deploy_log': 'deploy.log',
'scenario': 'os-nosdn-nofeature-ha'},
+ [],
{'name': 'daisy',
'image': 'daisy.qcow2',
'address': '10.20.11.2',
'net_file': 'network_baremetal.yml',
'bin_file': 'opnfv.bin',
'daisy_only': False,
- 'cleanup_only': False,
+ 'cleanup_only': True,
'remote_dir': '/home/daisy',
'work_dir': 'workdir',
'storage_dir': 'vms',
'pxe_bridge': 'pxebr',
'deploy_log': 'deploy.log',
'scenario': 'os-odl-nofeature-ha'},
+ ['disk_size invalid'],
{'name': 'daisy',
'image': 'daisy.qcow2',
'address': '10.20.0.2',
'gateway': '10.20.0.1',
'password': 'r00tme',
'disk_size': 50})])
-def test_create_DaisyDeployment_instance(kwargs, expect_dasiy_info, conf_file_dir, tmpdir):
+@mock.patch('deploy.deploy.deploy_schema_validate')
+@mock.patch('deploy.deploy.err_exit')
+def test_create_DaisyDeployment_instance(mock_err_exit, mock_deploy_schema_validate,
+ kwargs, expect_dasiy_info,
+ conf_file_dir, tmpdir,
+ check_err):
kwargs['deploy_file'] = os.path.join(conf_file_dir, kwargs['deploy_file'])
kwargs['net_file'] = os.path.join(conf_file_dir, kwargs['net_file'])
tmpdir.join(kwargs['bin_file']).write('testdata')
kwargs['work_dir'] = os.path.join(tmpsubdir.dirname, tmpsubdir.basename)
tmpsubdir = tmpdir.mkdir(kwargs['storage_dir'])
kwargs['storage_dir'] = os.path.join(tmpsubdir.dirname, tmpsubdir.basename)
+ mock_deploy_schema_validate.return_value = check_err
deploy = DaisyDeployment(**kwargs)
assert (deploy.lab_name, deploy.pod_name, deploy.src_deploy_file, deploy.net_file, deploy.bin_file,
(kwargs['lab_name'], kwargs['pod_name'], kwargs['deploy_file'], kwargs['net_file'],
kwargs['bin_file'], kwargs['daisy_only'], kwargs['cleanup_only'], kwargs['remote_dir'],
kwargs['work_dir'], kwargs['storage_dir'], kwargs['deploy_log'], kwargs['scenario'])
+ if check_err:
+ mock_err_exit.assert_called_once_with('Configuration deploy.yml check failed!')
+ else:
+ mock_err_exit.assert_not_called()
assert deploy.deploy_file_name == 'final_deploy.yml'
assert deploy.deploy_file == os.path.join(deploy.work_dir, 'final_deploy.yml')
expect_dasiy_info['image'] = os.path.join(kwargs['storage_dir'], expect_dasiy_info['image'])
assert deploy.daisy_server_info == expect_dasiy_info
+ tmpdir.remove()
+
+
+@pytest.mark.parametrize('kwargs', [
+ (
+ {
+ 'lab_name': 'zte',
+ 'pod_name': 'virtual1',
+ 'deploy_file': 'deploy_baremetal.yml',
+ 'net_file': 'network_baremetal.yml',
+ 'bin_file': 'opnfv.bin',
+ 'daisy_only': False,
+ 'cleanup_only': False,
+ 'remote_dir': '/home/daisy',
+ 'work_dir': 'workdir',
+ 'storage_dir': 'vms',
+ 'pxe_bridge': 'pxebr',
+ 'deploy_log': 'deploy.log',
+ 'scenario': 'os-nosdn-nofeature-ha'
+ }
+ ),
+ (
+ {
+ 'lab_name': 'zte',
+ 'pod_name': 'pod1',
+ 'deploy_file': 'deploy_baremetal.yml',
+ 'net_file': 'network_baremetal.yml',
+ 'bin_file': 'opnfv.bin',
+ 'daisy_only': False,
+ 'cleanup_only': True,
+ 'remote_dir': '/home/daisy',
+ 'work_dir': 'workdir',
+ 'storage_dir': 'vms',
+ 'pxe_bridge': 'pxebr',
+ 'deploy_log': 'deploy.log',
+ 'scenario': 'os-odl-nofeature-ha'
+ }
+ ),
+ (
+ {
+ 'lab_name': 'zte',
+ 'pod_name': 'pod1',
+ 'deploy_file': 'deploy_baremetal.yml',
+ 'net_file': 'network_baremetal.yml',
+ 'bin_file': 'opnfv.bin',
+ 'daisy_only': True,
+ 'cleanup_only': False,
+ 'remote_dir': '/home/daisy',
+ 'work_dir': 'workdir',
+ 'storage_dir': 'vms',
+ 'pxe_bridge': 'pxebr',
+ 'deploy_log': 'deploy.log',
+ 'scenario': 'os-odl-nofeature-ha'
+ }
+ )])
+@mock.patch.object(environment.BareMetalEnvironment, 'delete_old_environment')
+@mock.patch.object(environment.BareMetalEnvironment, 'create_daisy_server')
+@mock.patch.object(environment.BareMetalEnvironment, 'install_daisy')
+@mock.patch.object(environment.BareMetalEnvironment, 'deploy')
+def test_run_in_DaisyDeployment(mock_deploy, mock_install_daisy,
+ mock_create_daisy_server, mock_delete_old_environment,
+ conf_file_dir, tmpdir, kwargs):
+ kwargs['deploy_file'] = os.path.join(conf_file_dir, kwargs['deploy_file'])
+ kwargs['net_file'] = os.path.join(conf_file_dir, kwargs['net_file'])
+ tmpdir.join(kwargs['bin_file']).write('testdata')
+ kwargs['bin_file'] = os.path.join(tmpdir.dirname, tmpdir.basename, kwargs['bin_file'])
+ kwargs['deploy_log'] = os.path.join(tmpdir.dirname, tmpdir.basename, kwargs['deploy_log'])
+ tmpsubdir = tmpdir.mkdir(kwargs['work_dir'])
+ kwargs['work_dir'] = os.path.join(tmpsubdir.dirname, tmpsubdir.basename)
+ tmpsubdir = tmpdir.mkdir(kwargs['storage_dir'])
+ kwargs['storage_dir'] = os.path.join(tmpsubdir.dirname, tmpsubdir.basename)
+ daisy_deploy = DaisyDeployment(**kwargs)
+ daisy_deploy.run()
+ mock_delete_old_environment.asser_called_once_with()
+ if daisy_deploy.cleanup_only is False:
+ mock_create_daisy_server.assert_called_once_with()
+ if daisy_deploy.daisy_only is False:
+ mock_deploy.assert_called_once_with(daisy_deploy.deploy_file, daisy_deploy.net_file)
+ mock_install_daisy.assert_called_once_with(daisy_deploy.remote_dir, daisy_deploy.bin_file,
+ daisy_deploy.deploy_file_name, daisy_deploy.net_file_name)
+ else:
+ mock_deploy.assert_not_called()
+ mock_install_daisy.assert_not_called()
+ else:
+ mock_create_daisy_server.assert_not_called()
+ tmpdir.remove()
+
+
+@mock.patch('deploy.deploy.argparse.ArgumentParser.parse_args')
+@mock.patch('deploy.deploy.check_sudo_privilege')
+@mock.patch('deploy.deploy.save_log_to_file')
+@mock.patch('deploy.deploy.check_scenario_valid')
+@mock.patch('deploy.deploy.check_file_exists')
+@mock.patch('deploy.deploy.make_file_executable')
+@mock.patch('deploy.deploy.confirm_dir_exists')
+def test_parse_arguments(mock_confirm_dir_exists, mock_make_file_executable,
+ mock_check_file_exists, mock_check_scenario_valid,
+ mock_save_log_to_file, mock_check_sudo_privilege,
+ mock_parse_args, tmpdir):
+ class MockArg():
+ def __init__(self, lab_name, pod_name, bin_file, daisy_only,
+ cleanup_only, remote_dir, work_dir, storage_dir, pxe_bridge,
+ deploy_log, scenario):
+ self.lab_name = lab_name
+ self.pod_name = pod_name
+ self.bin_file = bin_file
+ self.daisy_only = daisy_only
+ self.cleanup_only = cleanup_only
+ self.remote_dir = remote_dir
+ self.work_dir = work_dir
+ self.storage_dir = storage_dir
+ self.pxe_bridge = pxe_bridge
+ self.deploy_log = deploy_log
+ self.scenario = scenario
+
+ bin_file_path = os.path.join(tmpdir.dirname, tmpdir.basename, 'opnfv.bin')
+ deploy_log_path = os.path.join(tmpdir.dirname, tmpdir.basename, 'deploy.log')
+ conf_base_dir = os.path.join(WORKSPACE, 'labs', 'zte', 'pod2')
+ deploy_file = os.path.join(conf_base_dir, 'daisy/config/deploy.yml')
+ net_file = os.path.join(conf_base_dir, 'daisy/config/network.yml')
+ cleanup_only = False
+ expected = {
+ 'lab_name': 'zte',
+ 'pod_name': 'pod2',
+ 'deploy_file': deploy_file,
+ 'net_file': net_file,
+ 'bin_file': bin_file_path,
+ 'daisy_only': False,
+ 'cleanup_only': cleanup_only,
+ 'remote_dir': '/home/daisy',
+ 'work_dir': '/tmp/workdir',
+ 'storage_dir': '/home/qemu/vms',
+ 'pxe_bridge': 'pxebr',
+ 'deploy_log': deploy_log_path,
+ 'scenario': 'os-nosdn-nofeature-noha'
+ }
+ mockarg = MockArg('zte', 'pod2', bin_file_path, False, cleanup_only, '/home/daisy', '/tmp/workdir',
+ '/home/qemu/vms', 'pxebr', deploy_log_path, 'os-nosdn-nofeature-noha')
+ mock_parse_args.return_value = mockarg
+ ret = parse_arguments()
+ assert ret == expected
+ mock_check_sudo_privilege.assert_called_once_with()
+ mock_save_log_to_file.assert_called_once_with(deploy_log_path)
+ mock_check_scenario_valid.assert_called_once_with('os-nosdn-nofeature-noha')
+ if cleanup_only is False:
+ mock_check_file_exists.assert_called_once_with(net_file)
+ mock_make_file_executable.assert_called_once_with(bin_file_path)
+ assert mock_confirm_dir_exists.call_count == 2
+ tmpdir.remove()
+
+@pytest.mark.parametrize('cleanup_only', [
+ (False), (True)])
+@mock.patch.object(DaisyDeployment, 'run')
+@mock.patch('deploy.deploy.parse_arguments')
+def test_main(mock_parse_arguments, mock_run, cleanup_only, tmpdir, conf_file_dir):
+ bin_file_path = os.path.join(tmpdir.dirname, tmpdir.basename, 'opnfv.bin')
+ with open(bin_file_path, 'w') as f:
+ f.write('test_data')
+ deploy_log_path = os.path.join(tmpdir.dirname, tmpdir.basename, 'deploy.log')
+ deploy_file = os.path.join(tmpdir.dirname, tmpdir.basename, 'deploy.yml')
+ src_deploy_file = os.path.join(conf_file_dir, 'deploy_baremetal.yml')
+ shutil.copyfile(src_deploy_file, deploy_file)
+ net_file = os.path.join(tmpdir.dirname, tmpdir.basename, 'network.yml')
+ src_network_file = os.path.join(conf_file_dir, 'network_baremetal.yml')
+ shutil.copyfile(src_network_file, net_file)
+ work_dir = os.path.join(tmpdir.dirname, tmpdir.basename)
+ kwargs = {
+ 'lab_name': 'zte',
+ 'pod_name': 'pod2',
+ 'deploy_file': deploy_file,
+ 'net_file': net_file,
+ 'bin_file': bin_file_path,
+ 'daisy_only': False,
+ 'cleanup_only': cleanup_only,
+ 'remote_dir': '/home/daisy',
+ 'work_dir': work_dir,
+ 'storage_dir': '/home/qemu/vms',
+ 'pxe_bridge': 'pxebr',
+ 'deploy_log': deploy_log_path,
+ 'scenario': 'os-nosdn-nofeature-noha'
+ }
+ mock_parse_arguments.return_value = kwargs
+ main()
+ mock_parse_arguments.assert_called_once_with()
+ mock_run.assert_called_once_with()
tmpdir.remove()
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
import os
-import mock
import pytest
import mock
Code Review / daisy.git / commitdiff