x/vmware-nsxlib
Code Issues Proposed changes

Add unit tests for the MP router resource

Browse Source 
Change-Id: Iccbbafc6e525c11fa28a78b73120d38f694413bb
This commit is contained in:
asarfaty 2020-04-30 10:55:26 +02:00
parent eb930ded56
commit 917ac30039
4 changed files with 224 additions and 24 deletions

2
vmware_nsxlib/tests/unit/v3/nsxlib_testcase.py

@ -26,6 +26,7 @@ from vmware_nsxlib.v3 import client as nsx_client
from vmware_nsxlib.v3 import client_cert
from vmware_nsxlib.v3 import cluster as nsx_cluster
from vmware_nsxlib.v3 import config
from vmware_nsxlib.v3 import nsx_constants
from vmware_nsxlib.v3 import utils
NSX_USER = 'admin'
@ -49,6 +50,7 @@ DNS_NAMESERVERS = ['1.1.1.1']
DNS_DOMAIN = 'openstacklocal'
JSESSIONID = 'my_sess_id'
LATEST_VERSION = nsx_constants.NSX_VERSION_3_1_0
def _mock_nsxlib():

39
vmware_nsxlib/tests/unit/v3/policy/test_resources.py

@ -27,8 +27,6 @@ from vmware_nsxlib.v3.policy import core_defs
from vmware_nsxlib.v3.policy import core_resources
TEST_TENANT = 'test'
# TODO(annak): move version forward with backend releases
LATEST_VERSION = nsx_constants.NSX_VERSION_3_1_0
class NsxPolicyLibTestCase(policy_testcase.TestPolicyApi):
@ -41,7 +39,8 @@ class NsxPolicyLibTestCase(policy_testcase.TestPolicyApi):
# Mock the nsx-lib for the passthrough api
with mock.patch("vmware_nsxlib.v3.NsxLib") as mock_lib:
mock_lib.return_value.get_version.return_value = LATEST_VERSION
mock_lib.return_value.get_version.return_value = (
nsxlib_testcase.LATEST_VERSION)
self.policy_lib = policy.NsxPolicyLib(nsxlib_config)
self.policy_api = self.policy_lib.policy_api
@ -3907,7 +3906,7 @@ class TestPolicyTier0(NsxPolicyLibTestCase):
self.resourceApi.update_route_redistribution_config(
tier0_id, config, service_id, tenant=TEST_TENANT)
expected_def = core_defs.Tier0LocaleServiceDef(
nsx_version=LATEST_VERSION, tier0_id=tier0_id,
nsx_version=nsxlib_testcase.LATEST_VERSION, tier0_id=tier0_id,
service_id=service_id, route_redistribution_config=config,
tenant=TEST_TENANT)
@ -3927,7 +3926,7 @@ class TestPolicyTier0(NsxPolicyLibTestCase):
self.policy_lib.feature_supported(
nsx_constants.FEATURE_ROUTE_REDISTRIBUTION_CONFIG))
with mock.patch.object(self.policy_lib, "get_version",
return_value=LATEST_VERSION):
return_value=nsxlib_testcase.LATEST_VERSION):
self.assertTrue(
self.policy_lib.feature_supported(
nsx_constants.FEATURE_ROUTE_REDISTRIBUTION_CONFIG))
@ -4056,7 +4055,7 @@ class TestPolicySegment(NsxPolicyLibTestCase):
if admin_state:
kwargs['admin_state'] = admin_state if 'UP' else 'DOWN'
expected_def = core_defs.SegmentDef(
nsx_version=LATEST_VERSION,
nsx_version=nsxlib_testcase.LATEST_VERSION,
segment_id=mock.ANY,
name=name,
**kwargs)
@ -4136,11 +4135,12 @@ class TestPolicySegment(NsxPolicyLibTestCase):
name=name,
admin_state=admin_state,
tenant=TEST_TENANT)
expected_def = core_defs.SegmentDef(nsx_version=LATEST_VERSION,
segment_id=segment_id,
name=name,
admin_state=admin_state,
tenant=TEST_TENANT)
expected_def = core_defs.SegmentDef(
nsx_version=nsxlib_testcase.LATEST_VERSION,
segment_id=segment_id,
name=name,
admin_state=admin_state,
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
def test_remove_connectivity_and_subnets(self):
@ -4314,13 +4314,14 @@ class TestPolicyIpPool(NsxPolicyLibTestCase):
with mock.patch.object(
self.policy_api, "create_or_update") as api_call, \
mock.patch.object(self.resourceApi, 'version', LATEST_VERSION):
mock.patch.object(self.resourceApi, 'version',
nsxlib_testcase.LATEST_VERSION):
self.resourceApi.allocate_block_subnet(
ip_pool_id, ip_block_id, size, ip_subnet_id,
tenant=TEST_TENANT, start_ip=start_ip)
expected_def = core_defs.IpPoolBlockSubnetDef(
nsx_version=LATEST_VERSION,
nsx_version=nsxlib_testcase.LATEST_VERSION,
ip_pool_id=ip_pool_id,
ip_block_id=ip_block_id,
ip_subnet_id=ip_subnet_id,
@ -4609,7 +4610,7 @@ class TestPolicySegmentPort(NsxPolicyLibTestCase):
self.policy_lib.feature_supported(
nsx_constants.FEATURE_SWITCH_HYPERBUS_MODE))
with mock.patch.object(self.policy_lib, "get_version",
return_value=LATEST_VERSION):
return_value=nsxlib_testcase.LATEST_VERSION):
self.assertTrue(
self.policy_lib.feature_supported(
nsx_constants.FEATURE_SWITCH_HYPERBUS_MODE))
@ -4631,7 +4632,8 @@ class TestPolicySegmentPort(NsxPolicyLibTestCase):
with mock.patch.object(
self.policy_api, "create_or_update") as api_call, \
mock.patch.object(self.resourceApi, 'version', LATEST_VERSION):
mock.patch.object(self.resourceApi, 'version',
nsxlib_testcase.LATEST_VERSION):
result = self.resourceApi.create_or_overwrite(
name, segment_id, description=description,
address_bindings=address_bindings,
@ -4643,7 +4645,7 @@ class TestPolicySegmentPort(NsxPolicyLibTestCase):
tenant=TEST_TENANT)
expected_def = core_defs.SegmentPortDef(
nsx_version=LATEST_VERSION,
nsx_version=nsxlib_testcase.LATEST_VERSION,
segment_id=segment_id,
port_id=mock.ANY,
name=name,
@ -4719,7 +4721,8 @@ class TestPolicySegmentPort(NsxPolicyLibTestCase):
hyperbus_mode = 'DISABLE'
with mock.patch.object(
self.policy_api, "create_or_update") as api_call, \
mock.patch.object(self.resourceApi, 'version', LATEST_VERSION):
mock.patch.object(self.resourceApi, 'version',
nsxlib_testcase.LATEST_VERSION):
self.resourceApi.attach(
segment_id, port_id,
attachment_type=attachment_type, vif_id=vif_id, app_id=app_id,
@ -4729,7 +4732,7 @@ class TestPolicySegmentPort(NsxPolicyLibTestCase):
tenant=TEST_TENANT)
expected_def = core_defs.SegmentPortDef(
nsx_version=LATEST_VERSION,
nsx_version=nsxlib_testcase.LATEST_VERSION,
segment_id=segment_id,
port_id=port_id,
attachment_type=attachment_type,

197
vmware_nsxlib/tests/unit/v3/test_router.py

@ -24,6 +24,7 @@ from vmware_nsxlib.tests.unit.v3 import nsxlib_testcase
from vmware_nsxlib.tests.unit.v3 import test_constants
from vmware_nsxlib.v3 import exceptions as nsxlib_exc
from vmware_nsxlib.v3 import nsx_constants
from vmware_nsxlib.v3 import router as nsx_router
class TestRouter(nsxlib_testcase.NsxClientTestCase):
@ -44,7 +45,18 @@ class TestRouter(nsxlib_testcase.NsxClientTestCase):
self.assertEqual(
tier0_groups_dict[tier0_uuid]['member_index_list'], [0])
def test_validate_tier0_fail(self):
def test_validate_tier0_fail_no_cluster(self):
tier0_groups_dict = {}
tier0_uuid = uuidutils.generate_uuid()
rtr = {'edge_cluster_id': None}
with mock.patch.object(self.nsxlib.router._router_client, 'get',
return_value=rtr):
self.assertRaises(
nsxlib_exc.NsxLibInvalidInput,
self.nsxlib.router.validate_tier0,
tier0_groups_dict, tier0_uuid)
def test_validate_tier0_fail_no_members(self):
tier0_groups_dict = {}
tier0_uuid = uuidutils.generate_uuid()
edge_cluster = copy.copy(test_constants.FAKE_EDGE_CLUSTER)
@ -102,6 +114,139 @@ class TestRouter(nsxlib_testcase.NsxClientTestCase):
resource_type=nsx_constants.LROUTERPORT_CENTRALIZED)
self.assertEqual(csp, port)
def test_update_advertisement(self):
router_id = test_constants.FAKE_ROUTER_UUID
nat = True
con = False
static = True
enabled = True
vip = False
snat = True
data = {'advertise_route_nat': nat,
'advertise_route_connected': con,
'advertise_route_static': static,
'enabled': enabled,
'advertise_lb_vip': vip,
'advertise_lb_snat_ip': snat}
api_client = self.nsxlib.router.nsxlib.logical_router.client
with mock.patch.object(api_client, 'get', return_value={}),\
mock.patch.object(api_client, 'update') as client_update,\
mock.patch("vmware_nsxlib.v3.NsxLib.get_version",
return_value=nsxlib_testcase.LATEST_VERSION):
self.nsxlib.router.update_advertisement(router_id, **data)
client_update.assert_called_with(
'logical-routers/%s/routing/advertisement' % router_id,
{'advertise_nat_routes': nat,
'advertise_nsx_connected_routes': con,
'advertise_static_routes': static,
'enabled': enabled,
'advertise_lb_vip': vip,
'advertise_lb_snat_ip': snat},
headers=None)
def test_update_advertisement_lb_unsupported(self):
router_id = test_constants.FAKE_ROUTER_UUID
nat = True
con = False
static = True
enabled = True
vip = False
snat = True
data = {'advertise_route_nat': nat,
'advertise_route_connected': con,
'advertise_route_static': static,
'enabled': enabled,
'advertise_lb_vip': vip,
'advertise_lb_snat_ip': snat}
api_client = self.nsxlib.router.nsxlib.logical_router.client
with mock.patch.object(api_client, 'get', return_value={}),\
mock.patch.object(api_client, 'update') as client_update,\
mock.patch("vmware_nsxlib.v3.NsxLib.get_version",
return_value='2.0.0'):
self.nsxlib.router.update_advertisement(router_id, **data)
client_update.assert_called_with(
'logical-routers/%s/routing/advertisement' % router_id,
{'advertise_nat_routes': nat,
'advertise_nsx_connected_routes': con,
'advertise_static_routes': static,
'enabled': enabled},
headers=None)
def test_delete_gw_snat_rule(self):
logical_router_id = test_constants.FAKE_ROUTER_UUID
gw_ip = '1.1.1.1'
with mock.patch.object(self.nsxlib.router.nsxlib.logical_router,
'delete_nat_rule_by_values') as del_api:
self.nsxlib.router.delete_gw_snat_rule(logical_router_id, gw_ip)
del_api.assert_called_with(
logical_router_id,
translated_network=gw_ip)
def test_delete_gw_snat_rule_by_source(self):
logical_router_id = test_constants.FAKE_ROUTER_UUID
gw_ip = '1.1.1.1'
source_net = '2.0.0.0/24'
with mock.patch.object(self.nsxlib.router.nsxlib.logical_router,
'delete_nat_rule_by_values') as del_api:
self.nsxlib.router.delete_gw_snat_rule_by_source(
logical_router_id, gw_ip, source_net)
del_api.assert_called_with(
logical_router_id,
translated_network=gw_ip,
match_source_network=source_net,
skip_not_found=False, strict_mode=True)
def test_delete_gw_snat_rules(self):
logical_router_id = test_constants.FAKE_ROUTER_UUID
gw_ip = '1.1.1.1'
with mock.patch.object(self.nsxlib.router.nsxlib.logical_router,
'delete_nat_rule_by_values') as del_api:
self.nsxlib.router.delete_gw_snat_rules(
logical_router_id, gw_ip)
del_api.assert_called_with(
logical_router_id,
translated_network=gw_ip,
skip_not_found=True, strict_mode=False)
def test_add_gw_snat_rule(self):
logical_router_id = test_constants.FAKE_ROUTER_UUID
gw_ip = '1.1.1.1'
with mock.patch.object(self.nsxlib.router.nsxlib.logical_router,
'add_nat_rule') as add_api:
self.nsxlib.router.add_gw_snat_rule(
logical_router_id, gw_ip)
add_api.assert_called_with(
logical_router_id,
translated_network=gw_ip,
action="SNAT",
bypass_firewall=True,
source_net=None,
rule_priority=nsx_router.GW_NAT_PRI,
tags=None,
display_name=None)
def test_update_router_edge_cluster(self):
logical_router_id = test_constants.FAKE_ROUTER_UUID
ec_id = test_constants.FAKE_EDGE_CLUSTER
with mock.patch.object(self.nsxlib.router.nsxlib.logical_router,
'update') as update_api:
self.nsxlib.router.update_router_edge_cluster(
logical_router_id, ec_id)
update_api.assert_called_with(
logical_router_id,
edge_cluster_id=ec_id)
def test_update_router_transport_zone(self):
logical_router_id = test_constants.FAKE_ROUTER_UUID
tz_id = test_constants.FAKE_TZ_UUID
with mock.patch.object(self.nsxlib.router.nsxlib.logical_router,
'update') as update_api:
self.nsxlib.router.update_router_transport_zone(
logical_router_id, tz_id)
update_api.assert_called_with(
logical_router_id,
transport_zone_id=tz_id)
def test_create_logical_router_intf_port_by_ls_id(self):
logical_router_id = uuidutils.generate_uuid()
display_name = 'dummy'
@ -137,6 +282,56 @@ class TestRouter(nsxlib_testcase.NsxClientTestCase):
'1.1.1.1', '2.2.2.2')
self.assertEqual(add_rule.call_count, 2)
def test_delete_fip_nat_rules(self):
with mock.patch.object(self.nsxlib.logical_router,
"delete_nat_rule_by_values") as del_rule:
self.nsxlib.router.delete_fip_nat_rules(
test_constants.FAKE_ROUTER_UUID,
'1.1.1.1', '2.2.2.2')
self.assertEqual(del_rule.call_count, 2)
def test_delete_fip_nat_rules_by_int(self):
with mock.patch.object(self.nsxlib.logical_router,
"delete_nat_rule_by_values") as del_rule:
self.nsxlib.router.delete_fip_nat_rules_by_internal_ip(
test_constants.FAKE_ROUTER_UUID, '1.1.1.1')
self.assertEqual(del_rule.call_count, 2)
def test_add_static_routes(self):
dest = '1.1.1.0/24'
nexthop = '2.2.2.2'
route = {'destination': dest, 'nexthop': nexthop}
with mock.patch.object(self.nsxlib.logical_router,
"add_static_route") as add_route:
self.nsxlib.router.add_static_routes(
test_constants.FAKE_ROUTER_UUID, route)
add_route.assert_called_once_with(
test_constants.FAKE_ROUTER_UUID,
dest, nexthop)
def test_del_static_routes(self):
dest = '1.1.1.0/24'
nexthop = '2.2.2.2'
route = {'destination': dest, 'nexthop': nexthop}
with mock.patch.object(self.nsxlib.logical_router,
"delete_static_route_by_values") as del_route:
self.nsxlib.router.delete_static_routes(
test_constants.FAKE_ROUTER_UUID, route)
del_route.assert_called_once_with(
test_constants.FAKE_ROUTER_UUID,
dest_cidr=dest, nexthop=nexthop)
def test_has_service_router(self):
logical_router_id = test_constants.FAKE_ROUTER_UUID
ec_id = test_constants.FAKE_EDGE_CLUSTER
lr = {'id': logical_router_id,
'edge_cluster_id': ec_id}
with mock.patch.object(self.nsxlib.router.nsxlib.logical_router,
'get', return_value=lr):
res = self.nsxlib.router.has_service_router(
logical_router_id)
self.assertTrue(res)
def test_get_tier0_router_tz(self):
tier0_uuid = uuidutils.generate_uuid()
self.nsxlib.feature_supported = mock.MagicMock()

10
vmware_nsxlib/v3/router.py

@ -263,16 +263,16 @@ class RouterLib(object):
route['destination'],
route['nexthop'])
def change_edge_firewall_status(self, nsx_router_id,
action=nsx_constants.FW_DISABLE):
return self.nsxlib.logical_router.change_edge_firewall_status(
nsx_router_id, action)
def delete_static_routes(self, nsx_router_id, route):
return self.nsxlib.logical_router.delete_static_route_by_values(
nsx_router_id, dest_cidr=route['destination'],
nexthop=route['nexthop'])
def change_edge_firewall_status(self, nsx_router_id,
action=nsx_constants.FW_DISABLE):
return self.nsxlib.logical_router.change_edge_firewall_status(
nsx_router_id, action)
def has_service_router(self, nsx_router_id):
lrouter = self._router_client.get(nsx_router_id)
if lrouter and lrouter.get('edge_cluster_id'):