51694b5683ef88a8eddf2a4c221e381626fd37bd
[transportpce.git] / tests / transportpce_tests / 1.2.1 / test_portmapping.py
1 #!/usr/bin/env python
2
3 ##############################################################################
4 # Copyright (c) 2017 Orange, Inc. and others.  All rights reserved.
5 #
6 # All rights reserved. This program and the accompanying materials
7 # are made available under the terms of the Apache License, Version 2.0
8 # which accompanies this distribution, and is available at
9 # http://www.apache.org/licenses/LICENSE-2.0
10 ##############################################################################
11
12 # pylint: disable=no-member
13 # pylint: disable=too-many-public-methods
14
15 import unittest
16 import time
17 import requests
18 from common import test_utils
19
20
21 class TransportPCEPortMappingTesting(unittest.TestCase):
22
23     processes = None
24
25     @classmethod
26     def setUpClass(cls):
27         cls.processes = test_utils.start_tpce()
28         cls.processes = test_utils.start_sims(['xpdra', 'roadma'])
29
30     @classmethod
31     def tearDownClass(cls):
32         # pylint: disable=not-an-iterable
33         for process in cls.processes:
34             test_utils.shutdown_process(process)
35         print("all processes killed")
36
37     def setUp(self):
38         print("execution of {}".format(self.id().split(".")[-1]))
39         time.sleep(10)
40
41 #    def test_01_restconfAPI(self):
42 #        response = test_utils.get_netconf_oper_request("controller-config")
43 #        self.assertEqual(response.status_code, requests.codes.ok)
44 #        res = response.json()
45 #        self.assertEqual(res['node'] [0] ['netconf-node-topology:connection-status'],
46 #                         'connected')
47
48 #     def test_02_restconfAPI(self):
49 #         response = test_utils.portmapping_request("controller-config")
50 #         self.assertEqual(response.status_code, requests.codes.not_found)
51 #         res = response.json()
52 #         self.assertIn(
53 #             {"error-type":"application", "error-tag":"data-missing",
54 #             "error-message":"Request could not be completed because the relevant data model content does not exist "},
55 #             res['errors']['error'])
56
57     def test_01_rdm_device_connection(self):
58         response = test_utils.mount_device("ROADMA01", 'roadma')
59         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
60
61     def test_02_rdm_device_connected(self):
62         response = test_utils.get_netconf_oper_request("ROADMA01")
63         self.assertEqual(response.status_code, requests.codes.ok)
64         res = response.json()
65         self.assertEqual(
66             res['node'][0]['netconf-node-topology:connection-status'],
67             'connected')
68         time.sleep(10)
69
70     def test_03_rdm_portmapping_info(self):
71         response = test_utils.portmapping_request("ROADMA01/node-info")
72         self.assertEqual(response.status_code, requests.codes.ok)
73         res = response.json()
74         self.assertEqual(
75             {u'node-info': {u'node-type': u'rdm',
76                             u'node-ip-address': u'127.0.0.12',
77                             u'node-clli': u'NodeA',
78                             u'openroadm-version': u'1.2.1', u'node-vendor': u'vendorA',
79                             u'node-model': u'2'}},
80             res)
81         time.sleep(3)
82
83     def test_04_rdm_portmapping_DEG1_TTP_TXRX(self):
84         response = test_utils.portmapping_request("ROADMA01/mapping/DEG1-TTP-TXRX")
85         self.assertEqual(response.status_code, requests.codes.ok)
86         res = response.json()
87         self.assertIn(
88             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
89              'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional'},
90             res['mapping'])
91
92     def test_05_rdm_portmapping_SRG1_PP7_TXRX(self):
93         response = test_utils.portmapping_request("ROADMA01/mapping/SRG1-PP7-TXRX")
94         self.assertEqual(response.status_code, requests.codes.ok)
95         res = response.json()
96         self.assertIn(
97             {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
98              'logical-connection-point': 'SRG1-PP7-TXRX', 'port-direction': 'bidirectional'},
99             res['mapping'])
100
101     def test_06_rdm_portmapping_SRG3_PP1_TXRX(self):
102         response = test_utils.portmapping_request("ROADMA01/mapping/SRG3-PP1-TXRX")
103         self.assertEqual(response.status_code, requests.codes.ok)
104         res = response.json()
105         self.assertIn(
106             {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0',
107              'logical-connection-point': 'SRG3-PP1-TXRX', 'port-direction': 'bidirectional'},
108             res['mapping'])
109
110     def test_07_xpdr_device_connection(self):
111         response = test_utils.mount_device("XPDRA01", 'xpdra')
112         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
113
114     def test_08_xpdr_device_connected(self):
115         response = test_utils.get_netconf_oper_request("XPDRA01")
116         self.assertEqual(response.status_code, requests.codes.ok)
117         res = response.json()
118         self.assertEqual(
119             res['node'][0]['netconf-node-topology:connection-status'],
120             'connected')
121         time.sleep(10)
122
123     def test_09_xpdr_portmapping_info(self):
124         response = test_utils.portmapping_request("XPDRA01/node-info")
125         self.assertEqual(response.status_code, requests.codes.ok)
126         res = response.json()
127         self.assertEqual(
128             {u'node-info': {u'node-type': u'xpdr',
129                             u'node-ip-address': u'127.0.0.10',
130                             u'node-clli': u'NodeA',
131                             u'openroadm-version': u'1.2.1', u'node-vendor': u'vendorA',
132                             u'node-model': u'1'}},
133             res)
134         time.sleep(3)
135
136     def test_10_xpdr_portmapping_NETWORK1(self):
137         response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-NETWORK1")
138         self.assertEqual(response.status_code, requests.codes.ok)
139         res = response.json()
140         self.assertIn(
141             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
142              'logical-connection-point': 'XPDR1-NETWORK1', 'port-direction': 'bidirectional',
143              'connection-map-lcp': 'XPDR1-CLIENT1', 'port-qual': 'xpdr-network',
144              'lcp-hash-val': 'OSvMgUyP+mE='},
145             res['mapping'])
146
147     def test_11_xpdr_portmapping_NETWORK2(self):
148         response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-NETWORK2")
149         self.assertEqual(response.status_code, requests.codes.ok)
150         res = response.json()
151         self.assertIn(
152             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET',
153              'logical-connection-point': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
154              'connection-map-lcp': 'XPDR1-CLIENT3', 'port-qual': 'xpdr-network',
155              'lcp-hash-val': 'OSvMgUyP+mI='},
156             res['mapping'])
157
158     def test_12_xpdr_portmapping_CLIENT1(self):
159         response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT1")
160         self.assertEqual(response.status_code, requests.codes.ok)
161         res = response.json()
162         self.assertIn(
163             {'supporting-port': 'C1',
164              'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
165              'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
166              'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client',
167              'lcp-hash-val': 'AO9UFkY/TLYw'},
168             res['mapping'])
169
170     def test_13_xpdr_portmapping_CLIENT2(self):
171         response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT2")
172         self.assertEqual(response.status_code, requests.codes.ok)
173         res = response.json()
174         self.assertIn(
175             {'supporting-port': 'C2',
176              'supporting-circuit-pack-name': '1/0/C2-PLUG-CLIENT',
177              'logical-connection-point': 'XPDR1-CLIENT2', 'port-direction': 'bidirectional',
178              'port-qual': 'xpdr-client',
179              'lcp-hash-val': 'AO9UFkY/TLYz'},
180             res['mapping'])
181
182     def test_14_xpdr_portmapping_CLIENT3(self):
183         response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT3")
184         self.assertEqual(response.status_code, requests.codes.ok)
185         res = response.json()
186         self.assertIn(
187             {'supporting-port': 'C3',
188              'supporting-circuit-pack-name': '1/0/C3-PLUG-CLIENT',
189              'logical-connection-point': 'XPDR1-CLIENT3',
190              'connection-map-lcp': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
191              'port-qual': 'xpdr-client', 'lcp-hash-val': 'AO9UFkY/TLYy'},
192             res['mapping'])
193
194     def test_15_xpdr_portmapping_CLIENT4(self):
195         response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT4")
196         self.assertEqual(response.status_code, requests.codes.ok)
197         res = response.json()
198         self.assertIn(
199             {'supporting-port': 'C4',
200              'supporting-circuit-pack-name': '1/0/C4-PLUG-CLIENT',
201              'logical-connection-point': 'XPDR1-CLIENT4', 'port-direction': 'bidirectional',
202              'port-qual': 'xpdr-client', 'lcp-hash-val': 'AO9UFkY/TLY1'},
203             res['mapping'])
204
205     def test_16_xpdr_device_disconnection(self):
206         response = test_utils.unmount_device("XPDRA01")
207         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
208
209     def test_17_xpdr_device_disconnected(self):
210         response = test_utils.get_netconf_oper_request("XPDRA01")
211         self.assertEqual(response.status_code, requests.codes.conflict)
212         res = response.json()
213         self.assertIn(
214             {"error-type": "application", "error-tag": "data-missing",
215              "error-message": "Request could not be completed because the relevant data model content does not exist"},
216             res['errors']['error'])
217
218     def test_18_xpdr_device_not_connected(self):
219         response = test_utils.portmapping_request("XPDRA01")
220         self.assertEqual(response.status_code, requests.codes.conflict)
221         res = response.json()
222         self.assertIn(
223             {"error-type": "application", "error-tag": "data-missing",
224              "error-message": "Request could not be completed because the relevant data model content does not exist"},
225             res['errors']['error'])
226
227     def test_19_rdm_device_disconnection(self):
228         response = test_utils.unmount_device("ROADMA01")
229         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
230
231     def test_20_rdm_device_disconnected(self):
232         response = test_utils.get_netconf_oper_request("ROADMA01")
233         self.assertEqual(response.status_code, requests.codes.conflict)
234         res = response.json()
235         self.assertIn(
236             {"error-type": "application", "error-tag": "data-missing",
237              "error-message": "Request could not be completed because the relevant data model content does not exist"},
238             res['errors']['error'])
239
240     def test_21_rdm_device_not_connected(self):
241         response = test_utils.portmapping_request("ROADMA01")
242         self.assertEqual(response.status_code, requests.codes.conflict)
243         res = response.json()
244         self.assertIn(
245             {"error-type": "application", "error-tag": "data-missing",
246              "error-message": "Request could not be completed because the relevant data model content does not exist"},
247             res['errors']['error'])
248
249
250 if __name__ == "__main__":
251     unittest.main(verbosity=2)