2 ##############################################################################
3 # Copyright (c) 2021 AT&T, Inc. and others. All rights reserved.
5 # All rights reserved. This program and the accompanying materials
6 # are made available under the terms of the Apache License, Version 2.0
7 # which accompanies this distribution, and is available at
8 # http://www.apache.org/licenses/LICENSE-2.0
9 ##############################################################################
11 # pylint: disable=no-member
12 # pylint: disable=too-many-public-methods
17 # pylint: disable=wrong-import-order
19 sys.path.append('transportpce_tests/common')
20 # pylint: disable=wrong-import-position
21 # pylint: disable=import-error
22 import test_utils_rfc8040 # nopep8
25 class TransportPCE400GPortMappingTesting(unittest.TestCase):
32 cls.processes = test_utils_rfc8040.start_tpce()
33 cls.processes = test_utils_rfc8040.start_sims([('xpdra2', cls.NODE_VERSION)])
36 def tearDownClass(cls):
37 # pylint: disable=not-an-iterable
38 for process in cls.processes:
39 test_utils_rfc8040.shutdown_process(process)
40 print("all processes killed")
43 # pylint: disable=consider-using-f-string
44 print("execution of {}".format(self.id().split(".")[-1]))
47 def test_01_xpdr_device_connection(self):
48 response = test_utils_rfc8040.mount_device("XPDR-A2",
49 ('xpdra2', self.NODE_VERSION))
50 self.assertEqual(response.status_code, requests.codes.created,
51 test_utils_rfc8040.CODE_SHOULD_BE_201)
53 # Check if the node appears in the ietf-network topology
54 def test_02_xpdr_device_connected(self):
55 response = test_utils_rfc8040.check_device_connection("XPDR-A2")
56 self.assertEqual(response['status_code'], requests.codes.ok)
57 self.assertEqual(response['connection-status'], 'connected')
59 # Check node info in the port-mappings
60 def test_03_xpdr_portmapping_info(self):
61 response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "node-info", None)
62 self.assertEqual(response['status_code'], requests.codes.ok)
65 'node-ip-address': '1.2.3.4',
67 'openroadm-version': '7.1',
68 'node-vendor': 'vendorA',
69 'node-model': 'model'},
70 response['node-info'])
72 # Check the if-capabilities and the other details for network
73 def test_04_tpdr_portmapping_NETWORK1(self):
74 response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR1-NETWORK1")
75 self.assertEqual(response['status_code'], requests.codes.ok)
77 {'supported-interface-capability':
78 ['org-openroadm-port-types:if-otsi-otsigroup'],
79 'supporting-port': 'L1',
80 'supporting-circuit-pack-name': '1/1/2-PLUG-NET',
81 'logical-connection-point': 'XPDR1-NETWORK1',
82 'port-qual': 'xpdr-network',
83 'port-direction': 'bidirectional',
84 'connection-map-lcp': 'XPDR1-CLIENT1',
85 'lcp-hash-val': 'AIGiVAQ4gDil',
86 'port-admin-state': 'InService',
87 'port-oper-state': 'InService',
88 'xponder-type': 'tpdr'
92 def test_05_tpdr_portmapping_CLIENT1(self):
93 response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR1-CLIENT1")
94 self.assertEqual(response['status_code'], requests.codes.ok)
96 {'supported-interface-capability': ['org-openroadm-port-types:if-400GE'],
97 'supporting-port': 'C1',
98 'supporting-circuit-pack-name': '1/1/1-PLUG-CLIENT',
99 'logical-connection-point': 'XPDR1-CLIENT1',
100 'port-direction': 'bidirectional',
101 'connection-map-lcp': 'XPDR1-NETWORK1',
102 'port-qual': 'xpdr-client',
103 'lcp-hash-val': 'AODABTVSOHH0',
104 'port-admin-state': 'InService',
105 'port-oper-state': 'InService',
106 'xponder-type': 'tpdr'
111 # Check the port-mapping for the switch-client and switch-network port-quals
112 def test_06_mpdr_portmapping_NETWORK1(self):
113 response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-NETWORK1")
114 self.assertEqual(response['status_code'], requests.codes.ok)
116 {'supported-interface-capability':
117 ['org-openroadm-port-types:if-otsi-otsigroup'],
118 'supporting-port': 'L1',
119 'supporting-circuit-pack-name': '1/2/2-PLUG-NET',
120 'logical-connection-point': 'XPDR2-NETWORK1',
121 'port-qual': 'switch-network',
122 'port-direction': 'bidirectional',
123 'lcp-hash-val': 'LY9PxYJqUbw=',
124 'port-admin-state': 'InService',
125 'port-oper-state': 'InService',
126 'xponder-type': 'mpdr'
130 def test_07_mpdr_portmapping_CLIENT1(self):
131 res = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mapping", "XPDR2-CLIENT1")
132 self.assertEqual(res['status_code'], requests.codes.ok)
133 self.assertIn('org-openroadm-port-types:if-100GE-ODU4',
134 res['mapping'][0]['supported-interface-capability'])
135 self.assertIn('org-openroadm-port-types:if-OCH-OTU4-ODU4',
136 res['mapping'][0]['supported-interface-capability'])
137 self.assertEqual('C1', res['mapping'][0]['supporting-port'])
138 self.assertEqual('1/2/1/1-PLUG-CLIENT', res['mapping'][0]['supporting-circuit-pack-name'])
139 self.assertEqual('XPDR2-CLIENT1', res['mapping'][0]['logical-connection-point'])
140 self.assertEqual('bidirectional', res['mapping'][0]['port-direction'])
141 self.assertEqual('switch-client', res['mapping'][0]['port-qual'])
142 self.assertEqual('AK+Cna4EclRH', res['mapping'][0]['lcp-hash-val'])
143 self.assertEqual('InService', res['mapping'][0]['port-admin-state'])
144 self.assertEqual('InService', res['mapping'][0]['port-oper-state'])
145 self.assertEqual('mpdr', res['mapping'][0]['xponder-type'])
146 self.assertEqual(1.1, float(res['mapping'][0]['mpdr-restrictions']['min-trib-slot']))
147 self.assertEqual(1.2, float(res['mapping'][0]['mpdr-restrictions']['max-trib-slot']))
149 # Added test to check mc-capability-profile for a transponder
150 def test_08_check_mccapprofile(self):
151 res = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "mc-capabilities", "XPDR-mcprofile")
152 self.assertEqual(res['status_code'], requests.codes.ok)
153 self.assertEqual(res['mc-capabilities'][0]['mc-node-name'], 'XPDR-mcprofile')
154 self.assertEqual(float(res['mc-capabilities'][0]['center-freq-granularity']), 3.125)
155 self.assertEqual(float(res['mc-capabilities'][0]['slot-width-granularity']), 6.25)
157 def test_09_mpdr_switching_pool(self):
158 response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "switching-pool-lcp", "1")
159 self.assertEqual(response['status_code'], requests.codes.ok)
160 self.assertEqual("blocking",
161 response['switching-pool-lcp'][0]['switching-pool-type'])
163 len(response['switching-pool-lcp'][0]['non-blocking-list']))
166 'interconnect-bandwidth': 0,
167 'lcp-list': ['XPDR2-NETWORK1', 'XPDR2-CLIENT2']},
168 response['switching-pool-lcp'][0]['non-blocking-list'])
170 def test_10_xpdr_device_disconnection(self):
171 response = test_utils_rfc8040.unmount_device("XPDR-A2")
172 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
174 def test_11_xpdr_device_disconnected(self):
175 response = test_utils_rfc8040.check_device_connection("XPDR-A2")
176 self.assertEqual(response['status_code'], requests.codes.conflict)
177 self.assertIn(response['connection-status']['error-type'], ('protocol', 'application'))
178 self.assertEqual(response['connection-status']['error-tag'], 'data-missing')
179 self.assertEqual(response['connection-status']['error-message'],
180 'Request could not be completed because the relevant data model content does not exist')
182 def test_12_xpdr_device_not_connected(self):
183 response = test_utils_rfc8040.get_portmapping_node_attr("XPDR-A2", "node-info", None)
184 self.assertEqual(response['status_code'], requests.codes.conflict)
185 self.assertIn(response['node-info']['error-type'], ('protocol', 'application'))
186 self.assertEqual(response['node-info']['error-tag'], 'data-missing')
187 self.assertEqual(response['node-info']['error-message'],
188 'Request could not be completed because the relevant data model content does not exist')
191 if __name__ == '__main__':
192 unittest.main(verbosity=2)