Change way to start simulators

[transportpce.git] / tests / transportpce_tests / 2.2.1 / test_otn_topology.py

diff --git a/tests/transportpce_tests/2.2.1/test_otn_topology.py b/tests/transportpce_tests/2.2.1/test_otn_topology.py
index f80c12d514d778a534dd6316985421621a861af9..84e3652c764aca4747e68fa576e8e38168917ecd 100644 (file)
--- a/tests/transportpce_tests/2.2.1/test_otn_topology.py
+++ b/tests/transportpce_tests/2.2.1/test_otn_topology.py
@@ -16,17 +16,20 @@ import unittest
 import time
 import logging
 import requests
-from common import test_utils
+import sys
+sys.path.append('transportpce_tests/common/')
+import test_utils
 
 
 class TransportPCEtesting(unittest.TestCase):
 
     processes = None
+    NODE_VERSION = '2.2.1'
 
     @classmethod
     def setUpClass(cls):
         cls.processes = test_utils.start_tpce()
-        cls.processes = test_utils.start_sims(['spdra'])
+        cls.processes = test_utils.start_sims([('spdra', cls.NODE_VERSION)])
 
     @classmethod
     def tearDownClass(cls):
@@ -39,7 +42,7 @@ class TransportPCEtesting(unittest.TestCase):
         time.sleep(5)
 
     def test_01_connect_SPDR_SA1(self):
-        response = test_utils.mount_device("SPDR-SA1", 'spdra')
+        response = test_utils.mount_device("SPDR-SA1", ('spdra', self.NODE_VERSION))
         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
         time.sleep(10)
 
@@ -86,8 +89,8 @@ class TransportPCEtesting(unittest.TestCase):
         self.assertIn('node', res['network'][0])
         if 'node' in res['network'][0]:
             nbNode = len(res['network'][0]['node'])
-            self.assertEqual(nbNode, 2)
-            listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2']
+            self.assertEqual(nbNode, 3)
+            listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2', 'SPDR-SA1-XPDR3']
             for i in range(0, nbNode):
                 nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
                 nodeId = res['network'][0]['node'][i]['node-id']
@@ -110,7 +113,7 @@ class TransportPCEtesting(unittest.TestCase):
                     elif tpType == 'XPONDER-NETWORK':
                         network += 1
                 self.assertTrue(client == 0)
-                if nodeId == 'SPDR-SA1-XPDR1':
+                if nodeId == 'SPDR-SA1-XPDR1' or nodeId == 'SPDR-SA1-XPDR3':
                     self.assertTrue(network == 1)
                 else:
                     # elif nodeId == 'SPDR-SA1-XPDR2':
@@ -130,13 +133,14 @@ class TransportPCEtesting(unittest.TestCase):
         res = response.json()
         self.assertEqual(response.status_code, requests.codes.ok)
         nbNode = len(res['network'][0]['node'])
-        self.assertEqual(nbNode, 2)
-        listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2']
+        self.assertEqual(nbNode, 3)
+        listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2', 'SPDR-SA1-XPDR3']
         CHECK_LIST = {
             'SPDR-SA1-XPDR1': {
                 'node-type': 'MUXPDR',
                 'xpdr-number': 1,
-                'port-types': 'org-openroadm-port-types:if-10GE-ODU2e',
+                'port-types': ['org-openroadm-port-types:if-10GE-ODU2', 'org-openroadm-port-types:if-10GE-ODU2e',
+                               'org-openroadm-port-types:if-10GE'],
                 'otn-common-types': 'org-openroadm-otn-common-types:ODU2e',
                 'network_nb': 1,
                 'nbl_nb': 4,
@@ -178,14 +182,16 @@ class TransportPCEtesting(unittest.TestCase):
                     tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
                     if tpType == 'XPONDER-CLIENT':
                         client += 1
-                        self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+                        print("tpId = {}".format(tpId))
+                        print("tp= {}".format(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]))
+                        nbIfCapType = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+                                 ['org-openroadm-otn-network-topology:tp-supported-interfaces']
+                                 ['supported-interface-capability'][0])
+                        for k in range(0, nbIfCapType):
+                            self.assertIn((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
                                              ['org-openroadm-otn-network-topology:tp-supported-interfaces']
                                              ['supported-interface-capability'][0]['if-cap-type']),
-                                         CHECK_LIST[nodeId]['port-types'])
-                        self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
-                                             ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']
-                                             ['rate']),
-                                         CHECK_LIST[nodeId]['otn-common-types'])
+                                           CHECK_LIST[nodeId]['port-types'])
                     elif tpType == 'XPONDER-NETWORK':
                         network += 1
                         self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
@@ -220,7 +226,8 @@ class TransportPCEtesting(unittest.TestCase):
                         for tp in CHECK_LIST[nodeId]['tp-unchecklist']:
                             self.assertNotIn(tp, nbl['tp-list'])
             else:
-                self.assertFalse(True)
+                self.assertEqual('SPDR-SA1-XPDR3', nodeId)
+                listNode.remove(nodeId)
         self.assertEqual(len(listNode), 0)
 
     def test_08_disconnect_SPDR_SA1(self):