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
|
from neutronclient.v2_0 import client as neutron
from novaclient import client as nova
from novaclient.exceptions import NotFound
import time
import datetime
import os
import subprocess
class FDSLibrary():
def __init__(self):
self.neutron_client = neutron.Client(username=os.getenv('OS_USERNAME'),
password=os.getenv('OS_PASSWORD'),
tenant_name=os.getenv('OS_TENANT_NAME'),
auth_url=os.getenv('OS_AUTH_URL'))
self.nova_client = nova.Client('2',
os.getenv('OS_USERNAME'),
os.getenv('OS_PASSWORD'),
os.getenv('OS_TENANT_NAME'),
os.getenv('OS_AUTH_URL'))
def create_network(self, name):
body = {'network': {'name': name}}
response = self.neutron_client.create_network(body=body)
return response
def create_subnet(self, name, network_id, cidr, dhcp=True, ip_version=4):
body = {'subnet': {
'name': name,
'enable_dhcp': dhcp,
'network_id': network_id,
'ip_version': ip_version,
'cidr': cidr
}}
response = self.neutron_client.create_subnet(body=body)
return response
def create_port(self, name, net_id, subnet_id, ip_address, security_groups=None):
body = {'port': {
'name': name,
'network_id': net_id,
'fixed_ips': [
{
'subnet_id': subnet_id,
'ip_address': ip_address
}
]
}}
if security_groups is not None:
# this is a list
body['port']['security_groups'] = security_groups
response = self.neutron_client.create_port(body=body)
return response
def create_server(self, name, image, flavor, port_ids, security_groups=None):
image = self.nova_client.images.find(name=image)
flavor = self.nova_client.flavors.find(name=flavor)
nics = [{'port-id': port_id} for port_id in port_ids]
response = self.nova_client.servers.create(name=name, image=image.id, flavor=flavor.id,
security_groups=security_groups, nics=nics)
for key in dir(response):
print key, getattr(response, key)
return response
def poll_server(self, vm_id, status, timeout=300):
try:
start = datetime.datetime.now()
delta = datetime.timedelta(seconds=timeout)
vm_obj = self.nova_client.servers.get(vm_id)
current_state = getattr(vm_obj, 'OS-EXT-STS:vm_state', None)
print "Found state '{}' while looking for state '{}'".format(current_state, status)
while datetime.datetime.now() - start < delta and self._states_are_not_equal(current_state, status):
time.sleep(1)
vm_obj = self.nova_client.servers.get(vm_id)
current_state = getattr(vm_obj, 'OS-EXT-STS:vm_state', None)
print "Found state '{}' while looking for state '{}'".format(current_state, status)
if self._states_are_not_equal(current_state, status):
raise Exception("Vm with id '{}' is in state '{}' after timeout of {} seconds"
"".format(vm_id, getattr(vm_obj, 'OS-EXT-STS:vm_state'), timeout))
except NotFound as e:
if status is not None:
raise e
else:
print "VM not found, which means it's been deleted"
def _states_are_not_equal(self, current_state, expected_state):
return current_state.lower() != expected_state
def delete_server(self, vm_id):
response = self.nova_client.servers.delete(vm_id)
return response
def delete_port(self, port_id):
response = self.neutron_client.delete_port(port_id)
return response
def delete_net(self, net_id):
response = self.neutron_client.delete_network(net_id)
return response
def ping_vm(self, ip_address):
try:
output = subprocess.check_output(['ping', '-c', '4', ip_address])
print output
except subprocess.CalledProcessError as e:
print e.output
raise Exception('Ping on ip {} failed'.format(ip_address))
|
