# pylint: disable=unsupported-assignment-operation
import os
+# pylint: disable=wrong-import-order
import sys
import time
import unittest
import requests
sys.path.append('transportpce_tests/common/')
-import test_utils
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
+import test_utils # nopep8
CREATED_SUCCESSFULLY = 'Result message should contain Xponder Roadm Link created successfully'
def setUp(self): # instruction executed before each test method
if self.init_failed:
self.fail('Feature installation failed')
+ # pylint: disable=consider-using-f-string
print("execution of {}".format(self.id().split(".")[-1]))
def test_01_get_tapi_topology_T100G(self):
response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
res = response.json()
- self.assertNotIn("node", res["output"]["topology"], 'Topology should contain no node')
+ self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'Topology should contain 1 node')
self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
def test_05_disconnect_roadmb(self):
time.sleep(10)
def test_07_check_tapi_topos(self):
- self.test_04_check_tapi_topos()
+ url = "{}/operations/tapi-topology:get-topology-details"
+ data = {
+ "tapi-topology:input": {
+ "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
+ }
+ }
+ response = test_utils.post_request(url, data)
+ self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
+ res = response.json()
+ self.assertNotIn("node", res["output"]["topology"], 'Topology should contain no node')
+ self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
def test_08_connect_rdma(self):
response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
res = response.json()
self.assertEqual(1, len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]),
'Node should contain 1 owned-node-edge-points')
- self.assertEqual("XPDR1-CLIENT1",
+ self.assertEqual("XPDR-A1-XPDR1+DSR+XPDR1-CLIENT1",
res["output"]["topology"]["node"][0]["owned-node-edge-point"][0]["name"][0]["value"],
- 'name of owned-node-edge-points should be XPDR1-CLIENT1')
+ 'name of owned-node-edge-points should be XPDR-A1-XPDR1+DSR+XPDR1-CLIENT1')
def test_14_check_tapi_topology_T0(self):
url = "{}/operations/tapi-topology:get-topology-details"
res = response.json()
self.assertEqual(2, len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]),
'Node should contain 2 owned-node-edge-points')
- self.assertEqual("XPDR1-CLIENT1",
+ self.assertEqual("XPDR-C1-XPDR1+DSR+XPDR1-CLIENT1",
res["output"]["topology"]["node"][0]["owned-node-edge-point"][0]["name"][0]["value"],
- 'name of owned-node-edge-points should be XPDR1-CLIENT1')
- self.assertEqual("XPDR1-CLIENT1",
+ 'name of owned-node-edge-points should be XPDR-C1-XPDR1+DSR+XPDR1-CLIENT1')
+ self.assertEqual("XPDR-A1-XPDR1+DSR+XPDR1-CLIENT1",
res["output"]["topology"]["node"][0]["owned-node-edge-point"][1]["name"][0]["value"],
- 'name of owned-node-edge-points should be XPDR1-CLIENT1')
+ 'name of owned-node-edge-points should be XPDR-A1-XPDR1+DSR+XPDR1-CLIENT1')
def test_19_check_tapi_topology_T0(self):
url = "{}/operations/tapi-topology:get-topology-details"
res = response.json()
links = res['network'][0]['ietf-network-topology:link']
for link in links:
- if (link["org-openroadm-common-network:link-type"] == "XPONDER-OUTPUT" or
- link["org-openroadm-common-network:link-type"] == "XPONDER-INPUT"):
+ if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT'):
link_name = link["link-id"]
response = test_utils.delete_request(url+link_name)
self.assertEqual(response.status_code, requests.codes.ok)
def count_object_with_double_key(list_dicts, key1, key2, value):
nb = 0
- for dict in list_dicts:
- if dict[key1][0][key2] == value:
+ for dictio in list_dicts:
+ if dictio[key1][0][key2] == value:
nb += 1
return nb