import requests
import urllib.parse
+from dict2xml import dict2xml
+from netconf_client.connect import connect_ssh
+from netconf_client.ncclient import Manager
+
# pylint: disable=import-error
import simulators
else:
TPCE_LOG = KARAF_LOG
+if 'USE_SIMS' not in os.environ:
+ SIMS_TO_USE = 'lightynode'
+ SIMS_TYPE = 'lightynode'
+else:
+ SIMS_TO_USE = os.environ['USE_SIMS']
+ print("Forcing to use SIMS " + SIMS_TO_USE)
+ if SIMS_TO_USE != 'None' or 'SIMS_TYPE' not in os.environ:
+ SIMS_TYPE = SIMS_TO_USE
+ else:
+ SIMS_TYPE = os.environ['SIMS_TYPE']
+ print("Forcing to use SIMS type" + SIMS_TYPE)
+
+
#
# Basic HTTP operations
#
def start_sims(sims_list):
- if 'USE_SIMS' in os.environ:
- sims_to_use = os.environ['USE_SIMS']
- print("Using SIMS " + sims_to_use)
- else:
- sims_to_use = 'lightynode'
- if sims_to_use == 'None':
+ if SIMS_TO_USE == 'None':
return None
- if sims_to_use == 'honeynode':
+ if SIMS_TO_USE == 'honeynode':
start_msg = HONEYNODE_OK_START_MSG
start_method = start_honeynode
else:
def sims_update_cp_port(sim: tuple, circuitpack: str, port: str, payload: dict):
+ if SIMS_TYPE == 'lightynode':
+ return sims_update_cp_port_ntcf(sim, circuitpack, payload)
+ if SIMS_TYPE == 'honeynode':
+ return sims_update_cp_port_rest(sim, circuitpack, port, payload)
+ return False
+
+
+def sims_update_cp_port_rest(sim: tuple, circuitpack: str, port: str, payload: dict):
# pylint: disable=consider-using-f-string
url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/{}/ports/{}".format(
SIMS[sim]['restconf_baseurl'],
urllib.parse.quote(circuitpack, safe=''),
urllib.parse.quote(port, safe=''))
body = {"ports": [payload]}
- print(sim)
- print(url)
response = requests.request("PUT",
url,
data=json.dumps(body),
auth=(ODL_LOGIN, ODL_PWD),
timeout=REQUEST_TIMEOUT)
return response.status_code == requests.codes.ok
+
+
+def sims_update_cp_port_ntcf(sim: tuple, circuitpack: str, payload: dict):
+ body = {"circuit-packs": {"circuit-pack-name": circuitpack, "ports": payload}}
+ xml_body = '<config><org-openroadm-device xmlns="http://org/openroadm/device">'
+ xml_body += dict2xml(body, indent=" ")
+ xml_body += '</org-openroadm-device></config>'
+ with connect_ssh(host='127.0.0.1',
+ port=int(SIMS[sim]['port']),
+ username=NODES_LOGIN,
+ password=NODES_PWD) as session:
+ mgr = Manager(session, timeout=120)
+ reply = mgr.edit_config(xml_body, target="candidate", default_operation="merge")
+ if "None" in str(reply):
+ return True
+ return False
# pylint: disable=no-member
# pylint: disable=too-many-public-methods
import os
-import json
import unittest
import time
import requests
self.assertEqual(response['output']['service']['lifecycle-state'], 'INSTALLED')
def test_14_change_status_line_port_xpdrc(self):
- url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1"
- body = {"ports": [{
+ self.assertTrue(test_utils.sims_update_cp_port(('xpdrc', self.NODE_VERSION_221), '1/0/1-PLUG-NET', '1',
+ {
"port-name": "1",
"port-type": "CFP2",
"administrative-state": "outOfService",
- "port-qual": "xpdr-network"}]}
- response = requests.request("PUT", url.format("http://127.0.0.1:8144/restconf"),
- data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
- timeout=test_utils.REQUEST_TIMEOUT)
- self.assertEqual(response.status_code, requests.codes.ok)
+ "port-qual": "xpdr-network"
+ }))
time.sleep(2)
def test_15_check_update_portmapping(self):
time.sleep(1)
def test_21_restore_status_line_port_xpdrc(self):
- url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1"
- body = {"ports": [{
+ self.assertTrue(test_utils.sims_update_cp_port(('xpdrc', self.NODE_VERSION_221), '1/0/1-PLUG-NET', '1',
+ {
"port-name": "1",
"port-type": "CFP2",
"administrative-state": "inService",
- "port-qual": "xpdr-network"}]}
- response = requests.request("PUT", url.format("http://127.0.0.1:8144/restconf"),
- data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
- timeout=test_utils.REQUEST_TIMEOUT)
- self.assertEqual(response.status_code, requests.codes.ok)
+ "port-qual": "xpdr-network"
+ }))
time.sleep(2)
def test_22_check_update_portmapping_ok(self):