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 print("execution of {}".format(self.id().split(".")[-1]))
46 def test_01_xpdr_device_connection(self):
47 response = test_utils_rfc8040.mount_device("XPDR-A2",
48 ('xpdra2', self.NODE_VERSION))
49 self.assertEqual(response.status_code, requests.codes.created,
50 test_utils_rfc8040.CODE_SHOULD_BE_201)
52 # Check if the node appears in the ietf-network topology
53 def test_02_xpdr_device_connected(self):
54 response = test_utils_rfc8040.check_device_connection("XPDR-A2")
55 self.assertEqual(response['status_code'], requests.codes.ok)
56 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_info("XPDR-A2")
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'])
73 # Check the if-capabilities and the other details for network
74 def test_04_tpdr_portmapping_NETWORK1(self):
75 response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR1-NETWORK1")
76 self.assertEqual(response['status_code'], requests.codes.ok)
78 {'supported-interface-capability':
79 ['org-openroadm-port-types:if-otsi-otsigroup'],
80 'supporting-port': 'L1',
81 'supporting-circuit-pack-name': '1/1/2-PLUG-NET',
82 'logical-connection-point': 'XPDR1-NETWORK1',
83 'port-qual': 'xpdr-network',
84 'port-direction': 'bidirectional',
85 'connection-map-lcp': 'XPDR1-CLIENT1',
86 'lcp-hash-val': 'AIGiVAQ4gDil',
87 'port-admin-state': 'InService',
88 'port-oper-state': 'InService',
89 'xponder-type': 'tpdr'
93 def test_05_tpdr_portmapping_CLIENT1(self):
94 response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR1-CLIENT1")
95 self.assertEqual(response['status_code'], requests.codes.ok)
97 {'supported-interface-capability': ['org-openroadm-port-types:if-400GE'],
98 'supporting-port': 'C1',
99 'supporting-circuit-pack-name': '1/1/1-PLUG-CLIENT',
100 'logical-connection-point': 'XPDR1-CLIENT1',
101 'port-direction': 'bidirectional',
102 'connection-map-lcp': 'XPDR1-NETWORK1',
103 'port-qual': 'xpdr-client',
104 'lcp-hash-val': 'AODABTVSOHH0',
105 'port-admin-state': 'InService',
106 'port-oper-state': 'InService',
107 'xponder-type': 'tpdr'
112 # Check the port-mapping for the switch-client and switch-network port-quals
113 def test_06_mpdr_portmapping_NETWORK1(self):
114 response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1")
115 self.assertEqual(response['status_code'], requests.codes.ok)
117 {'supported-interface-capability':
118 ['org-openroadm-port-types:if-otsi-otsigroup'],
119 'supporting-port': 'L1',
120 'supporting-circuit-pack-name': '1/2/2-PLUG-NET',
121 'logical-connection-point': 'XPDR2-NETWORK1',
122 'port-qual': 'switch-network',
123 'port-direction': 'bidirectional',
124 'lcp-hash-val': 'LY9PxYJqUbw=',
125 'port-admin-state': 'InService',
126 'port-oper-state': 'InService',
127 'xponder-type': 'mpdr'
131 def test_07_mpdr_portmapping_CLIENT1(self):
132 res = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-CLIENT1")
133 self.assertEqual(res['status_code'], requests.codes.ok)
134 self.assertIn('org-openroadm-port-types:if-100GE-ODU4',
135 res['mapping'][0]['supported-interface-capability'])
136 self.assertIn('org-openroadm-port-types:if-OCH-OTU4-ODU4',
137 res['mapping'][0]['supported-interface-capability'])
138 self.assertEqual('C1', res['mapping'][0]['supporting-port'])
139 self.assertEqual('1/2/1/1-PLUG-CLIENT', res['mapping'][0]['supporting-circuit-pack-name'])
140 self.assertEqual('XPDR2-CLIENT1', res['mapping'][0]['logical-connection-point'])
141 self.assertEqual('bidirectional', res['mapping'][0]['port-direction'])
142 self.assertEqual('switch-client', res['mapping'][0]['port-qual'])
143 self.assertEqual('AK+Cna4EclRH', res['mapping'][0]['lcp-hash-val'])
144 self.assertEqual('InService', res['mapping'][0]['port-admin-state'])
145 self.assertEqual('InService', res['mapping'][0]['port-oper-state'])
146 self.assertEqual('mpdr', res['mapping'][0]['xponder-type'])
148 "min-trib-slot": "1.1",
149 "max-trib-slot": "1.20"
150 }, res['mapping'][0]['mpdr-restrictions'])
152 # Added test to check mc-capability-profile for a transponder
153 def test_08_check_mccapprofile(self):
154 res = test_utils_rfc8040.portmapping_mc_capa_request("XPDR-A2", "XPDR-mcprofile")
155 self.assertEqual(res['status_code'], requests.codes.ok)
156 self.assertEqual(res['mc-capabilities'][0]['mc-node-name'], 'XPDR-mcprofile')
157 self.assertEqual(res['mc-capabilities'][0]['center-freq-granularity'], '3.125')
158 self.assertEqual(res['mc-capabilities'][0]['slot-width-granularity'], '6.25')
160 def test_09_xpdr_device_disconnection(self):
161 response = test_utils_rfc8040.unmount_device("XPDR-A2")
162 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
164 def test_10_xpdr_device_disconnected(self):
165 response = test_utils_rfc8040.check_device_connection("XPDR-A2")
166 self.assertEqual(response['status_code'], requests.codes.conflict)
168 {"error-tag": "data-missing",
169 "error-message": "Request could not be completed because the relevant data model content does not exist",
170 "error-type": "protocol"},
171 response['connection-status'])
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)
177 {"error-tag": "data-missing",
178 "error-message": "Request could not be completed because the relevant data model content does not exist",
179 "error-type": "protocol"},
180 response['node-info'])
183 if __name__ == '__main__':
184 unittest.main(verbosity=2)