1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
|
#!/usr/bin/python
from opnfv.deployment import factory
import functest.utils.openstack_utils as osutils
import functest.utils.functest_logger as ft_logger
import json
import logging
import os
import sys
import time
logger = ft_logger.Logger("netready-gluon-vping").getLogger()
logger.setLevel(logging.DEBUG)
class GluonVPing:
def __init__(self):
self.controller_node = None
self.port_ids = []
self.vpn_id = None
self.vpnbinding_ids = []
def prepare_test(self):
logger.info("Preparing Gluon vPing testcase.")
undercloud_ip = os.getenv("INSTALLER_IP")
handler = factory.Factory.get_handler('apex',
undercloud_ip,
'stack',
pkey_file='/root/.ssh/id_rsa')
nodes = handler.get_nodes()
self.controller_node = None
for node in nodes:
if node.is_controller():
self.controller_node = node
break
if self.controller_node is None:
logger.error("ERROR: no controller node found.")
raise Exception('No controller node found')
def run(self):
self._create_ports()
self._create_vpn()
self._create_vpnbinding()
#
# TODO boot VMs and ping
#
def _run_protonclient(self, options):
cmd = "protonclient --api net-l3vpn " + options
if self.controller_node is not None:
logger.debug("Running command: %s" % cmd)
output = self.controller_node.run_cmd(cmd)
if output is None:
raise Exception('Remote command failed')
else:
return output
def _create_ports(self):
def _build_create_port_options(name, mac, tenant_id):
return 'port-create' \
' --mac_address ' + mac + \
' --mtu 1500' \
' --admin_state_up True' \
' --name ' + name + \
' --vlan_transparency True' \
' --vnic_type normal' \
' --vif_type ovs' \
' --status ACTIVE' \
' --tenant_id ' + tenant_id
tenant_id = osutils.get_tenant_id(
osutils.get_keystone_client(),
'admin')
for i in [1, 2]:
options = _build_create_port_options(
'gluonPort' + str(i),
'c0:2a:14:04:43:0' + str(i),
tenant_id)
output = self._run_protonclient(options)
port_data = json.loads(output)
logger.info('Port %s created.' % port_data['id'])
self.port_ids.append(port_data['id'])
time.sleep(1)
def _create_vpn(self):
vpn_options = 'vpn-create '\
'--name "GluonVPN" ' \
'--ipv4_family 1000:1000 ' \
'--ipv6_family 1000:1000 ' \
'--route_distinguishers 1000:1000'
output = self._run_protonclient(vpn_options)
vpn_data = json.loads(output)
logger.info('VPN %s created.' % vpn_data['id'])
self.vpn_id = vpn_data['id']
time.sleep(1)
def _create_vpnbinding(self):
def _build_create_vpnbinding(port_id, counter):
return 'vpnbinding-create' \
' --interface_id ' + port_id + \
' --service_id ' + self.vpn_id + \
' --ipaddress 10.10.' + str(counter) + '.2' \
' --subnet_prefix 24' \
' --gateway 10.10.' + str(counter) + '.1'
counter = 0
for port_id in self.port_ids:
options = _build_create_vpnbinding(port_id, counter)
output = self._run_protonclient(options)
vpnbinding_data = json.loads(output)
logger.info('VPN-binding %s created.'
% vpnbinding_data['interface_id'])
self.vpnbinding_ids.append(vpnbinding_data['interface_id'])
counter += 1
time.sleep(1)
def _cleanup_ports(self):
for port_id in self.port_ids:
logger.info("Deleting port %s" % port_id)
self._run_protonclient('port-delete ' + str(port_id))
def _cleanup_vpn(self):
logger.info("Deleting VPN %s" % self.vpn_id)
self._run_protonclient('vpn-delete ' + str(self.vpn_id))
def _cleanup_vpnbindings(self):
for vpnbinding_id in self.vpnbinding_ids:
logger.info("Deleting VPN-binding %s" % vpnbinding_id)
self._run_protonclient('vpnbinding-delete ' + str(vpnbinding_id))
def cleanup(self):
logger.info("Cleaning up Gluon vPing testcase...")
self._cleanup_vpnbindings()
self._cleanup_vpn()
self._cleanup_ports()
def main():
logger.info("Running Gluon test suite...")
try:
gluon_vping = GluonVPing()
gluon_vping.prepare_test()
gluon_vping.run()
gluon_vping.cleanup()
except Exception as e:
logger.info("Gluon test suite failed: %s" % e)
return 1
logger.info("Gluon test suite successfully completed.")
return 0
if __name__ == '__main__':
sys.exit(main())
|
