Refactor - move common methods to utils module
Move some methods and constants to the utils module so that the code can be shared. Related: blueprint network-data-v2-ports Change-Id: I11c5ef7911511f88ba82d7a7468dca76201a6dc2
This commit is contained in:
parent
11c724b3cb
commit
bfa5dabc99
|
@ -20,6 +20,13 @@ import ipaddress
|
|||
import jsonschema
|
||||
import yaml
|
||||
|
||||
RES_ID = 'physical_resource_id'
|
||||
TYPE_NET = 'OS::Neutron::Net'
|
||||
TYPE_SUBNET = 'OS::Neutron::Subnet'
|
||||
RES_TYPE = 'resource_type'
|
||||
TYPE_SEGMENT = 'OS::Neutron::Segment'
|
||||
NET_VIP_SUFFIX = '_virtual_ip'
|
||||
|
||||
DOMAIN_NAME_REGEX = (r'^(?=^.{1,255}$)(?!.*\.\..*)(.{1,63}\.)'
|
||||
r'+(.{0,63}\.?)|(?!\.)(?!.*\.\..*)(^.{1,63}$)'
|
||||
r'|(^\.$)$')
|
||||
|
@ -310,6 +317,7 @@ def _get_detailed_errors(error, depth, absolute_schema_path, absolute_schema,
|
|||
details.extend(_get_detailed_errors(
|
||||
sub_error, depth + 1, schema_path, absolute_schema,
|
||||
filter_errors))
|
||||
|
||||
return details
|
||||
|
||||
|
||||
|
@ -328,6 +336,7 @@ def _find_type_in_schema_list(schemas, type):
|
|||
if ('properties' in schema and 'type' in schema['properties']
|
||||
and schema['properties']['type'] == type):
|
||||
return True, index
|
||||
|
||||
return False, 0
|
||||
|
||||
|
||||
|
@ -356,6 +365,7 @@ def _pretty_print_schema_path(absolute_schema_path, absolute_schema):
|
|||
current_schema = absolute_schema
|
||||
for i in current_path[1:]:
|
||||
current_schema = current_schema[i]
|
||||
|
||||
return '/'.join([str(x) for x in pretty_path])
|
||||
|
||||
|
||||
|
@ -403,3 +413,68 @@ def validate_json_schema(net_data):
|
|||
config_path, error.message))
|
||||
|
||||
return error_messages
|
||||
|
||||
|
||||
def tags_to_dict(resource_tags):
|
||||
tag_dict = dict()
|
||||
for tag in resource_tags:
|
||||
if not tag.startswith('tripleo_'):
|
||||
continue
|
||||
try:
|
||||
key, value = tag.rsplit('=')
|
||||
except ValueError:
|
||||
continue
|
||||
if key == 'tripleo_net_idx':
|
||||
value = int(value)
|
||||
tag_dict.update({key: value})
|
||||
|
||||
return tag_dict
|
||||
|
||||
|
||||
def wrap_ipv6(ip_address):
|
||||
"""Wrap the address in square brackets if it's an IPv6 address."""
|
||||
if ipaddress.ip_address(ip_address).version == 6:
|
||||
return '[{}]'.format(ip_address)
|
||||
|
||||
return ip_address
|
||||
|
||||
|
||||
def get_overcloud_network_resources(conn, stack_name):
|
||||
network_resource_dict = dict()
|
||||
networks = [res for res in conn.orchestration.resources(stack_name)
|
||||
if res.name == 'Networks'][0]
|
||||
networks = conn.orchestration.resources(networks.physical_resource_id)
|
||||
for net in networks:
|
||||
if net.name == 'NetworkExtraConfig':
|
||||
continue
|
||||
network_resource_dict[net.name] = dict()
|
||||
for res in conn.orchestration.resources(net.physical_resource_id):
|
||||
if res.resource_type == TYPE_SEGMENT:
|
||||
continue
|
||||
network_resource_dict[net.name][res.name] = {
|
||||
RES_ID: res.physical_resource_id,
|
||||
RES_TYPE: res.resource_type
|
||||
}
|
||||
|
||||
return network_resource_dict
|
||||
|
||||
|
||||
def create_name_id_maps(conn):
|
||||
net_name_map = {}
|
||||
net_id_map = {}
|
||||
cidr_prefix_map = {}
|
||||
for net in conn.network.networks():
|
||||
subnets = conn.network.subnets(network_id=net.id)
|
||||
net_id_map[net.id] = net.name
|
||||
net_name_map[net.name] = dict(id=net.id)
|
||||
subnets_map = net_name_map[net.name]['subnets'] = dict()
|
||||
|
||||
for s in subnets:
|
||||
subnets_map[s.name] = s.id
|
||||
cidr_prefix_map[s.id] = s.cidr.split('/')[-1]
|
||||
|
||||
net_maps = dict(by_id=net_id_map,
|
||||
by_name=net_name_map,
|
||||
cidr_prefix_map=cidr_prefix_map)
|
||||
|
||||
return net_maps
|
||||
|
|
|
@ -18,9 +18,9 @@
|
|||
import yaml
|
||||
|
||||
try:
|
||||
from ansible.module_utils import tripleo_common_utils as tc
|
||||
from ansible.module_utils import network_data_v2 as n_utils
|
||||
except ImportError:
|
||||
from tripleo_ansible.ansible_plugins.module_utils import tripleo_common_utils as tc
|
||||
from tripleo_ansible.ansible_plugins.module_utils import network_data_v2 as n_utils # noqa
|
||||
from ansible.module_utils.basic import AnsibleModule
|
||||
from ansible.module_utils.openstack import openstack_full_argument_spec
|
||||
from ansible.module_utils.openstack import openstack_module_kwargs
|
||||
|
@ -112,13 +112,6 @@ EXAMPLES = '''
|
|||
dest: /path/exported-network-data.yaml
|
||||
'''
|
||||
|
||||
TYPE_NET = 'OS::Neutron::Net'
|
||||
TYPE_SUBNET = 'OS::Neutron::Subnet'
|
||||
TYPE_SEGMENT = 'OS::Neutron::Segment'
|
||||
RES_ID = 'physical_resource_id'
|
||||
RES_TYPE = 'resource_type'
|
||||
|
||||
NET_VIP_SUFFIX = '_virtual_ip'
|
||||
|
||||
DEFAULT_NETWORK_MTU = 1500
|
||||
DEFAULT_NETWROK_SHARED = False
|
||||
|
@ -130,49 +123,11 @@ DEFAULT_SUBNET_IPV6_ADDRESS_MODE = None
|
|||
DEFAULT_SUBNET_IPV6_RA_MODE = None
|
||||
|
||||
|
||||
def get_overcloud_network_resources(conn, stack_name):
|
||||
network_resource_dict = dict()
|
||||
networks = [res for res in conn.orchestration.resources(stack_name)
|
||||
if res.name == 'Networks'][0]
|
||||
networks = conn.orchestration.resources(networks.physical_resource_id)
|
||||
for net in networks:
|
||||
if net.name == 'NetworkExtraConfig':
|
||||
continue
|
||||
network_resource_dict[net.name] = dict()
|
||||
for res in conn.orchestration.resources(net.physical_resource_id):
|
||||
if res.resource_type == TYPE_SEGMENT:
|
||||
continue
|
||||
network_resource_dict[net.name][res.name] = {
|
||||
RES_ID: res.physical_resource_id,
|
||||
RES_TYPE: res.resource_type
|
||||
}
|
||||
|
||||
return network_resource_dict
|
||||
|
||||
|
||||
def tripleo_resource_tags_to_dict(resource_tags):
|
||||
tag_dict = dict()
|
||||
for tag in resource_tags:
|
||||
if not tag.startswith('tripleo_'):
|
||||
continue
|
||||
try:
|
||||
key, value = tag.rsplit('=')
|
||||
except ValueError:
|
||||
continue
|
||||
|
||||
if key == 'tripleo_net_idx':
|
||||
value = int(value)
|
||||
|
||||
tag_dict.update({key: value})
|
||||
|
||||
return tag_dict
|
||||
|
||||
|
||||
def is_vip_network(conn, network_id):
|
||||
network_name = conn.network.get_network(network_id).name
|
||||
vip_ports = conn.network.ports(network_id=network_id,
|
||||
name='{}{}'.format(network_name,
|
||||
NET_VIP_SUFFIX))
|
||||
n_utils.NET_VIP_SUFFIX))
|
||||
try:
|
||||
next(vip_ports)
|
||||
return True
|
||||
|
@ -195,7 +150,7 @@ def get_network_info(conn, network_id):
|
|||
_dict.pop('vip')
|
||||
|
||||
network = conn.network.get_network(network_id)
|
||||
tag_dict = tripleo_resource_tags_to_dict(network.tags)
|
||||
tag_dict = n_utils.tags_to_dict(network.tags)
|
||||
|
||||
net_dict = {
|
||||
'name_lower': network.name,
|
||||
|
@ -243,7 +198,7 @@ def get_subnet_info(conn, subnet_id):
|
|||
|
||||
subnet = conn.network.get_subnet(subnet_id)
|
||||
segment = conn.network.get_segment(subnet.segment_id)
|
||||
tag_dict = tripleo_resource_tags_to_dict(subnet.tags)
|
||||
tag_dict = n_utils.tags_to_dict(subnet.tags)
|
||||
subnet_name = subnet.name
|
||||
|
||||
subnet_dict = {
|
||||
|
@ -286,22 +241,26 @@ def get_subnet_info(conn, subnet_id):
|
|||
return subnet_name, subnet_dict
|
||||
|
||||
|
||||
def parse_net_resource(conn, net_resource, indexed_networks, net_entry):
|
||||
for res in net_resource:
|
||||
if net_resource[res][n_utils.RES_TYPE] == n_utils.TYPE_NET:
|
||||
idx, net_dict = get_network_info(
|
||||
conn, net_resource[res][n_utils.RES_ID])
|
||||
net_entry.update(net_dict)
|
||||
if net_resource[res][n_utils.RES_TYPE] == n_utils.TYPE_SUBNET:
|
||||
subnet_name, subnet_dict = get_subnet_info(
|
||||
conn, net_resource[res][n_utils.RES_ID])
|
||||
net_entry['subnets'].update({subnet_name: subnet_dict})
|
||||
indexed_networks[idx] = net_entry
|
||||
|
||||
|
||||
def parse_net_resources(conn, net_resources):
|
||||
indexed_networks = dict()
|
||||
for net in net_resources:
|
||||
name = net.rpartition('Network')[0]
|
||||
net_entry = {'name': name, 'subnets': dict()}
|
||||
for res in net_resources[net]:
|
||||
res_dict = net_resources[net][res]
|
||||
if res_dict['resource_type'] == TYPE_NET:
|
||||
idx, net_dict = get_network_info(conn, res_dict[RES_ID])
|
||||
net_entry.update(net_dict)
|
||||
if res_dict['resource_type'] == TYPE_SUBNET:
|
||||
subnet_name, subnet_dict = get_subnet_info(conn,
|
||||
res_dict[RES_ID])
|
||||
net_entry['subnets'].update({subnet_name: subnet_dict})
|
||||
|
||||
indexed_networks[idx] = net_entry
|
||||
parse_net_resource(conn, net_resources[net], indexed_networks,
|
||||
net_entry)
|
||||
|
||||
network_data = [indexed_networks[i] for i in sorted(indexed_networks)]
|
||||
|
||||
|
@ -330,7 +289,8 @@ def run_module():
|
|||
|
||||
try:
|
||||
_, conn = openstack_cloud_from_module(module)
|
||||
net_resources = get_overcloud_network_resources(conn, stack_name)
|
||||
net_resources = n_utils.get_overcloud_network_resources(conn,
|
||||
stack_name)
|
||||
result['network_data'] = parse_net_resources(conn, net_resources)
|
||||
|
||||
result['changed'] = True if result['network_data'] else False
|
||||
|
|
|
@ -16,14 +16,13 @@
|
|||
# under the License.
|
||||
|
||||
from concurrent import futures
|
||||
import ipaddress
|
||||
import metalsmith
|
||||
import yaml
|
||||
|
||||
try:
|
||||
from ansible.module_utils import tripleo_common_utils as tc
|
||||
from ansible.module_utils import network_data_v2 as n_utils
|
||||
except ImportError:
|
||||
from tripleo_ansible.ansible_plugins.module_utils import tripleo_common_utils as tc
|
||||
from tripleo_ansible.ansible_plugins.module_utils import network_data_v2 as n_utils # noqa
|
||||
from ansible.module_utils.basic import AnsibleModule
|
||||
from ansible.module_utils.openstack import openstack_full_argument_spec
|
||||
from ansible.module_utils.openstack import openstack_module_kwargs
|
||||
|
@ -182,36 +181,6 @@ EXAMPLES = '''
|
|||
'''
|
||||
|
||||
|
||||
def wrap_ipv6(ip_address):
|
||||
"""Wrap the address in square brackets if it's an IPv6 address."""
|
||||
if ipaddress.ip_address(ip_address).version == 6:
|
||||
return '[{}]'.format(ip_address)
|
||||
|
||||
return ip_address
|
||||
|
||||
|
||||
def create_name_id_maps(conn):
|
||||
net_name_map = {}
|
||||
net_id_map = {}
|
||||
cidr_prefix_map = {}
|
||||
for net in conn.network.networks():
|
||||
subnets = conn.network.subnets(network_id=net.id)
|
||||
|
||||
net_id_map[net.id] = net.name
|
||||
net_name_map[net.name] = dict(id=net.id)
|
||||
subnets_map = net_name_map[net.name]['subnets'] = dict()
|
||||
|
||||
for s in subnets:
|
||||
subnets_map[s.name] = s.id
|
||||
cidr_prefix_map[s.id] = s.cidr.split('/')[-1]
|
||||
|
||||
net_maps = dict(by_id=net_id_map,
|
||||
by_name=net_name_map,
|
||||
cidr_prefix_map=cidr_prefix_map)
|
||||
|
||||
return net_maps
|
||||
|
||||
|
||||
def delete_ports(conn, ports):
|
||||
for port in ports:
|
||||
conn.network.delete_port(port.id)
|
||||
|
@ -427,7 +396,7 @@ def generate_node_port_map(result, net_maps, ports_by_node):
|
|||
node_net = node[net_name] = dict()
|
||||
node_net['ip_address'] = ip_address
|
||||
node_net['ip_subnet'] = '/'.join([ip_address, cidr_prefix])
|
||||
node_net['ip_address_uri'] = wrap_ipv6(ip_address)
|
||||
node_net['ip_address_uri'] = n_utils.wrap_ipv6(ip_address)
|
||||
|
||||
|
||||
def validate_instance_nets_in_net_map(instances, net_maps):
|
||||
|
@ -449,7 +418,7 @@ def manage_instances_ports(result, conn, stack, instances, concurrency, state,
|
|||
if concurrency < 1:
|
||||
concurrency = len(instances)
|
||||
|
||||
net_maps = create_name_id_maps(conn)
|
||||
net_maps = n_utils.create_name_id_maps(conn)
|
||||
validate_instance_nets_in_net_map(instances, net_maps)
|
||||
ports_by_node = dict()
|
||||
|
||||
|
|
|
@ -13,12 +13,16 @@
|
|||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import yaml
|
||||
import copy
|
||||
import mock
|
||||
import yaml
|
||||
|
||||
import openstack
|
||||
|
||||
from tripleo_ansible.tests import base as tests_base
|
||||
|
||||
from tripleo_ansible.ansible_plugins.module_utils import network_data_v2
|
||||
from tripleo_ansible.tests import stubs
|
||||
|
||||
|
||||
NET_DATA = yaml.safe_load('''
|
||||
|
@ -90,6 +94,11 @@ IPV6_SUBNET_KEYS = {'ipv6_subnet', 'ipv6_allocation_pools', 'routes_ipv6',
|
|||
|
||||
class TestNetworkDataV2(tests_base.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(TestNetworkDataV2, self).setUp()
|
||||
# Helper function to convert array to generator
|
||||
self.a2g = lambda x: (n for n in x)
|
||||
|
||||
def test_validator_ok(self):
|
||||
ipv4_only = copy.deepcopy(NET_DATA)
|
||||
ipv6_only = copy.deepcopy(NET_DATA)
|
||||
|
@ -409,3 +418,70 @@ class TestNetworkDataV2(tests_base.TestCase):
|
|||
(r"- subnets/additionalProperties/oneOf/ipv6_subnet"
|
||||
r"/vlan/type: 'not_an_int' is not of type 'integer'"
|
||||
r"\n"))
|
||||
|
||||
def test_tripleo_resource_tags_to_dict(self):
|
||||
tags = ['foo=bar', 'baz=qux', 'tripleo_foo=bar', 'tripleo_baz=qux',
|
||||
'tripleo_net_idx=3']
|
||||
expected = {'tripleo_foo': 'bar', 'tripleo_baz': 'qux',
|
||||
'tripleo_net_idx': 3}
|
||||
result = network_data_v2.tags_to_dict(tags)
|
||||
self.assertEqual(expected, result)
|
||||
|
||||
@mock.patch.object(openstack.connection, 'Connection', autospec=True)
|
||||
def test_create_name_id_maps(self, conn_mock):
|
||||
subnet1 = stubs.FakeNeutronSubnet(id='subnet1_id',
|
||||
name='subnet1',
|
||||
cidr='192.168.24.0/24')
|
||||
subnet2 = stubs.FakeNeutronSubnet(id='subnet2_id',
|
||||
name='subnet2',
|
||||
cidr='192.168.25.0/25')
|
||||
subnet3 = stubs.FakeNeutronSubnet(id='subnet3_id',
|
||||
name='subnet3',
|
||||
cidr='192.168.26.0/26')
|
||||
subnet4 = stubs.FakeNeutronSubnet(id='subnet4_id',
|
||||
name='subnet4',
|
||||
cidr='192.168.27.0/27')
|
||||
network1 = stubs.FakeNeutronNetwork(
|
||||
id='network1_id',
|
||||
name='network1',
|
||||
subnet_ids=['subnet1_id', 'subnet2_id']
|
||||
)
|
||||
network2 = stubs.FakeNeutronNetwork(
|
||||
id='network2_id',
|
||||
name='network2',
|
||||
subnet_ids=['subnet3_id', 'subnet4_id']
|
||||
)
|
||||
conn_mock.network.networks.return_value = self.a2g([network1,
|
||||
network2])
|
||||
conn_mock.network.subnets.side_effect = [self.a2g([subnet1, subnet2]),
|
||||
self.a2g([subnet3, subnet4])]
|
||||
net_maps = network_data_v2.create_name_id_maps(conn_mock)
|
||||
expected_by_name_map = {
|
||||
'network1': {
|
||||
'id': 'network1_id',
|
||||
'subnets': {
|
||||
'subnet1': 'subnet1_id',
|
||||
'subnet2': 'subnet2_id'
|
||||
}
|
||||
},
|
||||
'network2': {
|
||||
'id': 'network2_id',
|
||||
'subnets': {
|
||||
'subnet3': 'subnet3_id',
|
||||
'subnet4': 'subnet4_id'
|
||||
}
|
||||
}
|
||||
}
|
||||
expected_by_id_map = {
|
||||
'network1_id': 'network1',
|
||||
'network2_id': 'network2',
|
||||
}
|
||||
expected_cidr_prefix_map = {
|
||||
'subnet1_id': '24',
|
||||
'subnet2_id': '25',
|
||||
'subnet3_id': '26',
|
||||
'subnet4_id': '27',
|
||||
}
|
||||
self.assertEqual(expected_by_name_map, net_maps['by_name'])
|
||||
self.assertEqual(expected_by_id_map, net_maps['by_id'])
|
||||
self.assertEqual(expected_cidr_prefix_map, net_maps['cidr_prefix_map'])
|
||||
|
|
|
@ -16,6 +16,10 @@
|
|||
import mock
|
||||
import openstack
|
||||
|
||||
try:
|
||||
from ansible.module_utils import network_data_v2 as n_utils
|
||||
except ImportError:
|
||||
from tripleo_ansible.ansible_plugins.module_utils import network_data_v2 as n_utils # noqa
|
||||
from tripleo_ansible.ansible_plugins.modules import (
|
||||
tripleo_overcloud_network_extract as plugin)
|
||||
from tripleo_ansible.tests import base as tests_base
|
||||
|
@ -24,21 +28,13 @@ from tripleo_ansible.tests import stubs
|
|||
|
||||
class TestTripleoOvercloudNetworkExtract(tests_base.TestCase):
|
||||
|
||||
def test_tripleo_resource_tags_to_dict(self):
|
||||
tags = ['foo=bar', 'baz=qux', 'tripleo_foo=bar', 'tripleo_baz=qux',
|
||||
'tripleo_net_idx=3']
|
||||
expected = {'tripleo_foo': 'bar', 'tripleo_baz': 'qux',
|
||||
'tripleo_net_idx': 3}
|
||||
result = plugin.tripleo_resource_tags_to_dict(tags)
|
||||
self.assertEqual(expected, result)
|
||||
|
||||
@mock.patch.object(openstack.connection, 'Connection', autospec=True)
|
||||
def test_is_vip_network_true(self, conn_mock):
|
||||
net_name = 'external'
|
||||
net_id = '132f871f-eaec-4fed-9475-0d54465e0f00'
|
||||
fake_network = stubs.FakeNeutronNetwork(id=net_id, name=net_name)
|
||||
fake_port = stubs.FakeNeutronPort(
|
||||
name='{}{}'.format(net_name, plugin.NET_VIP_SUFFIX),
|
||||
name='{}{}'.format(net_name, n_utils.NET_VIP_SUFFIX),
|
||||
fixed_ips=[{'ip_address': '10.10.10.10', 'subnet_id': 'foo'}]
|
||||
)
|
||||
|
||||
|
@ -228,11 +224,11 @@ class TestTripleoOvercloudNetworkExtract(tests_base.TestCase):
|
|||
net_resources = {
|
||||
'StorageNetwork': {
|
||||
'StorageNetwork': {'physical_resource_id': 'fake-id',
|
||||
'resource_type': plugin.TYPE_NET},
|
||||
'resource_type': n_utils.TYPE_NET},
|
||||
'StorageSubnet': {'physical_resource_id': 'fake-id',
|
||||
'resource_type': plugin.TYPE_SUBNET},
|
||||
'resource_type': n_utils.TYPE_SUBNET},
|
||||
'StorageSubnet_leaf1': {'physical_resource_id': 'fake-id',
|
||||
'resource_type': plugin.TYPE_SUBNET}
|
||||
'resource_type': n_utils.TYPE_SUBNET}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -74,66 +74,6 @@ class TestTripleoOvercloudNetworkPorts(tests_base.TestCase):
|
|||
# Helper function to convert array to generator
|
||||
self.a2g = lambda x: (n for n in x)
|
||||
|
||||
@mock.patch.object(openstack.connection, 'Connection', autospec=True)
|
||||
def test_create_name_id_maps(self, conn_mock):
|
||||
subnet1 = stubs.FakeNeutronSubnet(id='subnet1_id',
|
||||
name='subnet1',
|
||||
cidr='192.168.24.0/24')
|
||||
subnet2 = stubs.FakeNeutronSubnet(id='subnet2_id',
|
||||
name='subnet2',
|
||||
cidr='192.168.25.0/25')
|
||||
subnet3 = stubs.FakeNeutronSubnet(id='subnet3_id',
|
||||
name='subnet3',
|
||||
cidr='192.168.26.0/26')
|
||||
subnet4 = stubs.FakeNeutronSubnet(id='subnet4_id',
|
||||
name='subnet4',
|
||||
cidr='192.168.27.0/27')
|
||||
network1 = stubs.FakeNeutronNetwork(
|
||||
id='network1_id',
|
||||
name='network1',
|
||||
subnet_ids=['subnet1_id', 'subnet2_id']
|
||||
)
|
||||
network2 = stubs.FakeNeutronNetwork(
|
||||
id='network2_id',
|
||||
name='network2',
|
||||
subnet_ids=['subnet3_id', 'subnet4_id']
|
||||
)
|
||||
conn_mock.network.networks.return_value = self.a2g([network1,
|
||||
network2])
|
||||
conn_mock.network.subnets.side_effect = [self.a2g([subnet1, subnet2]),
|
||||
self.a2g([subnet3, subnet4])]
|
||||
|
||||
net_maps = plugin.create_name_id_maps(conn_mock)
|
||||
expected_by_name_map = {
|
||||
'network1': {
|
||||
'id': 'network1_id',
|
||||
'subnets': {
|
||||
'subnet1': 'subnet1_id',
|
||||
'subnet2': 'subnet2_id'
|
||||
}
|
||||
},
|
||||
'network2': {
|
||||
'id': 'network2_id',
|
||||
'subnets': {
|
||||
'subnet3': 'subnet3_id',
|
||||
'subnet4': 'subnet4_id'
|
||||
}
|
||||
}
|
||||
}
|
||||
expected_by_id_map = {
|
||||
'network1_id': 'network1',
|
||||
'network2_id': 'network2',
|
||||
}
|
||||
expected_cidr_prefix_map = {
|
||||
'subnet1_id': '24',
|
||||
'subnet2_id': '25',
|
||||
'subnet3_id': '26',
|
||||
'subnet4_id': '27',
|
||||
}
|
||||
self.assertEqual(expected_by_name_map, net_maps['by_name'])
|
||||
self.assertEqual(expected_by_id_map, net_maps['by_id'])
|
||||
self.assertEqual(expected_cidr_prefix_map, net_maps['cidr_prefix_map'])
|
||||
|
||||
@mock.patch.object(openstack.connection, 'Connection', autospec=True)
|
||||
def test_delete_ports(self, mock_conn):
|
||||
port1 = stubs.FakeNeutronPort(id='port1_id')
|
||||
|
|
Loading…
Reference in New Issue