Browse Source

Test routes are not deleted on sync

This test adds support to store defined routes in unit tests and to
validate that no routes are recreated during sync operation

Change-Id: If2bc4b7334f472d83b307fecc6aad5b20f6b86af
(cherry picked from commit 5843ba4cc6)
tags/2018.2.9
Alin Iorga 2 months ago
parent
commit
d8b7f44700
3 changed files with 125 additions and 2 deletions
  1. +67
    -1
      networking_arista/tests/unit/l3Plugin/test_arista_l3_driver.py
  2. +51
    -1
      networking_arista/tests/unit/utils.py
  3. +7
    -0
      tox.ini

+ 67
- 1
networking_arista/tests/unit/l3Plugin/test_arista_l3_driver.py View File

@@ -945,7 +945,7 @@ class AristaL3SyncWorkerTestBase(
test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
"""Base test class for L3 Sync Worker test cases"""

def setUp(self, cleanup=True):
def setUp(self, cleanup=True, vrfDefaultRoute=False):
cfg.CONF.import_opt('network_vlan_ranges',
'neutron.plugins.ml2.drivers.type_vlan',
group='ml2_type_vlan')
@@ -953,6 +953,8 @@ class AristaL3SyncWorkerTestBase(
'ml2_type_vlan')
cfg.CONF.set_override('enable_cleanup', 'True' if cleanup else 'False',
'l3_arista')
cfg.CONF.set_override('vrf_default_route', str(vrfDefaultRoute),
'l3_arista')
setup_arista_config('value', mlag=True, vrf=True)
service_plugins = {'arista_l3': 'arista_l3'}
super(AristaL3SyncWorkerTestBase, self).setUp(
@@ -1002,6 +1004,7 @@ class AristaL3SyncWorkerCleanupTestCases(AristaL3SyncWorkerTestBase):
eos_vrf_name = '__OpenStack__r1-router1'
expected_vrfs = {eos_vrf_name:
{'rd': self._get_rd(eos_vrf_name),
'routes': {},
'svis': []}}
self.assertEqual(self.switch1._vrfs, expected_vrfs)
self.assertEqual(self.switch2._vrfs, expected_vrfs)
@@ -1183,3 +1186,66 @@ class AristaL3SyncWorkerNoCleanupTestCases(AristaL3SyncWorkerTestBase):
self.assertEqual(self.switch1._vlans, expected_vlans)
self.assertEqual(self.switch2._svis, expected_svis)
self.assertEqual(self.switch2._vlans, expected_vlans)


class AristaL3SyncWorkerVrfDefaultRouteCleanupTestCases(
AristaL3SyncWorkerTestBase):
"""Test cases for the L3 Sync Worker vrfDefaultRoute=True

This test also enabled cleanup=True
1. Test that default routes are not reset on sync
"""

def setUp(self):
super(AristaL3SyncWorkerVrfDefaultRouteCleanupTestCases,
self).setUp(vrfDefaultRoute=True)

def test_vrf_default_route_sync(self):
router_dict = {'router': {'name': 'router1',
'tenant_id': 't1',
'admin_state_up': True}}
router = self.driver.create_router(self.context, router_dict)
router_model = self.driver._get_router(self.context, router['id'])
net_dict = {'network': {'name': 'n1',
'tenant_id': 't1',
'admin_state_up': True,
'shared': False,
'provider:physical_network': 'default',
'provider:network_type': 'vlan',
'provider:segmentation_id': 100}}
net = self.plugin.create_network(self.context, net_dict)
subnet_dict = {'subnet':
{'tenant_id': net['tenant_id'],
'name': net['name'],
'network_id': net['id'],
'ip_version': 4,
'cidr': '10.0.0.0/24',
'gateway_ip': '10.0.0.1',
'allocation_pools': None,
'enable_dhcp': False,
'dns_nameservers': None,
'host_routes': None}}
subnet = self.plugin.create_subnet(self.context, subnet_dict)
ext_ips = [{'subnet_id': subnet['id'], 'ip_address': '10.0.0.2'}]
self.driver._create_gw_port(self.context, router['id'], router_model,
net['id'], ext_ips)
self.sync_worker.synchronize()

switch1_svi = {'vlan 100':
{'ip': '10.0.0.254',
'mask': '24',
'vip': '10.0.0.2'}
}
switch2_svi = {'vlan 100':
{'ip': '10.0.0.253',
'mask': '24',
'vip': '10.0.0.2'}
}
switch_vlan = {'100': {'dynamic': False}}
self.assertEqual(self.switch1._svis, switch1_svi)
self.assertEqual(self.switch2._svis, switch2_svi)
self.assertEqual(self.switch1._vlans, switch_vlan)
self.assertEqual(self.switch2._vlans, switch_vlan)
self.sync_worker.synchronize()
self.switch1.assert_command_not_received('no ip route vrf')
self.switch2.assert_command_not_received('no ip route vrf')

+ 51
- 1
networking_arista/tests/unit/utils.py View File

@@ -104,6 +104,7 @@ class MockSwitch(object):
self._vrfs = dict()
self._svis = dict()
self._vlans = dict()
self._route = dict()
self._acl_mode_re = re.compile('^(?P<delete>no )?ip access-list '
'(?P<acl>\S+)(?P<dyn> dynamic)?$')
self._interface_mode_re = re.compile(
@@ -116,6 +117,9 @@ class MockSwitch(object):
self._ip_address_re = re.compile(
'^ip address (?P<ip>[\d.]+)/(?P<mask>\d+)$')
self._vip_re = re.compile('^ip virtual-router address (?P<ip>[\d.]+)$')
self._vrf_route_re = re.compile(
'^(?P<delete>no )?ip route vrf (?P<vrf>\S+) '
'(?P<network>[\d.]+/\d+) (?P<next_hop>[\d.]+)$')
self._svi_vrf_re = re.compile('^vrf forwarding (?P<vrf>\S+)$')
self._rd_re = re.compile('^rd (?P<rd>\S+)$')
self._varp_mac_re = re.compile(
@@ -163,6 +167,31 @@ class MockSwitch(object):
vrfs['vrfs'][vrf_name] = {'interfaces': vrf['svis'],
'routeDistinguisher': vrf['rd']}
ret.append(vrfs)
elif command == 'show ip route vrf all':
vrfs = {'vrfs': {}}
for vrf_name, vrf in self._vrfs.items():
vrfs['vrfs'][vrf_name] = {'routes': {}}
vrf_routes = vrf.get('routes')
if not vrf_routes:
continue
for route, vrf_route in vrf['routes'].items():
vrf_routes[route] = {
"kernelProgrammed": True,
"directlyConnected": False,
"routeAction": "forward",
"routeLeaked": False,
"vias": [
{
"interface": "Management1",
"nexthopAddr": vrf_route
}
],
"metric": 0,
"hardwareProgrammed": True,
"routeType": "static",
"preference": 1
}
ret.append(vrfs)
elif command == 'show version':
ret.append({'version': '4.22.0F'})
elif command == 'enable':
@@ -198,10 +227,12 @@ class MockSwitch(object):
if delete:
del self._vrfs[vrf_name]
else:
self._vrfs[vrf_name] = {'svis': []}
self._vrfs[vrf_name] = {'svis': [], 'routes': {}}
self._mode = ('vrf', vrf_name)
elif 'vlan' in command:
self._parse_vlan(command)
elif 'ip route vrf' in command:
self._parse_ip_route_vrf(command)
elif command == 'exit':
self._mode = None
else:
@@ -256,6 +287,21 @@ class MockSwitch(object):
if varp_mac_match:
pass

def _parse_ip_route_vrf(self, command):
vrf_route_match = self._vrf_route_re.match(command)
if vrf_route_match:
delete = vrf_route_match.group('delete')
vrf = vrf_route_match.group('vrf')
if vrf not in self._vrfs:
assert False
vrf_dict = self._vrfs[vrf]
vrf_routes = vrf_dict['routes']
if delete:
del self._vrfs[vrf]['routes'][vrf_route_match.group('network')]
else:
vrf_routes[vrf_route_match.group('network')] = \
vrf_route_match.group('next_hop')

def _parse_vlan(self, command):
vlan_match = self._vlan_re.match(command)
delete = vlan_match.group('delete')
@@ -269,6 +315,10 @@ class MockSwitch(object):
def received_commands(self):
return self._commands

def assert_command_not_received(self, unexpected_cmd):
for cmd in self._commands:
assert unexpected_cmd not in cmd

def clear_received_commands(self):
self._commands = []



+ 7
- 0
tox.ini View File

@@ -37,6 +37,13 @@ commands =
pip freeze
stestr run {posargs}

[testenv:debug]
basepython = python3
commands =
{[testenv:dev]commands}
pip freeze
oslo_debug_helper {posargs}

[testenv:pep8]
basepython = python3
commands =


Loading…
Cancel
Save