+#!/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)