x
/
tobiko
Code Issues Proposed changes

Update ensure_router_interface by using an Heat stack 

Change-Id: I6605da8d6a7ce72a3207db7e05f195946d931b8c
This commit is contained in:
Federico Ressi 2022-06-02 15:31:24 +02:00
parent c1ab286885
commit 7c0c433a0d
4 changed files with 333 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
tobiko/openstack/stacks/__init__.py
View File

@ -65,6 +65,7 @@ L3haDifferentHostServerStackFixture = _l3ha.L3haDifferentHostServerStackFixture
L3haSameHostServerStackFixture = _l3ha.L3haSameHostServerStackFixture
NetworkStackFixture = _neutron.NetworkStackFixture
RouterInterfaceStackFixture = _neutron.RouterInterfaceStackFixture
RouterStackFixture = _neutron.RouterStackFixture
NetworkWithNetMtuWriteStackFixture = (
_neutron.NetworkWithNetMtuWriteStackFixture)
@ -75,6 +76,7 @@ has_external_network = _neutron.has_external_network
skip_unless_has_external_network = _neutron.skip_unless_has_external_network
get_floating_network_id = _neutron.get_floating_network_id
get_floating_network = _neutron.get_floating_network
ensure_router_interface = _neutron.ensure_router_interface
has_floating_network = _neutron.has_floating_network
skip_unless_has_floating_network = _neutron.skip_unless_has_floating_network
get_router_id = _neutron.get_router_id

189
tobiko/openstack/stacks/_neutron.py
View File

@ -155,8 +155,10 @@ class RouterStackFixture(ExternalNetworkStackFixture):
return requirements
def ensure_router_interface(self,
subnet: neutron.SubnetIdType):
ensure_router_interface(subnet=subnet,
subnet: neutron.SubnetIdType = None,
network: neutron.NetworkIdType = None):
ensure_router_interface(network=network,
subnet=subnet,
router=self.router_id,
client=self.neutron_client)
@ -167,18 +169,21 @@ class RouterStackFixture(ExternalNetworkStackFixture):
return self._neutron_client
def ensure_router_interface(subnet: neutron.SubnetIdType,
router: neutron.RouterIdType = None,
client: neutron.NeutronClientType = None) \
-> neutron.PortType:
def ensure_router_interface(
router: neutron.RouterIdType = None,
subnet: neutron.SubnetIdType = None,
network: neutron.NetworkIdType = None,
client: neutron.NeutronClientType = None,
add_cleanup=False) -> neutron.PortType:
client = neutron.neutron_client(client)
if router is None:
router = get_router_id()
router_id = neutron.get_router_id(router)
subnet_id = neutron.get_subnet_id(subnet)
client = neutron.neutron_client(client)
if subnet is None and network is None:
raise ValueError('Must specify a network or a subnet')
try:
port = neutron.find_port(fixed_ips=f"subnet_id={subnet_id}",
device_id=router_id,
port = neutron.find_port(network=network,
device=router,
subnet=subnet,
client=client)
except tobiko.ObjectNotFound:
pass
@ -187,37 +192,44 @@ def ensure_router_interface(subnet: neutron.SubnetIdType,
LOG.debug(f'Router interface already exist:\n{port_dump}')
return port
subnet = neutron.ensure_subnet_gateway(subnet=subnet,
client=client)
gateway_ip = subnet['gateway_ip']
try:
port = neutron.find_port(fixed_ips=[f'ip_address={gateway_ip}'],
client=client)
except tobiko.ObjectNotFound:
LOG.info(f"Add router interface: subnet={subnet_id}")
interface = neutron.add_router_interface(router=router,
subnet=subnet,
add_cleanup=False,
client=client)
port = neutron.find_port(fixed_ips=f"subnet_id={subnet_id}",
device_id=router_id)
if subnet is None:
assert network is not None
LOG.info("Add router interface to network "
f"{neutron.get_network_id(network)}")
for _subnet in neutron.list_subnets(network=network,
client=client):
neutron.ensure_subnet_gateway(subnet=_subnet,
client=client)
else:
port_dump = json.dumps(port, indent=4, sort_keys=True)
LOG.debug(f'Port with gateway IP already exists:\n{port_dump}')
port = neutron.create_port(client=client,
network=subnet['network_id'],
add_cleanup=False)
interface = neutron.add_router_interface(router=router,
port=port,
add_cleanup=False,
client=client)
subnet = neutron.ensure_subnet_gateway(subnet=subnet,
client=client)
gateway_ip = subnet['gateway_ip']
try:
port = neutron.find_port(fixed_ips=[f'ip_address={gateway_ip}'],
client=client)
except tobiko.ObjectNotFound:
LOG.info("Add router interface to subnet "
f"{neutron.get_subnet_id(subnet)}")
else:
# it must force router to bind new network port because subnet
# gateway IP address is already being used
port_dump = json.dumps(port, indent=4, sort_keys=True)
LOG.debug(f'Port with gateway IP already exists:\n{port_dump}')
if network is None:
network = subnet['network_id']
subnet = None
LOG.info("Add router interface to network "
f"{neutron.get_network_id(network)}")
interface_dump = json.dumps(interface, sort_keys=True, indent=4)
port_dump = json.dumps(port, indent=4, sort_keys=True)
LOG.info("Router interface added:\n"
f"{interface_dump}\n"
f"{port_dump}\n")
return port
stack = RouterInterfaceStackFixture(router=router,
subnet=subnet,
network=network,
neutron_client=client)
if add_cleanup:
tobiko.use_fixture(stack)
else:
tobiko.setup_fixture(stack)
return stack.port_details
@neutron.skip_if_missing_networking_extensions('port-security')
@ -669,3 +681,100 @@ class FloatingIpStackFixture(heat.HeatStackFixture):
if self._neutron_client is None:
self._neutron_client = self.router_stack.neutron_client
return self._neutron_client
class RouterInterfaceStackFixture(heat.HeatStackFixture):
#: Heat template file
template = _hot.heat_template_file('neutron/router_interface.yaml')
def __init__(self,
stack_name: str = None,
router: neutron.RouterIdType = None,
network: neutron.NetworkIdType = None,
subnet: neutron.SubnetIdType = None,
neutron_client: neutron.NeutronClientType = None):
self._router = router
self._network = network
self._subnet = subnet
self._neutron_client = neutron_client
super().__init__(stack_name=stack_name)
@property
def router(self) -> str:
if self._router is None:
self._router = get_router_id()
return neutron.get_router_id(self._router)
@property
def router_details(self) -> neutron.RouterType:
if self._router is None or isinstance(self._router, str):
self._router = neutron.get_router(self.router,
client=self.neutron_client)
return self._router
@property
def network(self) -> str:
if self._network is None:
subnet = self.subnet_details
if subnet is None:
raise ValueError('Must specify at least network or subnet')
self._network = subnet['network_id']
return neutron.get_network_id(self._network)
@property
def network_details(self) -> neutron.NetworkType:
if self._network is None or isinstance(self._network, str):
self._network = neutron.get_network(self.network,
client=self.neutron_client)
return self._network
@property
def subnet(self) -> typing.Optional[str]:
if self._subnet is None:
return None
else:
return neutron.get_subnet_id(self._subnet)
@property
def subnet_details(self) -> typing.Optional[neutron.SubnetType]:
if self._subnet is None:
return None
if isinstance(self._subnet, str):
self._subnet = neutron.get_subnet(self._subnet,
client=self.neutron_client)
return self._subnet
@property
def has_subnet(self) -> bool:
return self.subnet is not None
def setup_stack_name(self) -> str:
if self.stack_name is None:
if self.has_subnet:
self.stack_name = f"{self.fixture_name}-{self.subnet}"
else:
self.stack_name = f"{self.fixture_name}-{self.network}"
return self.stack_name
_port: typing.Optional[neutron.PortIdType] = None
@property
def port_details(self) -> neutron.PortType:
if self._port is None:
port_id = self.port_id
if port_id is None:
self._port = neutron.find_port(network_id=self.network_id,
device_id=self.router_id,
client=self.neutron_client)
else:
self._port = neutron.get_port(port_id,
client=self.neutron_client)
assert isinstance(self._port, dict)
return self._port
@property
def neutron_client(self) -> neutron.NeutronClientType:
if self._neutron_client is None:
self._neutron_client = neutron.get_neutron_client()
return self._neutron_client

103
tobiko/openstack/stacks/neutron/router_interface.yaml Normal file
View File

@ -0,0 +1,103 @@
heat_template_version: newton
description: Add a router interface on given network
parameters:
router:
description: Router to add port to
type: string
constraints:
- custom_constraint: neutron.router
network:
description: Network where to create router port
type: string
constraints:
- custom_constraint: neutron.network
has_subnet:
description: Whenever to specify subnet for router interface
type: boolean
default: false
subnet:
description: Subnet to be used for router interface
type: string
default: '<NO-SUBNET>'
conditions:
has_subnet:
{get_param: has_subnet}
create_port:
not: {get_param: has_subnet}
resources:
_router:
type: OS::Neutron::Router
external_id: {get_param: router}
_network:
type: OS::Neutron::Net
external_id: {get_param: network}
_subnet:
type: OS::Neutron::Subnet
condition: has_subnet
external_id: {get_param: subnet}
_router_interface_subnet:
type: OS::Neutron::RouterInterface
condition: has_subnet
properties:
router: {get_resource: _router}
subnet: {get_resource: _subnet}
_port:
type: OS::Neutron::Port
condition: create_port
properties:
network: {get_resource: _network}
_router_interface_port:
type: OS::Neutron::RouterInterface
condition: create_port
properties:
router: {get_resource: _router}
port: {get_resource: _port}
outputs:
router_id:
description: Router ID
value: {get_resource: _router}
network_id:
description: Tenant network ID
value: {get_resource: _network}
subnet_id:
description: Tenant subnet ID
condition: has_subnet
value: {get_resource: _subnet}
port_id:
description: Router tenant port ID
condition: create_port
value: {get_resource: _port}
external_network_id:
description: External router network ID
value: {get_attr: [_router, external_gateway_info, network_id]}
mtu:
description: Network MTU value (integer)
value: {get_attr: [_network, mtu]}

79
tobiko/tests/functional/openstack/stacks/test_neutron.py
View File

@ -15,6 +15,7 @@
# under the License.
from __future__ import absolute_import
from oslo_log import log
import testtools
import tobiko
@ -23,6 +24,9 @@ from tobiko.openstack import neutron
from tobiko.openstack import stacks
LOG = log.getLogger(__name__)
@keystone.skip_unless_has_keystone_credentials()
class NetworkTest(testtools.TestCase):
"""Tests network creation"""
@ -140,3 +144,78 @@ class RouterTest(testtools.TestCase):
def test_has_router(self):
self.assertTrue(stacks.has_router())
class RouterInterfaceTestRouter(stacks.RouterStackFixture):
pass
class RouterInterfaceTestNetwork(stacks.NetworkStackFixture):
pass
@keystone.skip_unless_has_keystone_credentials()
@stacks.skip_unless_has_floating_network
class RouterInterfaceTest(testtools.TestCase):
router_stack = tobiko.required_fixture(RouterInterfaceTestRouter)
network_stack = tobiko.required_fixture(RouterInterfaceTestNetwork)
required_fixtures = [router_stack, network_stack]
@classmethod
def tearDownClass(cls) -> None:
for fixture in cls.required_fixtures:
try:
tobiko.cleanup_fixture(fixture.fixture)
except Exception:
LOG.exception(f'Error cleaning up fixture: {fixture.fixture}')
def test_ensure_router_interface_with_subnet(self,
ip_version=4):
network = neutron.create_network()
subnet = neutron.create_subnet(network=network,
ip_version=ip_version)
self._test_ensure_router(subnet=subnet)
def test_ensure_router_interface_with_ipv6_subnet(self):
self.test_ensure_router_interface_with_subnet(ip_version=6)
def test_ensure_router_interface_with_routed_ipv4_subnet(self):
self._test_ensure_router(subnet=self.network_stack.ipv4_subnet_id)
def test_ensure_router_interface_with_routed_ipv6_subnet(self):
self._test_ensure_router(subnet=self.network_stack.ipv6_subnet_id)
def test_ensure_router_interface_with_network(self):
network = neutron.create_network()
neutron.create_subnet(network=network)
self._test_ensure_router(network=network)
def test_ensure_router_interface_with_routed_network(self):
self._test_ensure_router(network=self.network_stack.network_id)
def _test_ensure_router(self,
network: neutron.NetworkIdType = None,
subnet: neutron.SubnetIdType = None):
router = self.router_stack.router_details
self.assertRaises(tobiko.ObjectNotFound,
neutron.find_port,
device=router,
subnet=subnet,
network=network)
port = stacks.ensure_router_interface(router=router,
network=network,
subnet=subnet,
add_cleanup=True)
self.assertEqual(router['id'], port['device_id'])
self.assertEqual(neutron.find_port(device=router,
network=network,
subnet=subnet),
port)
self.assertEqual(port,
stacks.ensure_router_interface(router=router,
network=network,
subnet=subnet,
add_cleanup=True))
return port