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              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
  91             res['mapping'])
  92
  93     def test_05_rdm_portmapping_SRG1_PP7_TXRX(self):
  94         response = test_utils.portmapping_request("ROADMA01/mapping/SRG1-PP7-TXRX")
  95         self.assertEqual(response.status_code, requests.codes.ok)
  96         res = response.json()
  97         self.assertIn(
  98             {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
  99              'logical-connection-point': 'SRG1-PP7-TXRX', 'port-direction': 'bidirectional',
 100              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
 101             res['mapping'])
 102
 103     def test_06_rdm_portmapping_SRG3_PP1_TXRX(self):
 104         response = test_utils.portmapping_request("ROADMA01/mapping/SRG3-PP1-TXRX")
 105         self.assertEqual(response.status_code, requests.codes.ok)
 106         res = response.json()
 107         self.assertIn(
 108             {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0',
 109              'logical-connection-point': 'SRG3-PP1-TXRX', 'port-direction': 'bidirectional',
 110              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
 111             res['mapping'])
 112
 113     def test_07_xpdr_device_connection(self):
 114         response = test_utils.mount_device("XPDRA01", 'xpdra')
 115         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 116
 117     def test_08_xpdr_device_connected(self):
 118         response = test_utils.get_netconf_oper_request("XPDRA01")
 119         self.assertEqual(response.status_code, requests.codes.ok)
 120         res = response.json()
 121         self.assertEqual(
 122             res['node'][0]['netconf-node-topology:connection-status'],
 123             'connected')
 124         time.sleep(10)
 125
 126     def test_09_xpdr_portmapping_info(self):
 127         response = test_utils.portmapping_request("XPDRA01/node-info")
 128         self.assertEqual(response.status_code, requests.codes.ok)
 129         res = response.json()
 130         self.assertEqual(
 131             {u'node-info': {u'node-type': u'xpdr',
 132                             u'node-ip-address': u'127.0.0.10',
 133                             u'node-clli': u'NodeA',
 134                             u'openroadm-version': u'1.2.1', u'node-vendor': u'vendorA',
 135                             u'node-model': u'1'}},
 136             res)
 137         time.sleep(3)
 138
 139     def test_10_xpdr_portmapping_NETWORK1(self):
 140         response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-NETWORK1")
 141         self.assertEqual(response.status_code, requests.codes.ok)
 142         res = response.json()
 143         self.assertIn(
 144             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 145              'logical-connection-point': 'XPDR1-NETWORK1', 'port-direction': 'bidirectional',
 146              'connection-map-lcp': 'XPDR1-CLIENT1', 'port-qual': 'xpdr-network',
 147              'lcp-hash-val': 'OSvMgUyP+mE=',
 148              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
 149             res['mapping'])
 150
 151     def test_11_xpdr_portmapping_NETWORK2(self):
 152         response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-NETWORK2")
 153         self.assertEqual(response.status_code, requests.codes.ok)
 154         res = response.json()
 155         self.assertIn(
 156             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET',
 157              'logical-connection-point': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
 158              'connection-map-lcp': 'XPDR1-CLIENT3', 'port-qual': 'xpdr-network',
 159              'lcp-hash-val': 'OSvMgUyP+mI=',
 160              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
 161             res['mapping'])
 162
 163     def test_12_xpdr_portmapping_CLIENT1(self):
 164         response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT1")
 165         self.assertEqual(response.status_code, requests.codes.ok)
 166         res = response.json()
 167         self.assertIn(
 168             {'supporting-port': 'C1',
 169              'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
 170              'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
 171              'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client',
 172              'lcp-hash-val': 'AO9UFkY/TLYw',
 173              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
 174             res['mapping'])
 175
 176     def test_13_xpdr_portmapping_CLIENT2(self):
 177         response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT2")
 178         self.assertEqual(response.status_code, requests.codes.ok)
 179         res = response.json()
 180         self.assertIn(
 181             {'supporting-port': 'C2',
 182              'supporting-circuit-pack-name': '1/0/C2-PLUG-CLIENT',
 183              'logical-connection-point': 'XPDR1-CLIENT2', 'port-direction': 'bidirectional',
 184              'port-qual': 'xpdr-client',
 185              'lcp-hash-val': 'AO9UFkY/TLYz',
 186              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
 187             res['mapping'])
 188
 189     def test_14_xpdr_portmapping_CLIENT3(self):
 190         response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT3")
 191         self.assertEqual(response.status_code, requests.codes.ok)
 192         res = response.json()
 193         self.assertIn(
 194             {'supporting-port': 'C3',
 195              'supporting-circuit-pack-name': '1/0/C3-PLUG-CLIENT',
 196              'logical-connection-point': 'XPDR1-CLIENT3',
 197              'connection-map-lcp': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
 198              'port-qual': 'xpdr-client', 'lcp-hash-val': 'AO9UFkY/TLYy',
 199              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
 200             res['mapping'])
 201
 202     def test_15_xpdr_portmapping_CLIENT4(self):
 203         response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT4")
 204         self.assertEqual(response.status_code, requests.codes.ok)
 205         res = response.json()
 206         self.assertIn(
 207             {'supporting-port': 'C4',
 208              'supporting-circuit-pack-name': '1/0/C4-PLUG-CLIENT',
 209              'logical-connection-point': 'XPDR1-CLIENT4', 'port-direction': 'bidirectional',
 210              'port-qual': 'xpdr-client', 'lcp-hash-val': 'AO9UFkY/TLY1',
 211              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
 212             res['mapping'])
 213
 214     def test_16_xpdr_device_disconnection(self):
 215         response = test_utils.unmount_device("XPDRA01")
 216         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 217
 218     def test_17_xpdr_device_disconnected(self):
 219         response = test_utils.get_netconf_oper_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_18_xpdr_device_not_connected(self):
 228         response = test_utils.portmapping_request("XPDRA01")
 229         self.assertEqual(response.status_code, requests.codes.conflict)
 230         res = response.json()
 231         self.assertIn(
 232             {"error-type": "application", "error-tag": "data-missing",
 233              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 234             res['errors']['error'])
 235
 236     def test_19_rdm_device_disconnection(self):
 237         response = test_utils.unmount_device("ROADMA01")
 238         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 239
 240     def test_20_rdm_device_disconnected(self):
 241         response = test_utils.get_netconf_oper_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     def test_21_rdm_device_not_connected(self):
 250         response = test_utils.portmapping_request("ROADMA01")
 251         self.assertEqual(response.status_code, requests.codes.conflict)
 252         res = response.json()
 253         self.assertIn(
 254             {"error-type": "application", "error-tag": "data-missing",
 255              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 256             res['errors']['error'])
 257
 258
 259 if __name__ == "__main__":
 260     unittest.main(verbosity=2)