@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
@classmethod
def setUpClass(cls):
+ # pylint: disable=bare-except
try:
sample_files_parsed = False
TOPO_CLLINET_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
self.assertIn('Request processed', res["output"]["result"])
time.sleep(10)
- """to test case where SRG where the xpdr is connected to has no optical range data"""
+ #"""to test case where SRG where the xpdr is connected to has no optical range data"""
def test_31_connect_xprdA_to_roadmA(self):
response = test_utils.connect_xpdr_to_rdm_request("XPDRA01", "1", "2",
@classmethod
def setUpClass(cls):
+ # pylint: disable=bare-except
try:
sample_files_parsed = False
TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
# pylint: disable=no-member
# pylint: disable=too-many-public-methods
+# a pylint false positive due to unittest
+# pylint: disable=no-self-use
import time
import unittest
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
# Verify the termination points of the ROADMA
- def test_02_compareOpenroadmTopologyPortMapping_rdm(self):
+ def test_02_compare_Openroadm_topology_portmapping_rdm(self):
responseTopo = test_utils.get_ordm_topo_request("")
resTopo = responseTopo.json()
nbNode = len(resTopo['network'][0]['node'])
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
# #Verify the termination points related to XPDR
- def test_05_compareOpenroadmTopologyPortMapping_xpdr(self):
- self.test_02_compareOpenroadmTopologyPortMapping_rdm()
+ def test_05_compare_Openroadm_topology_portmapping_xpdr(self):
+ self.test_02_compare_Openroadm_topology_portmapping_rdm()
# Disconnect the XPDRA
def test_06_disconnect_device(self):
# pylint: disable=no-member
# pylint: disable=too-many-public-methods
+# pylint: disable=too-many-lines
import time
import unittest
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:model'], '2')
def test_04_getLinks_OpenroadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(len(dropLink), 0)
def test_05_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeA')
def test_08_getOpenRoadmNetwork(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertFalse(True)
def test_09_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
self.assertEqual(response.status_code, requests.codes.ok)
def test_12_getLinks_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(len(listNode), 0)
def test_17_getOpenRoadmNetwork(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(len(listNode), 0)
def test_18_getROADMLinkOpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
length = (res['network'][0]['ietf-network-topology:link'][i]
['org-openroadm-network-topology:OMS-attributes']['span']['link-concatenation'][0]
['SRLG-length'])
- if (spanLoss != None) & (length != None):
+ if (spanLoss is not None) & (length is not None):
find = True
self.assertTrue(find)
R2RLink.remove(link_id)
self.assertEqual(len(R2RLink), 0)
def test_20_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
'org-openroadm-network-topology:OMS-attributes']['span']["engineered-spanloss"]
length = res['network'][0]['ietf-network-topology:link'][i][
'org-openroadm-network-topology:OMS-attributes']['span']['link-concatenation'][0]['SRLG-length']
- if (spanLoss != None) & (length != None):
+ if (spanLoss is not None) & (length is not None):
find = True
self.assertTrue(find)
R2RLink.remove(link_id)
self.assertEqual(response.status_code, requests.codes.ok)
def test_32_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
self.assertNotEqual(res['network'][0]['node'][i]['node-id'], 'XPDRA01')
def test_38_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
self.assertIn('Request processed', res["output"]["result"])
time.sleep(10)
- """to test case where SRG where the xpdr is connected to has no optical range data"""
+ #"""to test case where SRG where the xpdr is connected to has no optical range data"""
def test_31_connect_xprdA_to_roadmA(self):
response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "2",
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
self.assertEqual(nb_links, 4)
for link in res['network'][0]['ietf-network-topology:link']:
linkId = link['link-id']
- if (linkId == 'OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1' or
- linkId == 'OTU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1'):
+ if (linkId in ('OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1',
+ 'OTU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1')):
self.assertEqual(link['org-openroadm-otn-network-topology:available-bandwidth'], 0)
self.assertEqual(link['org-openroadm-otn-network-topology:used-bandwidth'], 100000)
- elif (linkId == 'ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1' or
- linkId == 'ODU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1'):
+ elif (linkId in ('ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1',
+ 'ODU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1')):
self.assertEqual(link['org-openroadm-otn-network-topology:available-bandwidth'], 100000)
self.assertEqual(link['org-openroadm-otn-network-topology:used-bandwidth'], 0)
self.assertEqual(link['transportpce-topology:otn-link-type'], 'ODTU4')
self.assertEqual(nb_links, 4)
for link in res['network'][0]['ietf-network-topology:link']:
linkId = link['link-id']
- if (linkId == 'ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1' or
- linkId == 'ODU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1'):
+ if (linkId in ('ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1',
+ 'ODU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1')):
self.assertEqual(link['org-openroadm-otn-network-topology:available-bandwidth'], 90000)
self.assertEqual(link['org-openroadm-otn-network-topology:used-bandwidth'], 10000)
if tp['tp-id'] == 'XPDR1-NETWORK1':
xpdrTpPortConAt = tp['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']
self.assertEqual(len(xpdrTpPortConAt['ts-pool']), 72)
- tsPoolList = [i for i in range(1, 9)]
+ tsPoolList = list(range(1, 9))
self.assertNotIn(tsPoolList, xpdrTpPortConAt['ts-pool'])
self.assertEqual(len(xpdrTpPortConAt['odtu-tpn-pool'][0]['tpn-pool']), 79)
self.assertNotIn(1, xpdrTpPortConAt['odtu-tpn-pool'][0]['tpn-pool'])
self.assertEqual(nb_links, 4)
for link in res['network'][0]['ietf-network-topology:link']:
linkId = link['link-id']
- if (linkId == 'ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1' or
- linkId == 'ODU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1'):
+ if (linkId in ('ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1',
+ 'ODU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1')):
self.assertEqual(link['org-openroadm-otn-network-topology:available-bandwidth'], 100000)
self.assertEqual(link['org-openroadm-otn-network-topology:used-bandwidth'], 0)
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
self.assertNotIn('ietf-network-topology:link', res['network'][0])
def test_05_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
self.assertNotIn('ietf-network-topology:link', res['network'][0])
def test_07_getNodes_OtnTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_otn_topo_request()
res = response.json()
self.assertEqual(response.status_code, requests.codes.ok)
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
# pylint: disable=no-member
# pylint: disable=too-many-public-methods
+# some pylint false positives specific to tapi test
+# pylint: disable=unsubscriptable-object
+# pylint: disable=unsupported-assignment-operation
import os
import sys
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
# pylint: disable=no-member
# pylint: disable=too-many-public-methods
+# pylint: disable=too-many-lines
import unittest
import time
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:model'], 'model2')
def test_04_getLinks_OpenroadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(len(dropLink), 0)
def test_05_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeA')
def test_08_getOpenRoadmNetwork(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertFalse(True)
def test_09_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
self.assertEqual(response.status_code, requests.codes.ok)
def test_12_getLinks_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(len(listNode), 0)
def test_17_getOpenRoadmNetwork(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(len(listNode), 0)
def test_18_getROADMLinkOpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
'org-openroadm-network-topology:OMS-attributes']['span']["engineered-spanloss"]
length = res['network'][0]['ietf-network-topology:link'][i][
'org-openroadm-network-topology:OMS-attributes']['span']['link-concatenation'][0]['SRLG-length']
- if (spanLoss != None) & (length != None):
+ if (spanLoss is not None) & (length is not None):
find = True
self.assertTrue(find)
R2RLink.remove(link_id)
self.assertEqual(len(R2RLink), 0)
def test_20_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
'org-openroadm-network-topology:OMS-attributes']['span']["engineered-spanloss"]
length = res['network'][0]['ietf-network-topology:link'][i][
'org-openroadm-network-topology:OMS-attributes']['span']['link-concatenation'][0]['SRLG-length']
- if (spanLoss != None) & (length != None):
+ if (spanLoss is not None) & (length is not None):
find = True
self.assertTrue(find)
R2RLink.remove(link_id)
# 'ROADMC-DEG2-DEG2-CTP-TXRXtoROADMC-DEG1-DEG1-CTP-TXRX')
def test_32_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
self.assertNotEqual(res['network'][0]['node'][i]['node-id'], 'XPDR-A1')
def test_38_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
"..", "..", "..", "karaf", "target", "assembly", "bin", "client")
return subprocess.run([executable],
input='feature:install ' + feature_name + '\n feature:list | grep tapi \n logout \n',
- universal_newlines=True)
+ universal_newlines=True, check=False)
def get_request(url):
data=json.dumps(data),
headers=TYPE_APPLICATION_JSON,
auth=(ODL_LOGIN, ODL_PWD))
- else:
- return requests.request(
- "POST", url.format(RESTCONF_BASE_URL),
- headers=TYPE_APPLICATION_JSON,
- auth=(ODL_LOGIN, ODL_PWD))
+
+ return requests.request(
+ "POST", url.format(RESTCONF_BASE_URL),
+ headers=TYPE_APPLICATION_JSON,
+ auth=(ODL_LOGIN, ODL_PWD))
def post_xmlrequest(url, data):
data=data,
headers=TYPE_APPLICATION_XML,
auth=(ODL_LOGIN, ODL_PWD))
+ return None
def put_request(url, data):
def service_path_request(operation: str, servicename: str, wavenumber: str, nodes):
attr = {"renderer:input": {
- "renderer:service-name": servicename,
- "renderer:wave-number": wavenumber,
- "renderer:modulation-format": "qpsk",
- "renderer:operation": operation,
- "renderer:nodes": nodes}}
+ "renderer:service-name": servicename,
+ "renderer:wave-number": wavenumber,
+ "renderer:modulation-format": "qpsk",
+ "renderer:operation": operation,
+ "renderer:nodes": nodes}}
return post_request(URL_SERVICE_PATH, attr)
def create_ots_oms_request(nodeid: str, lcp: str):
attr = {"input": {
- "node-id": nodeid,
- "logical-connection-point": lcp}}
+ "node-id": nodeid,
+ "logical-connection-point": lcp}}
return post_request(URL_CREATE_OTS_OMS, attr)
return subprocess.Popen(
[HONEYNODE_EXECUTABLE, node_port, os.path.join(SAMPLES_DIRECTORY, node_config_file_name)],
stdout=outfile, stderr=outfile)
+ return None
def wait_until_log_contains(log_file, regexp, time_to_wait=20):
+ # pylint: disable=lost-exception
stringfound = False
filefound = False
line = None
signal.alarm(self.seconds)
def __exit__(self, type, value, traceback):
+ # pylint: disable=W0622
signal.alarm(0)