3 ##############################################################################
4 #Copyright (c) 2017 Orange, Inc. and others. All rights reserved.
6 # All rights reserved. This program and the accompanying materials
7 # are made available under the terms of the Apache License, Version 2.0
8 # which accompanies this distribution, and is available at
9 # http://www.apache.org/licenses/LICENSE-2.0
10 ##############################################################################
23 class TransportGNPYtesting(unittest.TestCase):
27 restconf_baseurl = "http://localhost:8181/restconf"
30 def __init_logfile(cls):
31 if os.path.isfile("./transportpce_tests/gnpy.log"):
32 os.remove("transportpce_tests/gnpy.log")
36 executable = "../karaf/target/assembly/bin/karaf"
37 with open('transportpce_tests/odl.log', 'w') as outfile:
38 cls.odl_process = subprocess.Popen(
39 ["bash", executable, "server"], stdout=outfile,
40 stdin=open(os.devnull))
48 def tearDownClass(cls):
49 for child in psutil.Process(cls.odl_process.pid).children():
50 child.send_signal(signal.SIGINT)
52 cls.odl_process.send_signal(signal.SIGINT)
53 cls.odl_process.wait()
58 #Mount the different topologies
59 def test_01_connect_clliNetwork(self):
60 url = ("{}/config/ietf-network:networks/network/clli-network"
61 .format(self.restconf_baseurl))
62 topo_clliNet_file = "sample_configs/gnpy/clliNetwork.json"
63 if os.path.isfile(topo_clliNet_file):
64 with open(topo_clliNet_file, 'r') as clli_net:
65 body = clli_net.read()
66 headers = {'content-type': 'application/json'}
67 response = requests.request(
68 "PUT", url, data=body, headers=headers,
69 auth=('admin', 'admin'))
70 self.assertEqual(response.status_code, requests.codes.ok)
73 def test_02_connect_openroadmNetwork(self):
74 url = ("{}/config/ietf-network:networks/network/openroadm-network"
75 .format(self.restconf_baseurl))
76 topo_ordNet_file = "sample_configs/gnpy/openroadmNetwork.json"
77 if os.path.isfile(topo_ordNet_file):
78 with open(topo_ordNet_file, 'r') as ord_net:
80 headers = {'content-type': 'application/json'}
81 response = requests.request(
82 "PUT", url, data=body, headers=headers,
83 auth=('admin', 'admin'))
84 self.assertEqual(response.status_code, requests.codes.ok)
87 def test_03_connect_openroadmTopology(self):
88 url = ("{}/config/ietf-network:networks/network/openroadm-topology"
89 .format(self.restconf_baseurl))
90 topo_ordTopo_file = "sample_configs/gnpy/openroadmTopology.json"
91 if os.path.isfile(topo_ordTopo_file):
92 with open(topo_ordTopo_file, 'r') as ord_topo:
93 body = ord_topo.read()
94 headers = {'content-type': 'application/json'}
95 response = requests.request(
96 "PUT", url, data=body, headers=headers,
97 auth=('admin', 'admin'))
98 self.assertEqual(response.status_code, requests.codes.ok)
102 def test_04_path_computation_xpdr_bi(self):
103 url = ("{}/operations/transportpce-pce:path-computation-request"
104 .format(self.restconf_baseurl))
106 "service-name": "service-1",
107 "resource-reserve": "true",
108 "pce-metric": "hop-count",
109 "service-handler-header": {
110 "request-id": "request-1"
114 "service-rate": "100",
119 "service-rate": "100",
124 headers = {'content-type': 'application/json',
125 "Accept": "application/json"}
126 response = requests.request(
127 "POST", url, data=json.dumps(body), headers=headers,
128 auth=('admin', 'admin'))
129 self.assertEqual(response.status_code, requests.codes.ok)
130 res = response.json()
131 self.assertEqual(res['output']['gnpy-response'][0]['path-dir'], 'A-to-Z')
132 self.assertEqual(res['output']['gnpy-response'][0]['feasibility'],True)
133 self.assertEqual(res['output']['gnpy-response'][1]['path-dir'], 'Z-to-A')
134 self.assertEqual(res['output']['gnpy-response'][1]['feasibility'],True)
137 #Disconnect the different topology
138 def test_05_disconnect_openroadmTopology(self):
139 url = ("{}/config/ietf-network:networks/network/openroadm-topology"
140 .format(self.restconf_baseurl))
142 headers = {'content-type': 'application/json'}
143 response = requests.request(
144 "DELETE", url, data=json.dumps(data), headers=headers,
145 auth=('admin', 'admin'))
146 self.assertEqual(response.status_code, requests.codes.ok)
149 def test_06_disconnect_openroadmNetwork(self):
151 url = ("{}/config/ietf-network:networks/network/openroadm-network"
152 .format(self.restconf_baseurl))
154 headers = {'content-type': 'application/json'}
155 response = requests.request(
156 "DELETE", url, data=json.dumps(data), headers=headers,
157 auth=('admin', 'admin'))
158 self.assertEqual(response.status_code, requests.codes.ok)
161 def test_07_disconnect_clliNetwork(self):
162 url = ("{}/config/ietf-network:networks/network/clli-network"
163 .format(self.restconf_baseurl))
165 headers = {'content-type': 'application/json'}
166 response = requests.request(
167 "DELETE", url, data=json.dumps(data), headers=headers,
168 auth=('admin', 'admin'))
169 self.assertEqual(response.status_code, requests.codes.ok)
172 if __name__ == "__main__":
173 #logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG)
174 unittest.main(verbosity=2)