161 lines
6.9 KiB
Python
161 lines
6.9 KiB
Python
# Copyright 2021 Red Hat, Inc.
|
|
# All Rights Reserved.
|
|
#
|
|
# 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 copy
|
|
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_ovn_mac_addresses as plugin)
|
|
from tripleo_ansible.tests import base as tests_base
|
|
from tripleo_ansible.tests import stubs
|
|
|
|
|
|
FAKE_NETWORK = stubs.FakeNeutronNetwork(
|
|
name=plugin.NET_NAME,
|
|
id='fake_ovn_mac_addr_net_id',
|
|
description=plugin.NET_DESCRIPTION
|
|
)
|
|
|
|
FAKE_PORT = stubs.FakeNeutronPort(
|
|
name='server-01_ovn_physnet_network01',
|
|
dns_name='server-01',
|
|
tags=['tripleo_ovn_physnet=network01', 'tripleo_stack_name=stack',
|
|
'tripleo_role_name=Compute']
|
|
)
|
|
|
|
|
|
@mock.patch.object(openstack.connection, 'Connection', autospec=True)
|
|
class TestTripleoOVNMacAddresses(tests_base.TestCase):
|
|
|
|
def setUp(self):
|
|
super(TestTripleoOVNMacAddresses, self).setUp()
|
|
|
|
# Helper function to convert array to generator
|
|
self.a2g = lambda x: (n for n in x)
|
|
|
|
def test_create_ovn_mac_address_network(self, mock_conn):
|
|
result = dict(changed=False)
|
|
mock_conn.network.find_network.return_value = None
|
|
mock_conn.network.create_network.return_value = FAKE_NETWORK
|
|
|
|
net_id = plugin.create_ovn_mac_address_network(result, mock_conn)
|
|
|
|
mock_conn.network.create_network.assert_called_with(
|
|
name=plugin.NET_NAME,
|
|
description=plugin.NET_DESCRIPTION)
|
|
self.assertTrue(result['changed'])
|
|
self.assertEqual(FAKE_NETWORK.id, net_id)
|
|
|
|
def test_create_ovn_mac_address_network_already_exists(self, mock_conn):
|
|
result = dict(changed=False)
|
|
mock_conn.network.find_network.return_value = FAKE_NETWORK
|
|
|
|
net_id = plugin.create_ovn_mac_address_network(result, mock_conn)
|
|
|
|
mock_conn.network.create_network.assert_not_called()
|
|
self.assertFalse(result['changed'])
|
|
self.assertEqual(FAKE_NETWORK.id, net_id)
|
|
|
|
def test_port_exists_port_not_found(self, mock_conn):
|
|
net_id = FAKE_NETWORK.id
|
|
tags = ['tripleo_stack_name=stack', 'tripleo_role_name=Compute']
|
|
name = 'server-01_ovn_physnet_network01'
|
|
mock_conn.network.ports.return_value = self.a2g([])
|
|
self.assertFalse(plugin.port_exists(mock_conn, net_id, tags, name))
|
|
mock_conn.network.ports.assert_called_with(network_id=net_id,
|
|
name=name, tags=tags)
|
|
|
|
def test_port_exists_port_found(self, mock_conn):
|
|
net_id = FAKE_NETWORK.id
|
|
tags = ['tripleo_stack_name=stack', 'tripleo_role_name=Compute']
|
|
name = 'server-01_ovn_physnet_network01'
|
|
mock_conn.network.ports.return_value = self.a2g([FAKE_PORT])
|
|
self.assertTrue(plugin.port_exists(mock_conn, net_id, tags, name))
|
|
mock_conn.network.ports.assert_called_with(network_id=net_id,
|
|
name=name, tags=tags)
|
|
|
|
@mock.patch.object(plugin, 'port_exists', autospec=True)
|
|
def test_create_ovn_mac_address_ports(self, mock_port_exists, mock_conn):
|
|
result = dict(changed=False)
|
|
tags = ['tripleo_stack_name=overcloud',
|
|
'tripleo_role=Controller']
|
|
physnets = ['net-a', 'net-b']
|
|
server = 'controller-0'
|
|
mock_port_exists.return_value = False
|
|
plugin.create_ovn_mac_address_ports(result, mock_conn,
|
|
FAKE_NETWORK.id, tags,
|
|
physnets, server)
|
|
mock_conn.network.create_port.assert_has_calls(
|
|
[mock.call(network_id=FAKE_NETWORK.id,
|
|
name=server + '_ovn_physnet_net-a',
|
|
dns_name=server),
|
|
mock.call(network_id=FAKE_NETWORK.id,
|
|
name=server + '_ovn_physnet_net-b',
|
|
dns_name=server)])
|
|
mock_conn.network.set_tags.assert_has_calls(
|
|
[mock.call(mock.ANY, tags + ['tripleo_ovn_physnet=net-a']),
|
|
mock.call(mock.ANY, tags + ['tripleo_ovn_physnet=net-b'])])
|
|
|
|
@mock.patch.object(plugin, 'port_exists', autospec=True)
|
|
def test_create_ovn_mac_address_ports_exists(self, mock_port_exists,
|
|
mock_conn):
|
|
result = dict(changed=False)
|
|
tags = ['tripleo_stack_name=overcloud',
|
|
'tripleo_role=Controller']
|
|
physnets = ['net-a', 'net-b']
|
|
server = 'controller-0.example.com'
|
|
mock_port_exists.return_value = True
|
|
plugin.create_ovn_mac_address_ports(result, mock_conn,
|
|
FAKE_NETWORK.id, tags,
|
|
physnets, server)
|
|
mock_conn.network.create_port.assert_not_called()
|
|
mock_conn.network.set_tags.assert_not_called()
|
|
|
|
def test_delete_ports_for_removed_nodes(self, mock_conn):
|
|
result = dict(changed=False)
|
|
servers = ['server-01', 'server-a', 'server-b']
|
|
physnets = ['network01', 'net-a', 'net-b']
|
|
mock_conn.network.ports.return_value = self.a2g([FAKE_PORT])
|
|
plugin.remove_obsolete_ports(result, mock_conn, 'net_id',
|
|
['fake_tags'], servers, physnets)
|
|
mock_conn.network.delete_port.assert_not_called()
|
|
self.assertFalse(result['changed'])
|
|
|
|
# Verify port is deleted if server was deleted, (scale down)
|
|
mock_conn.reset_mock()
|
|
mock_conn.network.ports.return_value = self.a2g([FAKE_PORT])
|
|
servers = ['server-a', 'server-b']
|
|
plugin.remove_obsolete_ports(result, mock_conn, 'net_id',
|
|
['fake_tags'], servers, physnets)
|
|
mock_conn.network.delete_port.assert_called_with(FAKE_PORT)
|
|
self.assertTrue(result['changed'])
|
|
|
|
# Verify port is deleted if physnet no longer in bridge mappings
|
|
mock_conn.reset_mock()
|
|
result = dict(changed=False)
|
|
mock_conn.network.ports.return_value = self.a2g([FAKE_PORT])
|
|
servers = ['server-01', 'server-a', 'server-b']
|
|
physnets = ['net-a', 'net-b']
|
|
plugin.remove_obsolete_ports(result, mock_conn, 'net_id',
|
|
['fake_tags'], servers, physnets)
|
|
mock_conn.network.delete_port.assert_called_with(FAKE_PORT)
|
|
self.assertTrue(result['changed'])
|