from netconf_client.connect import connect_ssh
from netconf_client.ncclient import Manager
+
# pylint: disable=import-error
import simulators
if "None" in str(reply):
return True
return False
+
+
+def sims_update_pm_interact(sim: tuple, payload: dict):
+ if SIMS_TYPE == 'lightynode':
+ return sims_update_pm_interact_ntcf(sim, payload)
+ if SIMS_TYPE == 'honeynode':
+ return sims_update_pm_interact_rest(sim, payload)
+ return False
+
+
+def sims_update_pm_interact_rest(sim: tuple, payload: dict):
+ # pylint: disable=consider-using-f-string
+ url = "{}/operations/pm-handling:pm-interact".format(SIMS[sim]['restconf_baseurl'])
+ body = {"input": payload}
+ response = requests.request("POST",
+ 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_pm_interact_ntcf(sim: tuple, payload: dict):
+ # pylint: disable=line-too-long
+ xml_body = '<pm-interact xmlns="http://honeynode-simulator/pm-handling">'
+ xml_body += dict2xml(payload, indent=" ")
+ xml_body += '</pm-interact>'
+ new_xml = xml_body.replace("<pm-resource-instance>/org-openroadm-device:org-openroadm-device/org-openroadm-device:interface[org-openroadm-device:name='OTS-DEG2-TTP-TXRX']</pm-resource-instance>",
+ "<pm-resource-instance xmlns:a=\"http://org/openroadm/device\">/a:org-openroadm-device/a:interface[a:name='OTS-DEG2-TTP-TXRX']</pm-resource-instance>")
+ 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.dispatch(new_xml)
+ if "netconf_client.ncclient.RPCReply" in str(reply):
+ return True
+ return False
# pylint: disable=too-many-public-methods
# pylint: disable=too-many-lines
-import json
import unittest
import time
import requests
}
]
+ pm_handling_degrade_body = {
+ "rpc-action": "set",
+ "pm-to-be-set-or-created": {
+ "current-pm-entry": [
+ {
+ "pm-resource-instance": "/org-openroadm-device:org-openroadm-device/org-openroadm-device"
+ ":interface[org-openroadm-device:name='OTS-DEG2-TTP-TXRX']",
+ "pm-resource-type": "interface",
+ "pm-resource-type-extension": "",
+ "current-pm": [
+ {
+ "type": "opticalPowerInput",
+ "extension": "",
+ "location": "nearEnd",
+ "direction": "rx",
+ "measurement": [
+ {
+ "granularity": "15min",
+ "pmParameterValue": -30,
+ "pmParameterUnit": "dBm",
+ "validity": "complete"
+ },
+ {
+ "granularity": "24Hour",
+ "pmParameterValue": -21.3,
+ "pmParameterUnit": "dBm",
+ "validity": "complete"
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ }
+ }
+
+ pm_handling_recover_body = {
+ "rpc-action": "clear",
+ "pm-to-get-clear-or-delete": {
+ "current-pm-entry": [
+ {
+ "pm-resource-instance": "/org-openroadm-device:org-openroadm-device/org-openroadm-device"
+ ":interface[org-openroadm-device:name='OTS-DEG2-TTP-TXRX']",
+ "pm-resource-type": "interface",
+ "pm-resource-type-extension": "",
+ "current-pm": [
+ {
+ "type": "opticalPowerInput",
+ "extension": "",
+ "location": "nearEnd",
+ "direction": "rx"
+ }
+ ]
+ }
+ ]
+ }
+ }
+
@classmethod
def setUpClass(cls):
cls.processes = test_utils.start_tpce()
# Degrade ROADM-A1-ROADM-C1 link
def test_14_set_pm_ROADMA_OTS_DEG2_TTP_TXRX_OpticalPowerInput(self):
- url = "{}/operations/pm-handling:pm-interact"
- body = {
- "input": {
- "rpc-action": "set",
- "pm-to-be-set-or-created": {
- "current-pm-entry": [
- {
- "pm-resource-instance": "/org-openroadm-device:org-openroadm-device/org-openroadm-device"
- ":interface[org-openroadm-device:name='OTS-DEG2-TTP-TXRX']",
- "pm-resource-type": "interface",
- "pm-resource-type-extension": "",
- "current-pm": [
- {
- "type": "opticalPowerInput",
- "extension": "",
- "location": "nearEnd",
- "direction": "rx",
- "measurement": [
- {
- "granularity": "15min",
- "pmParameterValue": -30,
- "pmParameterUnit": "dBm",
- "validity": "complete"
- },
- {
- "granularity": "24Hour",
- "pmParameterValue": -21.3,
- "pmParameterUnit": "dBm",
- "validity": "complete"
- }
- ]
- }
- ]
- }
- ]
- }
- }
- }
- response = requests.request("POST", url.format("http://127.0.0.1:8141/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)
- self.assertEqual(response.json()['output']['status-message'], "The PMs has been successfully set !")
+ self.assertTrue(test_utils.sims_update_pm_interact(('roadma', self.NODE_VERSION_221),
+ self.pm_handling_degrade_body))
time.sleep(2)
def test_15_get_eth_service1(self):
# Restore ROADM-A1-ROADM-C1 link
def test_17_clear_pm_ROADMA_OTS_DEG2_TTP_TXRX_OpticalPowerInput(self):
- url = "{}/operations/pm-handling:pm-interact"
- body = {
- "input": {
- "rpc-action": "clear",
- "pm-to-get-clear-or-delete": {
- "current-pm-entry": [
- {
- "pm-resource-instance": "/org-openroadm-device:org-openroadm-device/org-openroadm-device"
- ":interface[org-openroadm-device:name='OTS-DEG2-TTP-TXRX']",
- "pm-resource-type": "interface",
- "pm-resource-type-extension": "",
- "current-pm": [
- {
- "type": "opticalPowerInput",
- "extension": "",
- "location": "nearEnd",
- "direction": "rx"
- }
- ]
- }
- ]
- }
- }
- }
- response = requests.request("POST", url.format("http://127.0.0.1:8141/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)
- self.assertEqual(response.json()['output']['status-message'], "The PMs has been successfully released !")
+ self.assertTrue(test_utils.sims_update_pm_interact(('roadma', self.NODE_VERSION_221),
+ self.pm_handling_recover_body))
time.sleep(2)
def test_18_get_eth_service1(self):
# Degrade ROADM-A1-ROADM-C1 link
def test_36_set_pm_ROADMA_OTS_DEG2_TTP_TXRX_OpticalPowerInput(self):
- url = "{}/operations/pm-handling:pm-interact"
- body = {
- "input": {
- "rpc-action": "set",
- "pm-to-be-set-or-created": {
- "current-pm-entry": [
- {
- "pm-resource-instance": "/org-openroadm-device:org-openroadm-device/org-openroadm-device"
- ":interface[org-openroadm-device:name='OTS-DEG2-TTP-TXRX']",
- "pm-resource-type": "interface",
- "pm-resource-type-extension": "",
- "current-pm": [
- {
- "type": "opticalPowerInput",
- "extension": "",
- "location": "nearEnd",
- "direction": "rx",
- "measurement": [
- {
- "granularity": "15min",
- "pmParameterValue": -30,
- "pmParameterUnit": "dBm",
- "validity": "complete"
- },
- {
- "granularity": "24Hour",
- "pmParameterValue": -21.3,
- "pmParameterUnit": "dBm",
- "validity": "complete"
- }
- ]
- }
- ]
- }
- ]
- }
- }
- }
- response = requests.request("POST", url.format("http://127.0.0.1:8141/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)
- self.assertEqual(response.json()['output']['status-message'], "The PMs has been successfully set !")
+ self.test_14_set_pm_ROADMA_OTS_DEG2_TTP_TXRX_OpticalPowerInput()
time.sleep(self.WAITING * 2)
def test_37_get_eth_service1(self):
# Restore ROADM-A1-ROADM-C1 link
def test_41_clear_pm_ROADMA_OTS_DEG2_TTP_TXRX_OpticalPowerInput(self):
- url = "{}/operations/pm-handling:pm-interact"
- body = {
- "input": {
- "rpc-action": "clear",
- "pm-to-get-clear-or-delete": {
- "current-pm-entry": [
- {
- "pm-resource-instance": "/org-openroadm-device:org-openroadm-device/org-openroadm-device"
- ":interface[org-openroadm-device:name='OTS-DEG2-TTP-TXRX']",
- "pm-resource-type": "interface",
- "pm-resource-type-extension": "",
- "current-pm": [
- {
- "type": "opticalPowerInput",
- "extension": "",
- "location": "nearEnd",
- "direction": "rx"
- }
- ]
- }
- ]
- }
- }
- }
- response = requests.request("POST", url.format("http://127.0.0.1:8141/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)
- self.assertEqual(response.json()['output']['status-message'], "The PMs has been successfully released !")
- time.sleep(2)
+ self.test_17_clear_pm_ROADMA_OTS_DEG2_TTP_TXRX_OpticalPowerInput()
def test_42_get_eth_service1(self):
self.test_13_get_eth_service1()
commands =
./build.sh
+[testenv:sims]
+passenv = USE_SIMS
+allowlist_externals = install_lightynode.sh
+depends = buildcontroller
+commands =
+ ./install_lightynode.sh {posargs:}
+
[testenv:sims121]
passenv = USE_SIMS
allowlist_externals = install_sims.sh
[testenv:tests71]
depends = buildcontroller,build_karaf_tests71,sims71,tests_tapi
-allowlist_externals = launch_tests.sh install_honeynode.sh
+allowlist_externals = launch_tests.sh
passenv = LAUNCHER
USE_LIGHTY
+ USE_SIMS
NO_ODL_STARTUP
USE_ODL_RESTCONF_VERSION
setenv =
# USE_LIGHTY=True
# USE_ODL_RESTCONF_VERSION=draft-bierman02
- USE_SIMS=honeynode
USE_ODL_ALT_KARAF_ENV=./karaf71.env
USE_ODL_ALT_KARAF_INSTALL_DIR=karaf71
commands =
- ./install_honeynode.sh 7.1
./launch_tests.sh 7.1 {posargs:}
[testenv:build_karaf_tests_hybrid]
[testenv:tests_hybrid]
depends = buildcontroller,build_karaf_tests_hybrid,sims121,sims221,sims71,tests121,tests221,tests71
#the last dependency is to avoid temporarily concurrent ressources problem in parallel mode
-allowlist_externals = launch_tests.sh install_honeynode.sh
+allowlist_externals = launch_tests.sh
passenv = LAUNCHER
USE_LIGHTY
USE_SIMS
NO_ODL_STARTUP
USE_ODL_RESTCONF_VERSION
setenv =
- USE_SIMS=honeynode
# USE_LIGHTY=True
# USE_ODL_RESTCONF_VERSION=draft-bierman02
USE_ODL_ALT_KARAF_ENV=./karaf121.env
USE_ODL_ALT_KARAF_INSTALL_DIR=karaf_hybrid
commands =
# nosetests --with-xunit transportpce_tests/hybrid/test01_device_change_notifications.py
- ./install_honeynode.sh
./launch_tests.sh hybrid {posargs:}
[testenv:tests_tapi]
[testenv:with_docker]
depends = buildcontroller,sims221
-allowlist_externals = launch_tests.sh dockercmd.sh install_honeynode.sh
+allowlist_externals = launch_tests.sh dockercmd.sh
passenv = LAUNCHER
USE_LIGHTY
+ USE_SIMS
NO_ODL_STARTUP
USE_ODL_RESTCONF_VERSION
# OLM_TIMER1
# OLM_TIMER2
setenv =
- USE_SIMS=honeynode
OLM_TIMER1=3000
OLM_TIMER2=2000
INSTALL_TAPI=True
# USE_ODL_RESTCONF_VERSION=draft-bierman02
commands =
- ./install_honeynode.sh 2.2.1
./dockercmd.sh run -d -p 2181:2181 -p 9092:9092 --env ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 --name tpce_kafka1 teivah/kafka:2.0.0
./launch_tests.sh with_docker {posargs:}
./dockercmd.sh container rm -f tpce_kafka1