X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F1.2.1%2Ftest_end2end.py;h=3f7ca3b52aff45306c1664bf6eba47d95da1c572;hb=c834356f10cb8bc04c8b0a9eb2a438b2e9c50632;hp=0e8e3041106a88292fa3ad8e78f6f95de023de3b;hpb=64975190383a3682d35b8e434b8be43606f1f81a;p=transportpce.git diff --git a/tests/transportpce_tests/1.2.1/test_end2end.py b/tests/transportpce_tests/1.2.1/test_end2end.py index 0e8e30411..3f7ca3b52 100644 --- a/tests/transportpce_tests/1.2.1/test_end2end.py +++ b/tests/transportpce_tests/1.2.1/test_end2end.py @@ -8,24 +8,25 @@ # http://www.apache.org/licenses/LICENSE-2.0 ############################################################################## +# pylint: disable=no-member +# pylint: disable=too-many-public-methods -import json +import base64 import time import unittest - import requests -from common import test_utils - +import sys +sys.path.append('transportpce_tests/common/') +import test_utils class TransportPCEFulltesting(unittest.TestCase): - cr_serv_sample_data = {"input": { "sdnc-request-header": { "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58", "rpc-action": "service-create", "request-system-id": "appname", "notification-url": - "http://localhost:8585/NotificationServer/notify" + "http://localhost:8585/NotificationServer/notify" }, "service-name": "service1", "common-id": "ASATT1234567", @@ -35,40 +36,40 @@ class TransportPCEFulltesting(unittest.TestCase): "node-id": "XPDRA01", "service-format": "Ethernet", "clli": "SNJSCAMCJP8", - "tx-direction": { - "port": { - "port-device-name": - "ROUTER_SNJSCAMCJP8_000000.00_00", - "port-type": "router", - "port-name": "Gigabit Ethernet_Tx.ge-5/0/0.0", - "port-rack": "000000.00", - "port-shelf": "00" - }, - "lgx": { - "lgx-device-name": - "LGX Panel_SNJSCAMCJP8_000000.00_00", - "lgx-port-name": "LGX Back.3", - "lgx-port-rack": "000000.00", - "lgx-port-shelf": "00" - } - }, + "tx-direction": { + "port": { + "port-device-name": + "ROUTER_SNJSCAMCJP8_000000.00_00", + "port-type": "router", + "port-name": "Gigabit Ethernet_Tx.ge-5/0/0.0", + "port-rack": "000000.00", + "port-shelf": "00" + }, + "lgx": { + "lgx-device-name": + "LGX Panel_SNJSCAMCJP8_000000.00_00", + "lgx-port-name": "LGX Back.3", + "lgx-port-rack": "000000.00", + "lgx-port-shelf": "00" + } + }, "rx-direction": { - "port": { - "port-device-name": - "ROUTER_SNJSCAMCJP8_000000.00_00", - "port-type": "router", - "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0", - "port-rack": "000000.00", - "port-shelf": "00" - }, - "lgx": { - "lgx-device-name": - "LGX Panel_SNJSCAMCJP8_000000.00_00", - "lgx-port-name": "LGX Back.4", - "lgx-port-rack": "000000.00", - "lgx-port-shelf": "00" - } - }, + "port": { + "port-device-name": + "ROUTER_SNJSCAMCJP8_000000.00_00", + "port-type": "router", + "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0", + "port-rack": "000000.00", + "port-shelf": "00" + }, + "lgx": { + "lgx-device-name": + "LGX Panel_SNJSCAMCJP8_000000.00_00", + "lgx-port-name": "LGX Back.4", + "lgx-port-rack": "000000.00", + "lgx-port-shelf": "00" + } + }, "optic-type": "gray" }, "service-z-end": { @@ -76,40 +77,40 @@ class TransportPCEFulltesting(unittest.TestCase): "node-id": "XPDRC01", "service-format": "Ethernet", "clli": "SNJSCAMCJT4", - "tx-direction": { - "port": { - "port-device-name": - "ROUTER_SNJSCAMCJT4_000000.00_00", - "port-type": "router", - "port-name": "Gigabit Ethernet_Tx.ge-1/0/0.0", - "port-rack": "000000.00", - "port-shelf": "00" - }, - "lgx": { - "lgx-device-name": - "LGX Panel_SNJSCAMCJT4_000000.00_00", - "lgx-port-name": "LGX Back.29", - "lgx-port-rack": "000000.00", - "lgx-port-shelf": "00" - } - }, + "tx-direction": { + "port": { + "port-device-name": + "ROUTER_SNJSCAMCJT4_000000.00_00", + "port-type": "router", + "port-name": "Gigabit Ethernet_Tx.ge-1/0/0.0", + "port-rack": "000000.00", + "port-shelf": "00" + }, + "lgx": { + "lgx-device-name": + "LGX Panel_SNJSCAMCJT4_000000.00_00", + "lgx-port-name": "LGX Back.29", + "lgx-port-rack": "000000.00", + "lgx-port-shelf": "00" + } + }, "rx-direction": { - "port": { - "port-device-name": - "ROUTER_SNJSCAMCJT4_000000.00_00", - "port-type": "router", - "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0", - "port-rack": "000000.00", - "port-shelf": "00" - }, - "lgx": { - "lgx-device-name": - "LGX Panel_SNJSCAMCJT4_000000.00_00", - "lgx-port-name": "LGX Back.30", - "lgx-port-rack": "000000.00", - "lgx-port-shelf": "00" - } - }, + "port": { + "port-device-name": + "ROUTER_SNJSCAMCJT4_000000.00_00", + "port-type": "router", + "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0", + "port-rack": "000000.00", + "port-shelf": "00" + }, + "lgx": { + "lgx-device-name": + "LGX Panel_SNJSCAMCJT4_000000.00_00", + "lgx-port-name": "LGX Back.30", + "lgx-port-rack": "000000.00", + "lgx-port-shelf": "00" + } + }, "optic-type": "gray" }, "due-date": "2016-11-28T00:00:01Z", @@ -118,14 +119,19 @@ class TransportPCEFulltesting(unittest.TestCase): } processes = None WAITING = 20 + NODE_VERSION = '1.2.1' @classmethod def setUpClass(cls): cls.processes = test_utils.start_tpce() - cls.processes = test_utils.start_sims(['xpdra', 'roadma-full', 'roadmc-full', 'xpdrc']) + cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION), + ('roadma-full', cls.NODE_VERSION), + ('roadmc-full', cls.NODE_VERSION), + ('xpdrc', cls.NODE_VERSION)]) @classmethod def tearDownClass(cls): + # pylint: disable=not-an-iterable for process in cls.processes: test_utils.shutdown_process(process) print("all processes killed") @@ -135,19 +141,19 @@ class TransportPCEFulltesting(unittest.TestCase): # connect netconf devices def test_01_connect_xpdrA(self): - response = test_utils.mount_device("XPDRA01", 'xpdra') + response = test_utils.mount_device("XPDRA01", ('xpdra', self.NODE_VERSION)) self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_02_connect_xpdrC(self): - response = test_utils.mount_device("XPDRC01", 'xpdrc') + response = test_utils.mount_device("XPDRC01", ('xpdrc', self.NODE_VERSION)) self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_03_connect_rdmA(self): - response = test_utils.mount_device("ROADMA01", 'roadma-full') + response = test_utils.mount_device("ROADMA01", ('roadma-full', self.NODE_VERSION)) self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_04_connect_rdmC(self): - response = test_utils.mount_device("ROADMC01", 'roadmc-full') + response = test_utils.mount_device("ROADMC01", ('roadmc-full', self.NODE_VERSION)) self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_05_connect_xprdA_N1_to_roadmA_PP1(self): @@ -189,7 +195,6 @@ class TransportPCEFulltesting(unittest.TestCase): def test_09_add_omsAttributes_ROADMA_ROADMC(self): # Config ROADMA-ROADMC oms-attributes data = {"span": { - "clfi": "fiber1", "auto-spanloss": "true", "spanloss-base": 11.4, "spanloss-current": 12, @@ -205,7 +210,6 @@ class TransportPCEFulltesting(unittest.TestCase): def test_10_add_omsAttributes_ROADMC_ROADMA(self): # Config ROADMC-ROADMA oms-attributes data = {"span": { - "clfi": "fiber1", "auto-spanloss": "true", "spanloss-base": 11.4, "spanloss-current": 12, @@ -245,13 +249,13 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(2) def test_13_check_xc1_ROADMA(self): - response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-1") + response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 self.assertDictEqual( dict({ - 'connection-number': 'SRG1-PP1-TXRX-DEG1-TTP-TXRX-1', + 'connection-number': 'SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768', 'wavelength-number': 1, 'opticalControlMode': 'gainLoss', 'target-output-power': -3.0 @@ -259,21 +263,21 @@ class TransportPCEFulltesting(unittest.TestCase): res['roadm-connections'][0] ) self.assertDictEqual( - {'src-if': 'SRG1-PP1-TXRX-1'}, + {'src-if': 'SRG1-PP1-TXRX-761:768'}, res['roadm-connections'][0]['source']) self.assertDictEqual( - {'dst-if': 'DEG1-TTP-TXRX-1'}, + {'dst-if': 'DEG1-TTP-TXRX-761:768'}, res['roadm-connections'][0]['destination']) time.sleep(5) def test_14_check_xc1_ROADMC(self): - response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-1") + response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 self.assertDictEqual( dict({ - 'connection-number': 'SRG1-PP1-TXRX-DEG2-TTP-TXRX-1', + 'connection-number': 'SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768', 'wavelength-number': 1, 'opticalControlMode': 'gainLoss', 'target-output-power': 2.0 @@ -281,10 +285,10 @@ class TransportPCEFulltesting(unittest.TestCase): res['roadm-connections'][0] ) self.assertDictEqual( - {'src-if': 'SRG1-PP1-TXRX-1'}, + {'src-if': 'SRG1-PP1-TXRX-761:768'}, res['roadm-connections'][0]['source']) self.assertDictEqual( - {'dst-if': 'DEG2-TTP-TXRX-1'}, + {'dst-if': 'DEG2-TTP-TXRX-761:768'}, res['roadm-connections'][0]['destination']) time.sleep(5) @@ -295,58 +299,54 @@ class TransportPCEFulltesting(unittest.TestCase): liste_tp = res['node'][0]['ietf-network-topology:termination-point'] for ele in liste_tp: if ele['tp-id'] == 'XPDR1-NETWORK1': - self.assertEqual({u'frequency': 196.1, u'width': 40}, + self.assertEqual({u'frequency': 196.1, + u'width': 40}, ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength']) - if ele['tp-id'] == 'XPDR1-CLIENT1' or ele['tp-id'] == 'XPDR1-CLIENT3': - self.assertNotIn( - 'org-openroadm-network-topology:xpdr-client-attributes', - dict.keys(ele)) + if ele['tp-id'] == 'XPDR1-CLIENT2' or ele['tp-id'] == 'XPDR1-CLIENT1': + self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele)) if ele['tp-id'] == 'XPDR1-NETWORK2': - self.assertNotIn( - 'org-openroadm-network-topology:xpdr-network-attributes', - dict.keys(ele)) + self.assertNotIn('org-openroadm-network-topology:xpdr-network-attributes', dict.keys(ele)) time.sleep(3) def test_16_check_topo_ROADMA_SRG1(self): response = test_utils.get_ordm_topo_request("node/ROADMA01-SRG1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() - self.assertNotIn({u'index': 1}, - res['node'][0][ - u'org-openroadm-network-topology:srg-attributes'][ - 'available-wavelengths']) + freq_map = base64.b64decode( + res['node'][0]['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map']) + freq_map_array = [int(x) for x in freq_map] + self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available") liste_tp = res['node'][0]['ietf-network-topology:termination-point'] for ele in liste_tp: if ele['tp-id'] == 'SRG1-PP1-TXRX': - self.assertIn({u'index': 1, u'frequency': 196.1, u'width': 40}, - ele['org-openroadm-network-topology:' - 'pp-attributes']['used-wavelength'] - ) + freq_map = base64.b64decode( + ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map']) + freq_map_array = [int(x) for x in freq_map] + self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available") if ele['tp-id'] == 'SRG1-PP2-TXRX': - self.assertNotIn('used-wavelength', dict.keys(ele)) + self.assertNotIn('avail-freq-maps', dict.keys(ele)) time.sleep(3) def test_17_check_topo_ROADMA_DEG1(self): response = test_utils.get_ordm_topo_request("node/ROADMA01-DEG1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() - self.assertNotIn({u'index': 1}, - res['node'][0][ - u'org-openroadm-network-topology:' - u'degree-attributes'][ - 'available-wavelengths']) + freq_map = base64.b64decode( + res['node'][0]['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map']) + freq_map_array = [int(x) for x in freq_map] + self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available") liste_tp = res['node'][0]['ietf-network-topology:termination-point'] for ele in liste_tp: - if ele['tp-id'] == 'DEG1-CTP-TXRX': - self.assertIn({u'index': 1, u'frequency': 196.1, u'width': 40}, - ele['org-openroadm-network-topology:' - 'ctp-attributes'][ - 'used-wavelengths']) - if ele['tp-id'] == 'DEG1-TTP-TXRX': - self.assertIn({u'index': 1, u'frequency': 196.1, u'width': 40}, - ele['org-openroadm-network-topology:' - 'tx-ttp-attributes'][ - 'used-wavelengths']) + if ele['tp-id'] == 'DEG2-CTP-TXRX': + freq_map = base64.b64decode( + ele['org-openroadm-network-topology:ctp-attributes']['avail-freq-maps'][0]['freq-map']) + freq_map_array = [int(x) for x in freq_map] + self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available") + if ele['tp-id'] == 'DEG2-TTP-TXRX': + freq_map = base64.b64decode( + ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map']) + freq_map_array = [int(x) for x in freq_map] + self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available") time.sleep(3) def test_18_connect_xprdA_N2_to_roadmA_PP2(self): @@ -411,23 +411,23 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_24_check_xc2_ROADMA(self): - response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP2-TXRX-2") + response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP2-TXRX-753:760") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 self.assertDictEqual( dict({ - 'connection-number': 'DEG1-TTP-TXRX-SRG1-PP2-TXRX-2', - 'wavelength-number': 2, - 'opticalControlMode': 'power' - }, **res['roadm-connections'][0]), + 'connection-number': 'DEG1-TTP-TXRX-SRG1-PP2-TXRX-753:760', + 'wavelength-number': 2, + 'opticalControlMode': 'power' + }, **res['roadm-connections'][0]), res['roadm-connections'][0] ) self.assertDictEqual( - {'src-if': 'DEG1-TTP-TXRX-2'}, + {'src-if': 'DEG1-TTP-TXRX-753:760'}, res['roadm-connections'][0]['source']) self.assertDictEqual( - {'dst-if': 'SRG1-PP2-TXRX-2'}, + {'dst-if': 'SRG1-PP2-TXRX-753:760'}, res['roadm-connections'][0]['destination']) def test_25_check_topo_XPDRA(self): @@ -437,81 +437,67 @@ class TransportPCEFulltesting(unittest.TestCase): liste_tp = res['node'][0]['ietf-network-topology:termination-point'] for ele in liste_tp: if ele['tp-id'] == 'XPDR1-NETWORK1': - self.assertEqual({u'frequency': 196.1, u'width': 40}, - ele['org-openroadm-network-topology:' - 'xpdr-network-attributes'][ - 'wavelength']) + self.assertEqual({u'frequency': 196.1, + u'width': 40}, + ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength']) if ele['tp-id'] == 'XPDR1-NETWORK2': - self.assertEqual({u'frequency': 196.05, u'width': 40}, - ele['org-openroadm-network-topology:' - 'xpdr-network-attributes'][ - 'wavelength']) - if ele['tp-id'] == 'XPDR1-CLIENT1' or \ - ele['tp-id'] == 'XPDR1-CLIENT3': - self.assertNotIn( - 'org-openroadm-network-topology:xpdr-client-attributes', - dict.keys(ele)) + self.assertEqual({u'frequency': 196.05, + u'width': 40}, + ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength']) + if ele['tp-id'] == 'XPDR1-CLIENT1' or ele['tp-id'] == 'XPDR1-CLIENT2': + self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele)) time.sleep(10) def test_26_check_topo_ROADMA_SRG1(self): response = test_utils.get_ordm_topo_request("node/ROADMA01-SRG1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() - self.assertNotIn({u'index': 1}, res['node'][0][ - u'org-openroadm-network-topology:srg-attributes'][ - 'available-wavelengths']) - self.assertNotIn({u'index': 2}, res['node'][0][ - u'org-openroadm-network-topology:srg-attributes'][ - 'available-wavelengths']) + freq_map = base64.b64decode( + res['node'][0]['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map']) + freq_map_array = [int(x) for x in freq_map] + self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available") + self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available") liste_tp = res['node'][0]['ietf-network-topology:termination-point'] for ele in liste_tp: if ele['tp-id'] == 'SRG1-PP1-TXRX': - self.assertIn({u'index': 1, u'frequency': 196.1, u'width': 40}, - ele['org-openroadm-network-topology:' - 'pp-attributes']['used-wavelength']) - self.assertNotIn({u'index': 2, u'frequency': 196.05, - u'width': 40}, - ele['org-openroadm-network-topology:' - 'pp-attributes']['used-wavelength']) + freq_map = base64.b64decode( + ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map']) + freq_map_array = [int(x) for x in freq_map] + self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available") + self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available") if ele['tp-id'] == 'SRG1-PP2-TXRX': - self.assertIn({u'index': 2, u'frequency': 196.05, u'width': 40}, - ele['org-openroadm-network-topology:' - 'pp-attributes']['used-wavelength']) - self.assertNotIn({u'index': 1, u'frequency': 196.1, - u'width': 40}, - ele['org-openroadm-network-topology:' - 'pp-attributes']['used-wavelength']) + freq_map = base64.b64decode( + ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map']) + freq_map_array = [int(x) for x in freq_map] + self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available") + self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available") if ele['tp-id'] == 'SRG1-PP3-TXRX': - self.assertNotIn('org-openroadm-network-topology:pp-attributes', - dict.keys(ele)) + self.assertNotIn('org-openroadm-network-topology:pp-attributes', dict.keys(ele)) time.sleep(10) def test_27_check_topo_ROADMA_DEG1(self): response = test_utils.get_ordm_topo_request("node/ROADMA01-DEG1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() - self.assertNotIn({u'index': 1}, res['node'][0][ - u'org-openroadm-network-topology:degree-attributes'][ - 'available-wavelengths']) - self.assertNotIn({u'index': 2}, res['node'][0][ - u'org-openroadm-network-topology:degree-attributes'][ - 'available-wavelengths']) + freq_map = base64.b64decode( + res['node'][0]['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map']) + freq_map_array = [int(x) for x in freq_map] + self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available") + self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available") liste_tp = res['node'][0]['ietf-network-topology:termination-point'] for ele in liste_tp: - if ele['tp-id'] == 'DEG1-CTP-TXRX': - self.assertIn({u'index': 1, u'frequency': 196.1, u'width': 40}, - ele['org-openroadm-network-topology:' - 'ctp-attributes']['used-wavelengths']) - self.assertIn({u'index': 2, u'frequency': 196.05, u'width': 40}, - ele['org-openroadm-network-topology:' - 'ctp-attributes']['used-wavelengths']) - if ele['tp-id'] == 'DEG1-TTP-TXRX': - self.assertIn({u'index': 1, u'frequency': 196.1, u'width': 40}, - ele['org-openroadm-network-topology:' - 'tx-ttp-attributes']['used-wavelengths']) - self.assertIn({u'index': 2, u'frequency': 196.05, u'width': 40}, - ele['org-openroadm-network-topology:' - 'tx-ttp-attributes']['used-wavelengths']) + if ele['tp-id'] == 'DEG2-CTP-TXRX': + freq_map = base64.b64decode( + ele['org-openroadm-network-topology:ctp-attributes']['avail-freq-maps'][0]['freq-map']) + freq_map_array = [int(x) for x in freq_map] + self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available") + self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available") + if ele['tp-id'] == 'DEG2-TTP-TXRX': + freq_map = base64.b64decode( + ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map']) + freq_map_array = [int(x) for x in freq_map] + self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available") + self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available") time.sleep(10) # creation service test on a non-available resource @@ -575,8 +561,8 @@ class TransportPCEFulltesting(unittest.TestCase): for ele in liste_tp: if ((ele[u'org-openroadm-common-network:tp-type'] == 'XPONDER-CLIENT') - and (ele['tp-id'] == 'XPDR1-CLIENT1' or ele[ - 'tp-id'] == 'XPDR1-CLIENT3')): + and (ele['tp-id'] == 'XPDR1-CLIENT1' or ele[ + 'tp-id'] == 'XPDR1-CLIENT3')): self.assertNotIn( 'org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele)) @@ -594,41 +580,52 @@ class TransportPCEFulltesting(unittest.TestCase): response = test_utils.get_ordm_topo_request("node/ROADMA01-SRG1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() - self.assertIn({u'index': 1}, res['node'][0][ - u'org-openroadm-network-topology:srg-attributes'][ - 'available-wavelengths']) - self.assertIn({u'index': 2}, res['node'][0][ - u'org-openroadm-network-topology:srg-attributes'][ - 'available-wavelengths']) + freq_map = base64.b64decode( + res['node'][0]['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map']) + freq_map_array = [int(x) for x in freq_map] + self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available") + self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available") liste_tp = res['node'][0]['ietf-network-topology:termination-point'] for ele in liste_tp: - if ele['tp-id'] == 'SRG1-PP1-TXRX' or \ - ele['tp-id'] == 'SRG1-PP1-TXRX': - self.assertNotIn('org-openroadm-network-topology:pp-attributes', - dict.keys(ele)) + if ele['tp-id'] == 'SRG1-PP1-TXRX' or ele['tp-id'] == 'SRG1-PP2-TXRX': + freq_map = base64.b64decode( + ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map']) + freq_map_array = [int(x) for x in freq_map] + self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available") + self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available") + elif ele['tp-id'] == 'SRG1-CP-TXRX': + freq_map = base64.b64decode( + ele['org-openroadm-network-topology:cp-attributes']['avail-freq-maps'][0]['freq-map']) + freq_map_array = [int(x) for x in freq_map] + self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available") + self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available") else: - self.assertNotIn('org-openroadm-network-topology:pp-attributes', - dict.keys(ele)) + self.assertNotIn('org-openroadm-network-topology:pp-attributes', dict.keys(ele)) time.sleep(10) def test_35_check_topo_ROADMA_DEG1(self): response = test_utils.get_ordm_topo_request("node/ROADMA01-DEG1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() - self.assertIn({u'index': 1}, res['node'][0][ - u'org-openroadm-network-topology:degree-attributes'][ - 'available-wavelengths']) - self.assertIn({u'index': 2}, res['node'][0][ - u'org-openroadm-network-topology:degree-attributes'][ - 'available-wavelengths']) + freq_map = base64.b64decode( + res['node'][0]['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map']) + freq_map_array = [int(x) for x in freq_map] + self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available") + self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available") liste_tp = res['node'][0]['ietf-network-topology:termination-point'] for ele in liste_tp: - if ele['tp-id'] == 'DEG1-CTP-TXRX': - self.assertNotIn('org-openroadm-network-topology:' - 'ctp-attributes', dict.keys(ele)) - if ele['tp-id'] == 'DEG1-TTP-TXRX': - self.assertNotIn('org-openroadm-network-topology:' - 'tx-ttp-attributes', dict.keys(ele)) + if ele['tp-id'] == 'DEG2-CTP-TXRX': + freq_map = base64.b64decode( + ele['org-openroadm-network-topology:ctp-attributes']['avail-freq-maps'][0]['freq-map']) + freq_map_array = [int(x) for x in freq_map] + self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available") + self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available") + if ele['tp-id'] == 'DEG2-TTP-TXRX': + freq_map = base64.b64decode( + ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map']) + freq_map_array = [int(x) for x in freq_map] + self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available") + self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available") time.sleep(10) # test service-create for Optical Channel (OC) service from srg-pp to srg-pp @@ -663,13 +660,13 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(1) def test_38_check_xc1_ROADMA(self): - response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-1") + response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 self.assertDictEqual( dict({ - 'connection-number': 'SRG1-PP1-TXRX-DEG1-TTP-TXRX-1', + 'connection-number': 'SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768', 'wavelength-number': 1, 'opticalControlMode': 'gainLoss', 'target-output-power': -3.0 @@ -677,21 +674,21 @@ class TransportPCEFulltesting(unittest.TestCase): res['roadm-connections'][0] ) self.assertDictEqual( - {'src-if': 'SRG1-PP1-TXRX-1'}, + {'src-if': 'SRG1-PP1-TXRX-761:768'}, res['roadm-connections'][0]['source']) self.assertDictEqual( - {'dst-if': 'DEG1-TTP-TXRX-1'}, + {'dst-if': 'DEG1-TTP-TXRX-761:768'}, res['roadm-connections'][0]['destination']) time.sleep(7) def test_39_check_xc1_ROADMC(self): - response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-1") + response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 self.assertDictEqual( dict({ - 'connection-number': 'SRG1-PP1-TXRX-DEG2-TTP-TXRX-1', + 'connection-number': 'SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768', 'wavelength-number': 1, 'opticalControlMode': 'gainLoss', 'target-output-power': 2.0 @@ -699,10 +696,10 @@ class TransportPCEFulltesting(unittest.TestCase): res['roadm-connections'][0] ) self.assertDictEqual( - {'src-if': 'SRG1-PP1-TXRX-1'}, + {'src-if': 'SRG1-PP1-TXRX-761:768'}, res['roadm-connections'][0]['source']) self.assertDictEqual( - {'dst-if': 'DEG2-TTP-TXRX-1'}, + {'dst-if': 'DEG2-TTP-TXRX-761:768'}, res['roadm-connections'][0]['destination']) time.sleep(7) @@ -737,13 +734,13 @@ class TransportPCEFulltesting(unittest.TestCase): time.sleep(2) def test_42_check_xc2_ROADMA(self): - response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP2-TXRX-DEG1-TTP-TXRX-2") + response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP2-TXRX-DEG1-TTP-TXRX-753:760") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2 self.assertDictEqual( dict({ - 'connection-number': 'SRG1-PP2-TXRX-DEG1-TTP-TXRX-2', + 'connection-number': 'SRG1-PP2-TXRX-DEG1-TTP-TXRX-753:760', 'wavelength-number': 2, 'opticalControlMode': 'gainLoss', 'target-output-power': -3.0 @@ -751,10 +748,10 @@ class TransportPCEFulltesting(unittest.TestCase): res['roadm-connections'][0] ) self.assertDictEqual( - {'src-if': 'SRG1-PP2-TXRX-2'}, + {'src-if': 'SRG1-PP2-TXRX-753:760'}, res['roadm-connections'][0]['source']) self.assertDictEqual( - {'dst-if': 'DEG1-TTP-TXRX-2'}, + {'dst-if': 'DEG1-TTP-TXRX-753:760'}, res['roadm-connections'][0]['destination']) time.sleep(2) @@ -784,7 +781,7 @@ class TransportPCEFulltesting(unittest.TestCase): def test_46_get_no_oc_services(self): print("start test") response = test_utils.get_service_list_request("") - self.assertEqual(response.status_code, requests.codes.not_found) + self.assertEqual(response.status_code, requests.codes.conflict) res = response.json() self.assertIn( {