+
+#
+# simulators datastore operations
+#
+
+
+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]}
+ response = requests.request("PUT",
+ url,
+ data=json.dumps(body),
+ headers=TYPE_APPLICATION_JSON,
+ 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