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