diff --git a/vmware_nsxlib/tests/unit/v3/test_constants.py b/vmware_nsxlib/tests/unit/v3/test_constants.py index 78f66d10..e6e77ed7 100644 --- a/vmware_nsxlib/tests/unit/v3/test_constants.py +++ b/vmware_nsxlib/tests/unit/v3/test_constants.py @@ -16,6 +16,8 @@ from oslo_utils import uuidutils +from vmware_nsxlib.v3 import nsx_constants + FAKE_NAME = "fake_name" FAKE_SWITCH_UUID = uuidutils.generate_uuid() FAKE_IP_SET_UUID = uuidutils.generate_uuid() @@ -124,10 +126,20 @@ FAKE_ROUTER = { FAKE_ROUTER_PORT_UUID = uuidutils.generate_uuid() FAKE_ROUTER_PORT = { - "resource_type": "LogicalRouterLinkPort", + "resource_type": nsx_constants.LROUTERPORT_UPLINK, "revision": 0, "id": FAKE_ROUTER_PORT_UUID, "display_name": FAKE_NAME, + "logical_router_id": FAKE_ROUTER_UUID, + "subnets": [{'ip_addresses': ['172.20.1.60'], 'prefix_length': 24}] +} + +FAKE_ROUTER_LINKT1_PORT_UUID = uuidutils.generate_uuid() +FAKE_ROUTER_LINKT1_PORT = { + "resource_type": nsx_constants.LROUTERPORT_LINKONTIER1, + "revision": 0, + "id": FAKE_ROUTER_LINKT1_PORT_UUID, + "display_name": FAKE_NAME, "logical_router_id": FAKE_ROUTER_UUID } diff --git a/vmware_nsxlib/tests/unit/v3/test_resources.py b/vmware_nsxlib/tests/unit/v3/test_resources.py index ea33dd45..47b2d8c2 100644 --- a/vmware_nsxlib/tests/unit/v3/test_resources.py +++ b/vmware_nsxlib/tests/unit/v3/test_resources.py @@ -887,7 +887,8 @@ class LogicalRouterPortTestCase(BaseTestResource): 'tags': [], 'service_bindings': [{'service_id': { 'target_type': 'LogicalService', - 'target_id': fake_relay_uuid}}] + 'target_id': fake_relay_uuid}}], + 'linked_logical_switch_port_id': {'target_id': None} } with mock.patch("vmware_nsxlib.v3.NsxLib.get_version", @@ -922,21 +923,14 @@ class LogicalRouterPortTestCase(BaseTestResource): mock.patch("vmware_nsxlib.v3.NsxLib.get_version", return_value='2.0.0'): lrport.update(uuid, relay_service_uuid=fake_relay_uuid) - data = { - 'id': uuid, - 'display_name': fake_router_port['display_name'], - 'logical_router_id': fake_router_port['logical_router_id'], - 'resource_type': fake_router_port['resource_type'], - "revision": 0, - 'service_bindings': [{'service_id': { - 'target_type': 'LogicalService', - 'target_id': fake_relay_uuid}}] - } + fake_router_port['service_bindings'] = [{'service_id': { + 'target_type': 'LogicalService', + 'target_id': fake_relay_uuid}}] test_client.assert_json_call( 'put', lrport, 'https://1.2.3.4/api/v1/logical-router-ports/%s' % uuid, - data=jsonutils.dumps(data, sort_keys=True), + data=jsonutils.dumps(fake_router_port, sort_keys=True), headers=self.default_headers()) def test_get_logical_router_port_by_router_id(self): @@ -971,6 +965,66 @@ class LogicalRouterPortTestCase(BaseTestResource): 'logical_switch_id=%s' % switch_id, headers=self.default_headers()) + def test_get_tier1_link_port(self): + """Test getting a Tier0 router uplink port by router id.""" + router_id = test_constants.FAKE_ROUTER_PORT['logical_router_id'] + + # No ports found - raise an exception + lrport = self.get_mocked_resource(response={'results': []}) + self.assertRaises(exceptions.ResourceNotFound, + lrport.get_tier1_link_port, + router_id) + + # Non uplink ports found - raise an exception + lrport = self.get_mocked_resource(response={'results': [ + test_constants.FAKE_ROUTER_PORT]}) + self.assertRaises(exceptions.ResourceNotFound, + lrport.get_tier1_link_port, + router_id) + + # uplink port exists + lrport = self.get_mocked_resource(response={'results': [ + test_constants.FAKE_ROUTER_LINKT1_PORT]}) + result = lrport.get_tier1_link_port(router_id) + self.assertEqual(test_constants.FAKE_ROUTER_LINKT1_PORT, result) + + def test_get_tier0_uplink_port(self): + """Test getting a Tier0 router uplink port by router id.""" + router_id = test_constants.FAKE_ROUTER_PORT['logical_router_id'] + + # No ports found - return None + lrport = self.get_mocked_resource(response={'results': []}) + result = lrport.get_tier0_uplink_port(router_id) + self.assertIsNone(result) + + # Non uplink ports found - return None + lrport = self.get_mocked_resource(response={'results': [ + test_constants.FAKE_ROUTER_LINKT1_PORT]}) + result = lrport.get_tier0_uplink_port(router_id) + self.assertIsNone(result) + + # uplink port exists + lrport = self.get_mocked_resource(response={'results': [ + test_constants.FAKE_ROUTER_PORT]}) + result = lrport.get_tier0_uplink_port(router_id) + self.assertEqual(test_constants.FAKE_ROUTER_PORT, result) + + def test_get_tier0_uplink_port_ips(self): + """Test getting a Tier0 router uplink port by router id.""" + router_id = test_constants.FAKE_ROUTER_PORT['logical_router_id'] + + # No ports found - return empty list + lrport = self.get_mocked_resource(response={'results': []}) + result = lrport.get_tier0_uplink_ips(router_id) + self.assertEqual(0, len(result)) + + # uplink port exists, return ips + lrport = self.get_mocked_resource(response={'results': [ + test_constants.FAKE_ROUTER_PORT]}) + result = lrport.get_tier0_uplink_ips(router_id) + self.assertEqual(1, len(result)) + self.assertEqual('172.20.1.60', result[0]) + class IpPoolTestCase(BaseTestResource): diff --git a/vmware_nsxlib/v3/nsx_constants.py b/vmware_nsxlib/v3/nsx_constants.py index 8c2ab30b..87f85d26 100644 --- a/vmware_nsxlib/v3/nsx_constants.py +++ b/vmware_nsxlib/v3/nsx_constants.py @@ -40,7 +40,7 @@ BRIDGE_ENDPOINT = "BRIDGEENDPOINT" ROUTER_TYPE_TIER0 = "TIER0" ROUTER_TYPE_TIER1 = "TIER1" -LROUTERPORT_UPLINK = "LogicalRouterUplinkPort" +LROUTERPORT_UPLINK = "LogicalRouterUpLinkPort" LROUTERPORT_DOWNLINK = "LogicalRouterDownLinkPort" LROUTERPORT_CENTRALIZED = "LogicalRouterCentralizedServicePort" LROUTERPORT_LINKONTIER0 = "LogicalRouterLinkPortOnTIER0" diff --git a/vmware_nsxlib/v3/resources.py b/vmware_nsxlib/v3/resources.py index a289e053..00c969f9 100644 --- a/vmware_nsxlib/v3/resources.py +++ b/vmware_nsxlib/v3/resources.py @@ -343,6 +343,21 @@ class LogicalRouterPort(utils.NsxLibApiBase): manager=self.client.nsx_api_managers, operation="get router link port") + def get_tier0_uplink_port(self, logical_router_id): + logical_router_ports = self.get_by_router_id(logical_router_id) + for port in logical_router_ports: + if port['resource_type'] == nsx_constants.LROUTERPORT_UPLINK: + return port + + def get_tier0_uplink_ips(self, logical_router_id): + port = self.get_tier0_uplink_port(logical_router_id) + ips = [] + if port: + for subnet in port.get('subnets', []): + for ip_address in subnet.get('ip_addresses'): + ips.append(ip_address) + return ips + class MetaDataProxy(core_resources.NsxLibMetadataProxy): # TODO(asarfaty): keeping this for backwards compatibility.