NBI component - TFS API: (31551b44) · Commits · TFS / controller

src/common/type_checkers/Assertions.py

+50 −2

Original line number	Diff line number	Diff line
		@@ -465,7 +465,7 @@ def validate_slice(message):
		assert 'owner_string' in message['slice_owner']
		assert isinstance(message['slice_owner']['owner_string'], str)

		def validate_topology(message, num_devices=None, num_links=None):
		def validate_topology(message, num_devices=None, num_links=None, num_optical_links=None):
		assert isinstance(message, dict)
		assert len(message.keys()) == 5
		assert 'topology_id' in message
		@@ -482,7 +482,7 @@ def validate_topology(message, num_devices=None, num_links=None):
		for link_id in message['link_ids']: validate_link_id(link_id)
		assert 'optical_link_ids' in message
		assert isinstance(message['optical_link_ids'], list)
		#if num_links is not None: assert len(message['optical_link_ids']) == num_links
		if num_optical_links is not None: assert len(message['optical_link_ids']) == num_optical_links
		for link_id in message['optical_link_ids']: validate_link_id(link_id)

		def validate_endpoint(message):
		@@ -575,6 +575,47 @@ def validate_link(message):
		assert 'link_type' in message
		validate_link_type_enum(message['link_type'])

		def validate_optical_slot_map(message):
		assert isinstance(message, dict)
		for key, value in message.items():
		assert isinstance(key, str)
		assert isinstance(value, int)

		def validate_optical_link_details(message):
		assert isinstance(message, dict)
		assert len(message.keys()) == 9
		assert 'length' in message
		assert isinstance(message['length'], (int, float))
		assert 'src_port' in message
		assert isinstance(message['src_port'], str)
		assert 'dst_port' in message
		assert isinstance(message['dst_port'], str)
		assert 'local_peer_port' in message
		assert isinstance(message['local_peer_port'], str)
		assert 'remote_peer_port' in message
		assert isinstance(message['remote_peer_port'], str)
		assert 'used' in message
		assert isinstance(message['used'], bool)
		assert 'c_slots' in message
		validate_optical_slot_map(message['c_slots'])
		assert 'l_slots' in message
		validate_optical_slot_map(message['l_slots'])
		assert 's_slots' in message
		validate_optical_slot_map(message['s_slots'])

		def validate_optical_link(message):
		assert isinstance(message, dict)
		assert len(message.keys()) == 4
		assert 'link_id' in message
		validate_link_id(message['link_id'])
		assert 'name' in message
		assert isinstance(message['name'], str)
		assert 'link_endpoint_ids' in message
		assert isinstance(message['link_endpoint_ids'], list)
		for endpoint_id in message['link_endpoint_ids']: validate_endpoint_id(endpoint_id)
		assert 'optical_details' in message
		validate_optical_link_details(message['optical_details'])

		def validate_connection(message):
		assert isinstance(message, dict)
		assert len(message.keys()) in {4, 5}
		@@ -678,6 +719,13 @@ def validate_links(message):
		assert isinstance(message['links'], list)
		for link in message['links']: validate_link(link)

		def validate_optical_links(message):
		assert isinstance(message, dict)
		assert len(message.keys()) == 1
		assert 'optical_links' in message
		assert isinstance(message['optical_links'], list)
		for optical_link in message['optical_links']: validate_optical_link(optical_link)

		def validate_connections(message):
		assert isinstance(message, dict)
		assert len(message.keys()) == 1

src/nbi/tests/data/tfs_api_dummy.json

+26 −0

Original line number	Diff line number	Diff line
		@@ -232,6 +232,32 @@
		]
		}
		],
		"optical_links": [
		{
		"link_id": {"link_uuid": {"uuid": "OL:R1/502==R2/501"}}, "name": "OL:R1/502==R2/501",
		"link_endpoint_ids": [
		{
		"device_id": {"device_uuid": {"uuid": "R1"}}, "endpoint_uuid": {"uuid": "502"},
		"topology_id": {"context_id": {"context_uuid": {"uuid": "admin"}}, "topology_uuid": {"uuid": "admin"}}
		},
		{
		"device_id": {"device_uuid": {"uuid": "R2"}}, "endpoint_uuid": {"uuid": "501"},
		"topology_id": {"context_id": {"context_uuid": {"uuid": "admin"}}, "topology_uuid": {"uuid": "admin"}}
		}
		],
		"optical_details": {
		"length": 1.0,
		"src_port": "502",
		"dst_port": "501",
		"local_peer_port": "502",
		"remote_peer_port": "501",
		"used": true,
		"c_slots": {"0": 0, "1": 0, "2": 0},
		"l_slots": {"0": 0, "1": 0, "2": 0},
		"s_slots": {"0": 0, "1": 0, "2": 0}
		}
		}
		],
		"services": [
		{
		"service_id" : {"context_id": {"context_uuid": {"uuid": "admin"}}, "service_uuid": {"uuid": "SVC:R1/200==R2/200"}},

src/nbi/tests/test_tfs_api.py

+16 −1

Original line number	Diff line number	Diff line
		@@ -31,6 +31,7 @@ from common.type_checkers.Assertions import (
		validate_context, validate_context_ids, validate_contexts,
		validate_device, validate_device_ids, validate_devices,
		validate_link, validate_link_ids, validate_links,
		validate_optical_link, validate_optical_links,
		validate_service, validate_service_ids, validate_services,
		validate_slice, validate_slice_ids, validate_slices,
		validate_topologies, validate_topology, validate_topology_ids
		@@ -113,7 +114,7 @@ def test_rest_get_topology(
		context_uuid = urllib.parse.quote(DEFAULT_CONTEXT_NAME)
		topology_uuid = urllib.parse.quote(DEFAULT_TOPOLOGY_NAME)
		reply = do_rest_get_request('/tfs-api/context/{:s}/topology/{:s}'.format(context_uuid, topology_uuid))
		validate_topology(reply, num_devices=3, num_links=6)
		validate_topology(reply, num_devices=3, num_links=6, num_optical_links=1)


		# ----- Device ---------------------------------------------------------------------------------------------------------
		@@ -159,6 +160,20 @@ def test_rest_get_link(
		reply = do_rest_get_request('/tfs-api/link/{:s}'.format(link_uuid))
		validate_link(reply)

		def test_rest_get_optical_links(
		nbi_application : NbiApplication # pylint: disable=redefined-outer-name
		) -> None:
		reply = do_rest_get_request('/tfs-api/optical_links')
		validate_optical_links(reply)
		assert len(reply['optical_links']) == 1

		def test_rest_get_optical_link(
		nbi_application : NbiApplication # pylint: disable=redefined-outer-name
		) -> None:
		link_uuid = urllib.parse.quote('OL:R1/502==R2/501', safe='')
		reply = do_rest_get_request('/tfs-api/optical_link/{:s}'.format(link_uuid))
		validate_optical_link(reply)


		# ----- Service --------------------------------------------------------------------------------------------------------