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
18 sys.path.append('transportpce_tests/common')
19 import test_utils_rfc8040 # nopep8
22 class TransportPCE400GPortMappingTesting(unittest.TestCase):
29 cls.processes = test_utils_rfc8040.start_tpce()
30 cls.processes = test_utils_rfc8040.start_sims([('xpdra2', cls.NODE_VERSION)])
33 def tearDownClass(cls):
34 # pylint: disable=not-an-iterable
35 for process in cls.processes:
36 test_utils_rfc8040.shutdown_process(process)
37 print("all processes killed")
40 print("execution of {}".format(self.id().split(".")[-1]))
43 def test_01_xpdr_device_connection(self):
44 response = test_utils_rfc8040.mount_device("XPDR-A2",
45 ('xpdra2', self.NODE_VERSION))
46 self.assertEqual(response.status_code, requests.codes.created,
47 test_utils_rfc8040.CODE_SHOULD_BE_201)
49 # Check if the node appears in the ietf-network topology
50 def test_02_xpdr_device_connected(self):
51 response = test_utils_rfc8040.check_device_connection("XPDR-A2")
52 self.assertEqual(response['status_code'], requests.codes.ok)
53 self.assertEqual(response['connection-status'], 'connected')
56 # Check node info in the port-mappings
57 def test_03_xpdr_portmapping_info(self):
58 response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A2")
59 self.assertEqual(response['status_code'], requests.codes.ok)
62 'node-ip-address': '1.2.3.4',
64 'openroadm-version': '7.1',
65 'node-vendor': 'vendorA',
66 'node-model': 'model'},
67 response['node-info'])
70 # Check the if-capabilities and the other details for network
71 def test_04_tpdr_portmapping_NETWORK1(self):
72 response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR1-NETWORK1")
73 self.assertEqual(response['status_code'], requests.codes.ok)
75 {'supported-interface-capability':
76 ['org-openroadm-port-types:if-otsi-otsigroup'],
77 'supporting-port': 'L1',
78 'supporting-circuit-pack-name': '1/1/2-PLUG-NET',
79 'logical-connection-point': 'XPDR1-NETWORK1',
80 'port-qual': 'xpdr-network',
81 'port-direction': 'bidirectional',
82 'connection-map-lcp': 'XPDR1-CLIENT1',
83 'lcp-hash-val': 'AIGiVAQ4gDil',
84 'port-admin-state': 'InService',
85 'port-oper-state': 'InService',
86 'xponder-type': 'tpdr'
90 def test_05_tpdr_portmapping_CLIENT1(self):
91 response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR1-CLIENT1")
92 self.assertEqual(response['status_code'], requests.codes.ok)
94 {'supported-interface-capability': ['org-openroadm-port-types:if-400GE'],
95 'supporting-port': 'C1',
96 'supporting-circuit-pack-name': '1/1/1-PLUG-CLIENT',
97 'logical-connection-point': 'XPDR1-CLIENT1',
98 'port-direction': 'bidirectional',
99 'connection-map-lcp': 'XPDR1-NETWORK1',
100 'port-qual': 'xpdr-client',
101 'lcp-hash-val': 'AODABTVSOHH0',
102 'port-admin-state': 'InService',
103 'port-oper-state': 'InService',
104 'xponder-type': 'tpdr'
109 # Check the port-mapping for the switch-client and switch-network port-quals
110 def test_06_mpdr_portmapping_NETWORK1(self):
111 response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1")
112 self.assertEqual(response['status_code'], requests.codes.ok)
114 {'supported-interface-capability':
115 ['org-openroadm-port-types:if-otsi-otsigroup'],
116 'supporting-port': 'L1',
117 'supporting-circuit-pack-name': '1/2/2-PLUG-NET',
118 'logical-connection-point': 'XPDR2-NETWORK1',
119 'port-qual': 'switch-network',
120 'port-direction': 'bidirectional',
121 'lcp-hash-val': 'LY9PxYJqUbw=',
122 'port-admin-state': 'InService',
123 'port-oper-state': 'InService',
124 'xponder-type': 'mpdr'
128 def test_07_mpdr_portmapping_CLIENT1(self):
129 res = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-CLIENT1")
130 self.assertEqual(res['status_code'], requests.codes.ok)
131 self.assertIn('org-openroadm-port-types:if-100GE-ODU4',
132 res['mapping'][0]['supported-interface-capability'])
133 self.assertIn('org-openroadm-port-types:if-OCH-OTU4-ODU4',
134 res['mapping'][0]['supported-interface-capability'])
135 self.assertEqual('C1', res['mapping'][0]['supporting-port'])
136 self.assertEqual('1/2/1/1-PLUG-CLIENT', res['mapping'][0]['supporting-circuit-pack-name'])
137 self.assertEqual('XPDR2-CLIENT1', res['mapping'][0]['logical-connection-point'])
138 self.assertEqual('bidirectional', res['mapping'][0]['port-direction'])
139 self.assertEqual('switch-client', res['mapping'][0]['port-qual'])
140 self.assertEqual('AK+Cna4EclRH', res['mapping'][0]['lcp-hash-val'])
141 self.assertEqual('InService', res['mapping'][0]['port-admin-state'])
142 self.assertEqual('InService', res['mapping'][0]['port-oper-state'])
143 self.assertEqual('mpdr', res['mapping'][0]['xponder-type'])
145 "min-trib-slot": "1.1",
146 "max-trib-slot": "1.20"
147 }, res['mapping'][0]['mpdr-restrictions'])
149 # Added test to check mc-capability-profile for a transponder
150 def test_08_check_mccapprofile(self):
151 res = test_utils_rfc8040.portmapping_mc_capa_request("XPDR-A2", "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(res['mc-capabilities'][0]['center-freq-granularity'], '3.125')
155 self.assertEqual(res['mc-capabilities'][0]['slot-width-granularity'], '6.25')
157 def test_09_xpdr_device_disconnection(self):
158 response = test_utils_rfc8040.unmount_device("XPDR-A2")
159 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
161 def test_10_xpdr_device_disconnected(self):
162 response = test_utils_rfc8040.check_device_connection("XPDR-A2")
163 self.assertEqual(response['status_code'], requests.codes.conflict)
165 {"error-tag": "data-missing",
166 "error-message": "Request could not be completed because the relevant data model content does not exist",
167 "error-type": "protocol"},
168 response['connection-status'])
170 def test_11_xpdr_device_not_connected(self):
171 response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A2")
172 self.assertEqual(response['status_code'], requests.codes.conflict)
174 {"error-tag": "data-missing",
175 "error-message": "Request could not be completed because the relevant data model content does not exist",
176 "error-type": "protocol"},
177 response['node-info'])
180 if __name__ == '__main__':
181 unittest.main(verbosity=2)