#!/usr/bin/env python ############################################################################## #Copyright (c) 2017 Orange, Inc. and others. All rights reserved. # # All rights reserved. This program and the accompanying materials # are made available under the terms of the Apache License, Version 2.0 # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 ############################################################################## import json import os import psutil import requests import signal import shutil import subprocess import time import unittest class TransportPCEtesting(unittest.TestCase): odl_process = None restconf_baseurl = "http://127.0.0.1:8181/restconf" @classmethod def __start_odl(cls): executable = "../karaf/target/assembly/bin/karaf" with open('odl.log', 'w') as outfile: cls.odl_process = subprocess.Popen( ["bash", executable], stdout=outfile, stdin=open(os.devnull)) @classmethod def setUpClass(cls): #a class method called before tests in an individual class run. cls.__start_odl() time.sleep(90) @classmethod def tearDownClass(cls): for child in psutil.Process(cls.odl_process.pid).children(): child.send_signal(signal.SIGINT) child.wait() cls.odl_process.send_signal(signal.SIGINT) cls.odl_process.wait() def setUp(self): #instruction executed before each test method time.sleep(1) #Get existing PathDesciption def test_01_get_pathdescriptions(self): url = ("{}/operational/stubpce:path-description-list/pathDescriptions/NodeAToNodeZ_direct_1" .format(self.restconf_baseurl)) headers = {'content-type': 'application/json', "Accept": "application/json"} response = requests.request( "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( res['pathDescriptions'][0]['path-name'], 'NodeAToNodeZ_direct_1') time.sleep(1) #Get non existing PathDesciption def test_02_get_pathdescriptions(self): url = ("{}/operational/stubpce:path-description-list/pathDescriptions/Node" .format(self.restconf_baseurl)) headers = {'content-type': 'application/json', "Accept": "application/json"} response = requests.request( "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, 404) time.sleep(1) #Create Service 'test' with correct parameters def test_03_create_service(self): url = ("{}/operations/org-openroadm-service:service-create" .format(self.restconf_baseurl)) data = {"input": { "sdnc-request-header": { "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58", "rpc-action": "service-create", "request-system-id": "appname", "notification-url": "http://localhost:8585/NotificationServer/notify" }, "service-name": "test", "common-id": "ASATT1234567", "connection-type": "infrastructure", "service-a-end": { "service-rate": "100", "node-id": "XPDRA", "service-format": "Ethernet", "clli": "SNJSCAMCJP8", "tx-direction": { "port": { "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00", "port-type": "router", "port-name": "Gigabit Ethernet_Tx.ge-5/0/0.0", "port-rack": "000000.00", "port-shelf": "00" }, "lgx": { "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00", "lgx-port-name": "LGX Back.3", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" } }, "rx-direction": { "port": { "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00", "port-type": "router", "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0", "port-rack": "000000.00", "port-shelf": "00" }, "lgx": { "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00", "lgx-port-name": "LGX Back.4", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" } }, "optic-type": "gray" }, "service-z-end": { "service-rate": "100", "node-id": "XPDRC", "service-format": "Ethernet", "clli": "SNJSCAMCJT4", "tx-direction": { "port": { "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00", "port-type": "router", "port-name": "Gigabit Ethernet_Tx.ge-1/0/0.0", "port-rack": "000000.00", "port-shelf": "00" }, "lgx": { "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00", "lgx-port-name": "LGX Back.29", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" } }, "rx-direction": { "port": { "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00", "port-type": "router", "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0", "port-rack": "000000.00", "port-shelf": "00" }, "lgx": { "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00", "lgx-port-name": "LGX Back.30", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" } }, "optic-type": "gray" }, "due-date": "2016-11-28T00:00:01Z", "operator-contact": "pw1234" } } headers = {'content-type': 'application/json', "Accept": "application/json"} response = requests.request( "POST", url, data=json.dumps(data), headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('in progress', res['output']['configuration-response-common']['response-message']) time.sleep(30) #Create Service 'test' with not compliant parameter : no 'sdnc-request-header' parameter def test_04_create_service(self): url = ("{}/operations/org-openroadm-service:service-create" .format(self.restconf_baseurl)) data = {"input": { "service-name": "test", "common-id": "ASATT1234567", "connection-type": "infrastructure", "service-a-end": { "service-rate": "100", "node-id": "XPDRA", "service-format": "Ethernet", "clli": "SNJSCAMCJP8", "tx-direction": { "port": { "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00", "port-type": "router", "port-name": "Gigabit Ethernet_Tx.ge-5/0/0.0", "port-rack": "000000.00", "port-shelf": "00" }, "lgx": { "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00", "lgx-port-name": "LGX Back.3", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" } }, "rx-direction": { "port": { "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00", "port-type": "router", "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0", "port-rack": "000000.00", "port-shelf": "00" }, "lgx": { "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00", "lgx-port-name": "LGX Back.4", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" } }, "optic-type": "gray" }, "service-z-end": { "service-rate": "100", "node-id": "XPDRC", "service-format": "Ethernet", "clli": "SNJSCAMCJT4", "tx-direction": { "port": { "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00", "port-type": "router", "port-name": "Gigabit Ethernet_Tx.ge-1/0/0.0", "port-rack": "000000.00", "port-shelf": "00" }, "lgx": { "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00", "lgx-port-name": "LGX Back.29", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" } }, "rx-direction": { "port": { "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00", "port-type": "router", "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0", "port-rack": "000000.00", "port-shelf": "00" }, "lgx": { "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00", "lgx-port-name": "LGX Back.30", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" } }, "optic-type": "gray" }, "due-date": "2016-11-28T00:00:01Z", "operator-contact": "pw1234" } } headers = {'content-type': 'application/json', "Accept": "application/json"} response = requests.request( "POST", url, data=json.dumps(data), headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Service not compliant', res['output']['configuration-response-common']['response-message']) time.sleep(5) #Create Service 'test' with not compliant parameter : no 'tx-direction' for serviceAEnd def test_05_create_service(self): url = ("{}/operations/org-openroadm-service:service-create" .format(self.restconf_baseurl)) data = {"input": { "sdnc-request-header": { "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58", "rpc-action": "service-create", "request-system-id": "appname", "notification-url": "http://localhost:8585/NotificationServer/notify" }, "service-name": "test", "common-id": "ASATT1234567", "connection-type": "infrastructure", "service-a-end": { "service-rate": "100", "node-id": "XPDRA", "service-format": "Ethernet", "clli": "SNJSCAMCJP8", "rx-direction": { "port": { "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00", "port-type": "router", "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0", "port-rack": "000000.00", "port-shelf": "00" }, "lgx": { "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00", "lgx-port-name": "LGX Back.4", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" } }, "optic-type": "gray" }, "service-z-end": { "service-rate": "100", "node-id": "XPDRC", "service-format": "Ethernet", "clli": "SNJSCAMCJT4", "tx-direction": { "port": { "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00", "port-type": "router", "port-name": "Gigabit Ethernet_Tx.ge-1/0/0.0", "port-rack": "000000.00", "port-shelf": "00" }, "lgx": { "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00", "lgx-port-name": "LGX Back.29", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" } }, "rx-direction": { "port": { "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00", "port-type": "router", "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0", "port-rack": "000000.00", "port-shelf": "00" }, "lgx": { "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00", "lgx-port-name": "LGX Back.30", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" } }, "optic-type": "gray" }, "due-date": "2016-11-28T00:00:01Z", "operator-contact": "pw1234" } } headers = {'content-type': 'application/json', "Accept": "application/json"} response = requests.request( "POST", url, data=json.dumps(data), headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Service not compliant', res['output']['configuration-response-common']['response-message']) time.sleep(5) #Get 'test' service created def test_06_get_service(self): url = ("{}/operational/org-openroadm-service:service-list/services/test" .format(self.restconf_baseurl)) headers = {'content-type': 'application/json', "Accept": "application/json"} response = requests.request( "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( res['services'][0]['administrative-state'], 'inService') time.sleep(1) #get non existing service def test_07_get_service(self): url = ("{}/operational/org-openroadm-service:service-list/services/test1" .format(self.restconf_baseurl)) headers = {'content-type': 'application/json', "Accept": "application/json"} response = requests.request( "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, 404) time.sleep(1) #reconfigure 'test' to be 'test-new' def test_08_reconfigure_service(self): url = ("{}/operations/org-openroadm-service:service-reconfigure" .format(self.restconf_baseurl)) data = {"input": { "service-name": "test", "new-service-name": "test-new", "common-id": "ASATT1234567", "connection-type": "infrastructure", "service-a-end": { "service-rate": "100", "node-id": "XPDRA", "service-format": "Ethernet", "clli": "SNJSCAMCJP8", "tx-direction": { "port": { "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00", "port-type": "router", "port-name": "Gigabit Ethernet_Tx.ge-5/0/0.0", "port-rack": "000000.00", "port-shelf": "00" }, "lgx": { "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00", "lgx-port-name": "LGX Back.3", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" } }, "rx-direction": { "port": { "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00", "port-type": "router", "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0", "port-rack": "000000.00", "port-shelf": "00" }, "lgx": { "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00", "lgx-port-name": "LGX Back.4", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" } }, "optic-type": "gray" }, "service-z-end": { "service-rate": "100", "node-id": "XPDRC", "service-format": "Ethernet", "clli": "SNJSCAMCJT4", "tx-direction": { "port": { "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00", "port-type": "router", "port-name": "Gigabit Ethernet_Tx.ge-1/0/0.0", "port-rack": "000000.00", "port-shelf": "00" }, "lgx": { "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00", "lgx-port-name": "LGX Back.29", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" } }, "rx-direction": { "port": { "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00", "port-type": "router", "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0", "port-rack": "000000.00", "port-shelf": "00" }, "lgx": { "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00", "lgx-port-name": "LGX Back.30", "lgx-port-rack": "000000.00", "lgx-port-shelf": "00" } }, "optic-type": "gray" }, "hard-constraints": { "diversity": { "existing-service": [ "104/GE100/SNJSCAMCJP8/SNJSCAMCJT4" ], "existing-service-applicability": { "node": "true" } }, "exclude": { "fiber-bundle": [ "l(string)" ], "node-id": [ "SNJSCAMCJP8_000000.00" ] }, "latency": { "max-latency": "30" } } } } headers = {'content-type': 'application/json', "Accept": "application/json"} response = requests.request( "POST", url, data=json.dumps(data), headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('in progress', res['output']['status-message']) time.sleep(30) #get new service 'test-new' def test_09_get_service(self): url = ("{}/operational/org-openroadm-service:service-list/services/test-new" .format(self.restconf_baseurl)) headers = {'content-type': 'application/json', "Accept": "application/json"} response = requests.request( "GET", url, headers=headers, auth=('admin', 'admin')) res = response.json() self.assertEqual( res['services'][0]['operational-state'], 'inService') time.sleep(1) #Modify 'test-new' state def test_10_modify_service_state(self): url = ("{}/operations/servicehandler:service-state-modify" .format(self.restconf_baseurl)) data = {"input": { "service-name": "test-new", "operational-state": "outOfService" } } headers = {'content-type': 'application/json'} response = requests.request( "POST", url, data=json.dumps(data), headers=headers, auth=('admin', 'admin')) res = response.json() self.assertIn('Service state modified', res['output']['configuration-response-common']['response-message']) time.sleep(5) #get new service 'test-new' state def test_11_get_service(self): url = ("{}/operational/org-openroadm-service:service-list/services/test-new" .format(self.restconf_baseurl)) headers = {'content-type': 'application/json', "Accept": "application/json"} response = requests.request( "GET", url, headers=headers, auth=('admin', 'admin')) res = response.json() self.assertEqual( res['services'][0]['operational-state'], 'outOfService') time.sleep(1) #restore service 'test-new' def test_12_restore_service(self): url = ("{}/operations/org-openroadm-service:service-restoration" .format(self.restconf_baseurl)) data = {"input": { "service-name": "test-new", "option": "permanent" } } headers = {'content-type': 'application/json'} response = requests.request( "POST", url, data=json.dumps(data), headers=headers, auth=('admin', 'admin')) res = response.json() self.assertIn('in progress', res['output']['status-message']) time.sleep(60) #get new service 'test-new' state def test_13_get_service(self): url = ("{}/operational/org-openroadm-service:service-list/services/test-new" .format(self.restconf_baseurl)) headers = {'content-type': 'application/json', "Accept": "application/json"} response = requests.request( "GET", url, headers=headers, auth=('admin', 'admin')) res = response.json() self.assertEqual( res['services'][0]['operational-state'], 'inService') time.sleep(1) #Delete 'test-new' service def test_14_delete_service(self): url = ("{}/operations/org-openroadm-service:service-delete" .format(self.restconf_baseurl)) data = {"input": { "sdnc-request-header": { "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58", "rpc-action": "service-delete", "request-system-id": "appname", "notification-url": "http://localhost:8585/NotificationServer/notify" }, "service-delete-req-info": { "service-name": "test-new", "due-date": "2016-11-28T00:00:01Z", "tail-retention": "no" } } } headers = {'content-type': 'application/json'} response = requests.request( "POST", url, data=json.dumps(data), headers=headers, auth=('admin', 'admin')) res = response.json() self.assertIn('in progress', res['output']['configuration-response-common']['response-message']) time.sleep(30) #Delete non existing service def test_15_delete_service(self): url = ("{}/operations/org-openroadm-service:service-delete" .format(self.restconf_baseurl)) data = {"input": { "sdnc-request-header": { "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58", "rpc-action": "service-delete", "request-system-id": "appname", "notification-url": "http://localhost:8585/NotificationServer/notify" }, "service-delete-req-info": { "service-name": "test", "due-date": "2016-11-28T00:00:01Z", "tail-retention": "no" } } } headers = {'content-type': 'application/json'} response = requests.request( "POST", url, data=json.dumps(data), headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('not exists in datastore', res['output']['configuration-response-common']['response-message']) time.sleep(1) #Verify 'test' service deleted def test_16_get_service(self): url = ("{}/operational/org-openroadm-service:service-list/services/test-new" .format(self.restconf_baseurl)) headers = {'content-type': 'application/json', "Accept": "application/json"} response = requests.request( "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, 404) time.sleep(1) if __name__ == "__main__": unittest.main(verbosity=2)