tests/transportpce_tests/2.2.1/test_portmapping.py

   1 #!/usr/bin/env python
   2
   3 ##############################################################################
   4 # Copyright (c) 2019 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_rdm_device_connection(self):
  42         response = test_utils.mount_device("ROADM-A1", 'roadma')
  43         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  44
  45     def test_02_rdm_device_connected(self):
  46         response = test_utils.get_netconf_oper_request("ROADM-A1")
  47         self.assertEqual(response.status_code, requests.codes.ok)
  48         res = response.json()
  49         self.assertEqual(
  50             res['node'][0]['netconf-node-topology:connection-status'],
  51             'connected')
  52         time.sleep(10)
  53
  54     def test_03_rdm_portmapping_info(self):
  55         response = test_utils.portmapping_request("ROADM-A1/node-info")
  56         self.assertEqual(response.status_code, requests.codes.ok)
  57         res = response.json()
  58         self.assertEqual(
  59             {u'node-info': {u'node-type': u'rdm',
  60                             u'node-ip-address': u'127.0.0.11',
  61                             u'node-clli': u'NodeA',
  62                             u'openroadm-version': u'2.2.1', u'node-vendor': u'vendorA',
  63                             u'node-model': u'model2'}},
  64             res)
  65         time.sleep(3)
  66
  67     def test_04_rdm_portmapping_DEG1_TTP_TXRX(self):
  68         response = test_utils.portmapping_request("ROADM-A1/mapping/DEG1-TTP-TXRX")
  69         self.assertEqual(response.status_code, requests.codes.ok)
  70         res = response.json()
  71         self.assertIn(
  72             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '1/0',
  73              'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional'},
  74             res['mapping'])
  75
  76     def test_05_rdm_portmapping_DEG2_TTP_TXRX_with_ots_oms(self):
  77         response = test_utils.portmapping_request("ROADM-A1/mapping/DEG2-TTP-TXRX")
  78         self.assertEqual(response.status_code, requests.codes.ok)
  79         res = response.json()
  80         self.assertIn(
  81             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
  82              'logical-connection-point': 'DEG2-TTP-TXRX',
  83              'supporting-oms': 'OMS-DEG2-TTP-TXRX', 'supporting-ots': 'OTS-DEG2-TTP-TXRX',
  84              'port-direction': 'bidirectional'},
  85             res['mapping'])
  86
  87     def test_06_rdm_portmapping_SRG1_PP3_TXRX(self):
  88         response = test_utils.portmapping_request("ROADM-A1/mapping/SRG1-PP3-TXRX")
  89         self.assertEqual(response.status_code, requests.codes.ok)
  90         res = response.json()
  91         self.assertIn(
  92             {'supporting-port': 'C3', 'supporting-circuit-pack-name': '3/0',
  93              'logical-connection-point': 'SRG1-PP3-TXRX', 'port-direction': 'bidirectional'},
  94             res['mapping'])
  95
  96     def test_07_rdm_portmapping_SRG3_PP1_TXRX(self):
  97         response = test_utils.portmapping_request("ROADM-A1/mapping/SRG3-PP1-TXRX")
  98         self.assertEqual(response.status_code, requests.codes.ok)
  99         res = response.json()
 100         self.assertIn(
 101             {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0',
 102              'logical-connection-point': 'SRG3-PP1-TXRX', 'port-direction': 'bidirectional'},
 103             res['mapping'])
 104
 105     def test_08_xpdr_device_connection(self):
 106         response = test_utils.mount_device("XPDR-A1", 'xpdra')
 107         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 108
 109     def test_09_xpdr_device_connected(self):
 110         response = test_utils.get_netconf_oper_request("XPDR-A1")
 111         self.assertEqual(response.status_code, requests.codes.ok)
 112         res = response.json()
 113         self.assertEqual(
 114             res['node'][0]['netconf-node-topology:connection-status'],
 115             'connected')
 116         time.sleep(10)
 117
 118     def test_10_xpdr_portmapping_info(self):
 119         response = test_utils.portmapping_request("XPDR-A1/node-info")
 120         self.assertEqual(response.status_code, requests.codes.ok)
 121         res = response.json()
 122         self.assertEqual(
 123             {u'node-info': {u'node-type': u'xpdr',
 124                             u'node-ip-address': u'1.2.3.4',
 125                             u'node-clli': u'NodeA',
 126                             u'openroadm-version': u'2.2.1', u'node-vendor': u'vendorA',
 127                             u'node-model': u'model2'}},
 128             res)
 129         time.sleep(3)
 130
 131     def test_11_xpdr_portmapping_NETWORK1(self):
 132         response = test_utils.portmapping_request("XPDR-A1/mapping/XPDR1-NETWORK1")
 133         self.assertEqual(response.status_code, requests.codes.ok)
 134         res = response.json()
 135         self.assertIn(
 136             {'supported-interface-capability': ['org-openroadm-port-types:if-OCH'],
 137              'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 138              'logical-connection-point': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-network',
 139              'port-direction': 'bidirectional', 'connection-map-lcp': 'XPDR1-CLIENT1',
 140              'lcp-hash-val': 'AMkDwQ7xTmRI'},
 141             res['mapping'])
 142
 143     def test_12_xpdr_portmapping_NETWORK2(self):
 144         response = test_utils.portmapping_request("XPDR-A1/mapping/XPDR1-NETWORK2")
 145         self.assertEqual(response.status_code, requests.codes.ok)
 146         res = response.json()
 147         self.assertIn(
 148             {'supported-interface-capability': ['org-openroadm-port-types:if-OCH'],
 149              'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET',
 150              'logical-connection-point': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
 151              'connection-map-lcp': 'XPDR1-CLIENT2', 'port-qual': 'xpdr-network',
 152              'lcp-hash-val': 'AMkDwQ7xTmRL'},
 153             res['mapping'])
 154
 155     def test_13_xpdr_portmapping_CLIENT1(self):
 156         response = test_utils.portmapping_request("XPDR-A1/mapping/XPDR1-CLIENT1")
 157         self.assertEqual(response.status_code, requests.codes.ok)
 158         res = response.json()
 159         self.assertIn(
 160             {'supported-interface-capability': ['org-openroadm-port-types:if-100GE'],
 161              'supporting-port': 'C1',
 162              'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT',
 163              'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
 164              'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client',
 165              'lcp-hash-val': 'AJUUr6I5fALj'},
 166             res['mapping'])
 167
 168     def test_14_xpdr_portmapping_CLIENT2(self):
 169         response = test_utils.portmapping_request("XPDR-A1/mapping/XPDR1-CLIENT2")
 170         self.assertEqual(response.status_code, requests.codes.ok)
 171         res = response.json()
 172         self.assertIn(
 173             {'supported-interface-capability': ['org-openroadm-port-types:if-100GE'],
 174              'supporting-port': 'C1',
 175              'supporting-circuit-pack-name': '1/0/2-PLUG-CLIENT',
 176              'logical-connection-point': 'XPDR1-CLIENT2', 'port-direction': 'bidirectional',
 177              'connection-map-lcp': 'XPDR1-NETWORK2', 'port-qual': 'xpdr-client',
 178              'lcp-hash-val': 'AJUUr6I5fALg'},
 179             res['mapping'])
 180
 181     def test_15_xpdr_device_disconnection(self):
 182         response = test_utils.unmount_device("XPDR-A1")
 183         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 184
 185     def test_16_xpdr_device_disconnected(self):
 186         response = test_utils.get_netconf_oper_request("XPDR-A1")
 187         self.assertEqual(response.status_code, requests.codes.conflict)
 188         res = response.json()
 189         self.assertIn(
 190             {"error-type": "application", "error-tag": "data-missing",
 191              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 192             res['errors']['error'])
 193
 194     def test_17_xpdr_device_not_connected(self):
 195         response = test_utils.portmapping_request("XPDR-A1")
 196         self.assertEqual(response.status_code, requests.codes.conflict)
 197         res = response.json()
 198         self.assertIn(
 199             {"error-type": "application", "error-tag": "data-missing",
 200              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 201             res['errors']['error'])
 202
 203     def test_18_rdm_device_disconnection(self):
 204         response = test_utils.unmount_device("ROADM-A1")
 205         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 206
 207     def test_19_rdm_device_disconnected(self):
 208         response = test_utils.get_netconf_oper_request("ROADM-A1")
 209         self.assertEqual(response.status_code, requests.codes.conflict)
 210         res = response.json()
 211         self.assertIn(
 212             {"error-type": "application", "error-tag": "data-missing",
 213              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 214             res['errors']['error'])
 215
 216     def test_20_rdm_device_not_connected(self):
 217         response = test_utils.portmapping_request("ROADM-A1")
 218         self.assertEqual(response.status_code, requests.codes.conflict)
 219         res = response.json()
 220         self.assertIn(
 221             {"error-type": "application", "error-tag": "data-missing",
 222              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 223             res['errors']['error'])
 224
 225
 226 if __name__ == "__main__":
 227     unittest.main(verbosity=2)