# pylint: disable=too-many-public-methods
import os
+import json
import sys
import unittest
import time
import requests
-from common import test_utils
+sys.path.append('transportpce_tests/common/')
+import test_utils
class TransportNbiNotificationstesting(unittest.TestCase):
}
WAITING = 20 # nominal value is 300
+ NODE_VERSION = '2.2.1'
@classmethod
def setUpClass(cls):
# TODO: for lighty manage the activation of NBI notification feature
+ cls.init_failed = False
cls.processes = test_utils.start_tpce()
# NBI notification feature is not installed by default in Karaf
if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
print("NBI notification installation feature failed...")
test_utils.shutdown_process(cls.processes[0])
sys.exit(2)
- cls.processes = test_utils.start_sims(['xpdra', 'roadma', 'roadmc', 'xpdrc'])
+ cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION),
+ ('roadma', cls.NODE_VERSION),
+ ('roadmc', cls.NODE_VERSION),
+ ('xpdrc', cls.NODE_VERSION)])
@classmethod
def tearDownClass(cls):
print("execution of {}".format(self.id().split(".")[-1]))
def test_01_connect_xpdrA(self):
- response = test_utils.mount_device("XPDR-A1", 'xpdra')
+ response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION))
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_02_connect_xpdrC(self):
- response = test_utils.mount_device("XPDR-C1", 'xpdrc')
+ response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION))
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_03_connect_rdmA(self):
- response = test_utils.mount_device("ROADM-A1", 'roadma')
+ response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_04_connect_rdmC(self):
- response = test_utils.mount_device("ROADM-C1", 'roadmc')
+ response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION))
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_05_connect_xprdA_N1_to_roadmA_PP1(self):
response = test_utils.get_notifications_service_request(data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
- self.assertEqual(res['output']['notification-service'][-2]['service-name'], 'service1')
- self.assertEqual(res['output']['notification-service'][-2]['connection-type'], 'service')
- self.assertEqual(res['output']['notification-service'][-2]['message'], 'ServiceCreate request received ...')
self.assertEqual(res['output']['notification-service'][-1]['service-name'], 'service1')
+ self.assertEqual(res['output']['notification-service'][-1]['connection-type'], 'service')
self.assertEqual(res['output']['notification-service'][-1]['message'], 'ServiceCreate request failed ...')
self.assertEqual(res['output']['notification-service'][-1]['response-failed'],
'PCE path computation failed !')
response = test_utils.get_notifications_service_request(data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
- self.assertEqual(res['output']['notification-service'][-3]['service-name'], 'service1')
- self.assertEqual(res['output']['notification-service'][-3]['connection-type'], 'service')
- self.assertEqual(res['output']['notification-service'][-3]['message'], 'ServiceCreate request received ...')
- self.assertEqual(res['output']['notification-service'][-2]['service-name'], 'service1')
- self.assertEqual(res['output']['notification-service'][-2]['message'], 'PCE calculation done OK !')
self.assertEqual(res['output']['notification-service'][-1]['service-name'], 'service1')
+ self.assertEqual(res['output']['notification-service'][-1]['connection-type'], 'service')
self.assertEqual(res['output']['notification-service'][-1]['message'], 'Service implemented !')
time.sleep(2)
- def test_17_delete_eth_service1(self):
+ def test_17_get_notifications_alarm_service1(self):
+ data = {
+ "input": {
+ "connection-type": "service",
+ "id-consumer": "consumer",
+ "group-id": "transportpceTest"
+ }
+ }
+ response = test_utils.get_notifications_alarm_service_request(data)
+ self.assertEqual(response.status_code, requests.codes.ok)
+ res = response.json()
+ self.assertEqual(res['output']['notification-alarm-service'][-1]['service-name'], 'service1')
+ self.assertEqual(res['output']['notification-alarm-service'][-1]['connection-type'], 'service')
+ self.assertEqual(res['output']['notification-alarm-service'][-1]['operational-state'], 'inService')
+ self.assertEqual(res['output']['notification-alarm-service'][-1]['message'], 'The service is now inService')
+ time.sleep(2)
+
+ def test_18_change_status_port_roadma_srg(self):
+ url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
+ body = {"ports": [{
+ "port-name": "C1",
+ "logical-connection-point": "SRG1-PP1",
+ "port-type": "client",
+ "circuit-id": "SRG1",
+ "administrative-state": "outOfService",
+ "port-qual": "roadm-external"}]}
+ response = requests.request("PUT", 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))
+ self.assertEqual(response.status_code, requests.codes.ok)
+ time.sleep(2)
+
+ def test_19_get_notifications_alarm_service1(self):
+ data = {
+ "input": {
+ "connection-type": "service",
+ "id-consumer": "consumer",
+ "group-id": "transportpceTest"
+ }
+ }
+ response = test_utils.get_notifications_alarm_service_request(data)
+ self.assertEqual(response.status_code, requests.codes.ok)
+ res = response.json()
+ self.assertEqual(res['output']['notification-alarm-service'][-1]['service-name'], 'service1')
+ self.assertEqual(res['output']['notification-alarm-service'][-1]['connection-type'], 'service')
+ self.assertEqual(res['output']['notification-alarm-service'][-1]['operational-state'], 'outOfService')
+ self.assertEqual(res['output']['notification-alarm-service'][-1]['message'], 'The service is now outOfService')
+ time.sleep(2)
+
+ def test_20_restore_status_port_roadma_srg(self):
+ url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
+ body = {"ports": [{
+ "port-name": "C1",
+ "logical-connection-point": "SRG1-PP1",
+ "port-type": "client",
+ "circuit-id": "SRG1",
+ "administrative-state": "inService",
+ "port-qual": "roadm-external"}]}
+ response = requests.request("PUT", 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))
+ self.assertEqual(response.status_code, requests.codes.ok)
+ time.sleep(2)
+
+ def test_21_get_notifications_alarm_service1(self):
+ self.test_17_get_notifications_alarm_service1()
+
+ def test_22_delete_eth_service1(self):
response = test_utils.service_delete_request("service1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
res['output']['configuration-response-common']['response-message'])
time.sleep(20)
- def test_18_get_notifications_service1(self):
+ def test_23_get_notifications_service1(self):
data = {
"input": {
"connection-type": "service",
response = test_utils.get_notifications_service_request(data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
- self.assertEqual(res['output']['notification-service'][-2]['service-name'], 'service1')
- self.assertEqual(res['output']['notification-service'][-2]['connection-type'], 'service')
- self.assertEqual(res['output']['notification-service'][-2]['message'], 'ServiceDelete request received ...')
self.assertEqual(res['output']['notification-service'][-1]['service-name'], 'service1')
+ self.assertEqual(res['output']['notification-service'][-1]['connection-type'], 'service')
self.assertEqual(res['output']['notification-service'][-1]['message'], 'Service deleted !')
time.sleep(2)
- def test_19_disconnect_XPDRA(self):
+ def test_24_disconnect_XPDRA(self):
response = test_utils.unmount_device("XPDR-A1")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
- def test_20_disconnect_XPDRC(self):
+ def test_25_disconnect_XPDRC(self):
response = test_utils.unmount_device("XPDR-C1")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
- def test_21_disconnect_ROADMA(self):
+ def test_26_disconnect_ROADMA(self):
response = test_utils.unmount_device("ROADM-A1")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
- def test_22_disconnect_ROADMC(self):
+ def test_27_disconnect_ROADMC(self):
response = test_utils.unmount_device("ROADM-C1")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
Code Review / transportpce.git / blobdiff