from emuvim.api.tango.llcm import initialize_GK, parse_interface
from ipaddress import ip_network
+
PACKAGE_PATH = "misc/eu.5gtango.emulator-example-service.0.1.tgo"
# @unittest.skip("disabled")
def test_tango_llcm_start_service(self):
+ initialize_GK()
# create network
self.createNet(nswitches=0, ndatacenter=2, nhosts=2,
ndockers=0, enable_learning=True)
self.startNet()
time.sleep(3)
- print "starting tests"
# board package
files = {"package": open(PACKAGE_PATH, "rb")}
r = requests.post("http://127.0.0.1:56000/packages", files=files)
# check get request APIs
r3 = requests.get("http://127.0.0.1:56000/packages")
- self.assertEqual(len(json.loads(r3.text).get("service_uuid_list")), 1)
+ self.assertEqual(len(json.loads(r3.text)), 1)
r4 = requests.get("http://127.0.0.1:56000/instantiations")
- self.assertEqual(len(json.loads(r4.text).get(
- "service_instantiations_list")), 1)
+ self.assertEqual(len(json.loads(r4.text)), 1)
# check number of running nodes
- self.assertTrue(len(self.getContainernetContainers()) == 2)
- self.assertTrue(len(self.net.hosts) == 4)
- self.assertTrue(len(self.net.switches) == 2)
+ self.assertEqual(len(self.getContainernetContainers()), 2)
+ self.assertEqual(len(self.net.hosts), 4)
+ self.assertEqual(len(self.net.switches), 2)
# check compute list result (considering placement)
self.assertEqual(len(self.dc[0].listCompute()), 1)
self.assertEqual(len(self.dc[1].listCompute()), 1)
# check E-Line connection, by checking the IP addresses
for link in self.net.deployed_elines:
- vnf_src, intf_src, vnf_sap_docker_name = parse_interface(
+ vnf_src, intf_src = parse_interface(
link['connection_points_reference'][0])
- print vnf_src, intf_src
+ print(vnf_src, intf_src)
src = self.net.getNodeByName(vnf_src)
if not src:
continue
src_mask = [intf['netmask']
for intf in network_list if intf['intf_name'] == intf_src][0]
- vnf_dst, intf_dst, vnf_sap_docker_name = parse_interface(
+ vnf_dst, intf_dst = parse_interface(
link['connection_points_reference'][1])
dst = self.net.getNodeByName(vnf_dst)
if not dst:
dst_mask = [intf['netmask']
for intf in network_list if intf['intf_name'] == intf_dst][0]
- print "src = {0}:{1} ip={2} ".format(
- vnf_src, intf_src, src_ip, src_mask)
- print "dst = {0}:{1} ip={2} ".format(
- vnf_dst, intf_dst, dst_ip, dst_mask)
+ print("src = {0}:{1} ip={2} ".format(
+ vnf_src, intf_src, src_ip, src_mask))
+ print("dst = {0}:{1} ip={2} ".format(
+ vnf_dst, intf_dst, dst_ip, dst_mask))
# check if the E-Line IP's are in the same subnet
ret = ip_network(u'{0}'.format(src_ip, src_mask), strict=False)\
self.assertTrue(len(mgmt_ip) > 0)
ip_address = mgmt_ip[0]
ELAN_list.append(ip_address)
- print ip_address
+ print(ip_address)
# check ELAN connection by ping over the mgmt network (needs to be
# configured as ELAN in the test service)
for intf in network_list if intf['intf_name'] == 'mgmt']
self.assertTrue(len(mgmt_ip) > 0)
ip_address = mgmt_ip[0]
- print ELAN_list
- print ip_address
+ print(ELAN_list)
+ print(ip_address)
test_ip_list = list(ELAN_list)
test_ip_list.remove(ip_address)
for ip in test_ip_list:
# only take ip address, without netmask
p = self.net.ping([vnf], manualdestip=ip.split('/')[0])
- print p
+ print(p)
self.assertTrue(p <= 0.0)
# stop Mininet network
# @unittest.skip("disabled")
def test_tango_llcm_stop_service(self):
+ initialize_GK()
# create network
self.createNet(ndatacenter=2, nhosts=2)
# setup links
self.startNet()
time.sleep(3)
- print "starting tests"
+ print("starting tests")
# board package
files = {"package": open(PACKAGE_PATH, "rb")}
r = requests.post("http://127.0.0.1:56001/packages", files=files)
# check get request APIs
r3 = requests.get("http://127.0.0.1:56001/packages")
- self.assertEqual(len(json.loads(r3.text).get("service_uuid_list")), 1)
+ self.assertEqual(len(json.loads(r3.text)), 1)
r4 = requests.get("http://127.0.0.1:56001/instantiations")
- self.assertEqual(len(json.loads(r4.text).get(
- "service_instantiations_list")), 1)
+ self.assertEqual(len(json.loads(r4.text)), 1)
# check number of running nodes
- self.assertTrue(len(self.getContainernetContainers()) == 2)
- self.assertTrue(len(self.net.hosts) == 4)
- self.assertTrue(len(self.net.switches) == 2)
+ self.assertEqual(len(self.getContainernetContainers()), 2)
+ self.assertEqual(len(self.net.hosts), 4)
+ self.assertEqual(len(self.net.switches), 2)
# check compute list result (considering placement)
self.assertEqual(len(self.dc[0].listCompute()), 1)
self.assertEqual(len(self.dc[1].listCompute()), 1)
r5 = requests.get("http://127.0.0.1:56001/instantiations")
# note that there was 1 instance before
- self.assertTrue(len(json.loads(r5.text).get(
- "service_instantiations_list")), 0)
+ self.assertEqual(len(json.loads(r5.text)), 0)
# stop Mininet network
self.stopNet()