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')
60 # Check node info in the port-mappings
61 def test_03_xpdr_portmapping_info(self):
62 response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A2")
63 self.assertEqual(response['status_code'], requests.codes.ok)
66 'node-ip-address': '1.2.3.4',
68 'openroadm-version': '7.1',
69 'node-vendor': 'vendorA',
70 'node-model': 'model'},
71 response['node-info'])
74 # Check the if-capabilities and the other details for network
75 def test_04_tpdr_portmapping_NETWORK1(self):
76 response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR1-NETWORK1")
77 self.assertEqual(response['status_code'], requests.codes.ok)
79 {'supported-interface-capability':
80 ['org-openroadm-port-types:if-otsi-otsigroup'],
81 'supporting-port': 'L1',
82 'supporting-circuit-pack-name': '1/1/2-PLUG-NET',
83 'logical-connection-point': 'XPDR1-NETWORK1',
84 'port-qual': 'xpdr-network',
85 'port-direction': 'bidirectional',
86 'connection-map-lcp': 'XPDR1-CLIENT1',
87 'lcp-hash-val': 'AIGiVAQ4gDil',
88 'port-admin-state': 'InService',
89 'port-oper-state': 'InService',
90 'xponder-type': 'tpdr'
94 def test_05_tpdr_portmapping_CLIENT1(self):
95 response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR1-CLIENT1")
96 self.assertEqual(response['status_code'], requests.codes.ok)
98 {'supported-interface-capability': ['org-openroadm-port-types:if-400GE'],
99 'supporting-port': 'C1',
100 'supporting-circuit-pack-name': '1/1/1-PLUG-CLIENT',
101 'logical-connection-point': 'XPDR1-CLIENT1',
102 'port-direction': 'bidirectional',
103 'connection-map-lcp': 'XPDR1-NETWORK1',
104 'port-qual': 'xpdr-client',
105 'lcp-hash-val': 'AODABTVSOHH0',
106 'port-admin-state': 'InService',
107 'port-oper-state': 'InService',
108 'xponder-type': 'tpdr'
113 # Check the port-mapping for the switch-client and switch-network port-quals
114 def test_06_mpdr_portmapping_NETWORK1(self):
115 response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1")
116 self.assertEqual(response['status_code'], requests.codes.ok)
118 {'supported-interface-capability':
119 ['org-openroadm-port-types:if-otsi-otsigroup'],
120 'supporting-port': 'L1',
121 'supporting-circuit-pack-name': '1/2/2-PLUG-NET',
122 'logical-connection-point': 'XPDR2-NETWORK1',
123 'port-qual': 'switch-network',
124 'port-direction': 'bidirectional',
125 'lcp-hash-val': 'LY9PxYJqUbw=',
126 'port-admin-state': 'InService',
127 'port-oper-state': 'InService',
128 'xponder-type': 'mpdr'
132 def test_07_mpdr_portmapping_CLIENT1(self):
133 res = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-CLIENT1")
134 self.assertEqual(res['status_code'], requests.codes.ok)
135 self.assertIn('org-openroadm-port-types:if-100GE-ODU4',
136 res['mapping'][0]['supported-interface-capability'])
137 self.assertIn('org-openroadm-port-types:if-OCH-OTU4-ODU4',
138 res['mapping'][0]['supported-interface-capability'])
139 self.assertEqual('C1', res['mapping'][0]['supporting-port'])
140 self.assertEqual('1/2/1/1-PLUG-CLIENT', res['mapping'][0]['supporting-circuit-pack-name'])
141 self.assertEqual('XPDR2-CLIENT1', res['mapping'][0]['logical-connection-point'])
142 self.assertEqual('bidirectional', res['mapping'][0]['port-direction'])
143 self.assertEqual('switch-client', res['mapping'][0]['port-qual'])
144 self.assertEqual('AK+Cna4EclRH', res['mapping'][0]['lcp-hash-val'])
145 self.assertEqual('InService', res['mapping'][0]['port-admin-state'])
146 self.assertEqual('InService', res['mapping'][0]['port-oper-state'])
147 self.assertEqual('mpdr', res['mapping'][0]['xponder-type'])
149 "min-trib-slot": "1.1",
150 "max-trib-slot": "1.20"
151 }, res['mapping'][0]['mpdr-restrictions'])
153 # Added test to check mc-capability-profile for a transponder
154 def test_08_check_mccapprofile(self):
155 res = test_utils_rfc8040.portmapping_mc_capa_request("XPDR-A2", "XPDR-mcprofile")
156 self.assertEqual(res['status_code'], requests.codes.ok)
157 self.assertEqual(res['mc-capabilities'][0]['mc-node-name'], 'XPDR-mcprofile')
158 self.assertEqual(str(res['mc-capabilities'][0]['center-freq-granularity']), '3.125')
159 self.assertEqual(str(res['mc-capabilities'][0]['slot-width-granularity']), '6.25')
161 def test_09_xpdr_device_disconnection(self):
162 response = test_utils_rfc8040.unmount_device("XPDR-A2")
163 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
165 def test_10_xpdr_device_disconnected(self):
166 response = test_utils_rfc8040.check_device_connection("XPDR-A2")
167 self.assertEqual(response['status_code'], requests.codes.conflict)
168 self.assertIn(response['connection-status']['error-type'], ('protocol', 'application'))
169 self.assertEqual(response['connection-status']['error-tag'], 'data-missing')
170 self.assertEqual(response['connection-status']['error-message'],
171 'Request could not be completed because the relevant data model content does not exist')
173 def test_11_xpdr_device_not_connected(self):
174 response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A2")
175 self.assertEqual(response['status_code'], requests.codes.conflict)
176 self.assertIn(response['node-info']['error-type'], ('protocol', 'application'))
177 self.assertEqual(response['node-info']['error-tag'], 'data-missing')
178 self.assertEqual(response['node-info']['error-message'],
179 'Request could not be completed because the relevant data model content does not exist')
182 if __name__ == '__main__':
183 unittest.main(verbosity=2)