Revision cee3ee9b snf-tools/synnefo_tools/burnin/server_tests.py
b/snf-tools/synnefo_tools/burnin/server_tests.py | ||
---|---|---|
38 | 38 |
|
39 | 39 |
import sys |
40 | 40 |
import IPy |
41 |
import stat |
|
42 |
import base64 |
|
41 | 43 |
import random |
42 | 44 |
import socket |
43 | 45 |
|
... | ... | |
52 | 54 |
class GeneratedServerTestSuite(CycladesTests): |
53 | 55 |
"""Test Spawning Serverfunctionality""" |
54 | 56 |
use_image = Proper(value=None) |
57 |
personality = Proper(value=None) |
|
55 | 58 |
avail_flavors = Proper(value=None) |
56 | 59 |
use_flavor = Proper(value=None) |
57 | 60 |
server = Proper(value=None) |
58 | 61 |
ipv4 = Proper(value=None) |
59 | 62 |
ipv6 = Proper(value=None) |
63 |
username = Proper(value=None) |
|
64 |
password = Proper(value=None) |
|
60 | 65 |
|
61 | 66 |
def test_001_submit_create_server(self): |
62 | 67 |
"""Submit a create server request""" |
68 |
if self._image_is(self.use_image, "linux"): |
|
69 |
# Enforce personality test |
|
70 |
self.info("Creating personality content to be used") |
|
71 |
self.personality = [{ |
|
72 |
'path': "/root/test_inj_file", |
|
73 |
'owner': "root", |
|
74 |
'group': "root", |
|
75 |
'mode': stat.S_IRUSR | stat.S_IWUSR, |
|
76 |
'contents': base64.b64encode("This is a personality file") |
|
77 |
}] |
|
63 | 78 |
servername = "%s for %s" % (self.run_id, self.use_image['name']) |
64 | 79 |
self.use_flavor = random.choice(self.avail_flavors) |
65 | 80 |
|
66 | 81 |
self.server = self._create_server( |
67 |
servername, self.use_image, self.use_flavor) |
|
82 |
servername, self.use_image, self.use_flavor, self.personality) |
|
83 |
self.username = self._get_connection_username(self.server) |
|
84 |
self.password = self.server['adminPass'] |
|
68 | 85 |
|
69 | 86 |
def test_002_server_build_list(self): |
70 | 87 |
"""Test server is in BUILD state, in server list""" |
... | ... | |
147 | 164 |
# Update the server attribute |
148 | 165 |
self.server = server |
149 | 166 |
|
150 |
self.ipv4 = self._get_ip(server, 4) |
|
167 |
self.ipv4 = self._get_ip(server, version=4)
|
|
151 | 168 |
self.assertEquals(IPy.IP(self.ipv4).version(), 4) |
152 | 169 |
|
153 | 170 |
def test_008_server_has_ipv6(self): |
154 | 171 |
"""Test active server has a valid IPv6 address""" |
155 | 172 |
self._skip_if(not self.use_ipv6, "--no-ipv6 flag enabled") |
156 | 173 |
|
157 |
self.ipv6 = self._get_ip(self.server, 6) |
|
174 |
self.ipv6 = self._get_ip(self.server, version=6)
|
|
158 | 175 |
self.assertEquals(IPy.IP(self.ipv6).version(), 6) |
159 | 176 |
|
160 | 177 |
def test_009_server_ping_ipv4(self): |
161 | 178 |
"""Test server responds to ping on IPv4 address""" |
162 |
self._insist_on_ping(self.ipv4, 4) |
|
179 |
self._insist_on_ping(self.ipv4, version=4)
|
|
163 | 180 |
|
164 | 181 |
def test_010_server_ping_ipv6(self): |
165 | 182 |
"""Test server responds to ping on IPv6 address""" |
166 | 183 |
self._skip_if(not self.use_ipv6, "--no-ipv6 flag enabled") |
167 |
self._insist_on_ping(self.ipv6, 6) |
|
184 |
self._insist_on_ping(self.ipv6, version=6)
|
|
168 | 185 |
|
169 | 186 |
def test_011_submit_shutdown(self): |
170 | 187 |
"""Test submit request to shutdown server""" |
... | ... | |
186 | 203 |
"""Test server OS is actually up and running again""" |
187 | 204 |
self.test_009_server_ping_ipv4() |
188 | 205 |
|
206 |
def test_016_ssh_to_server_ipv4(self): |
|
207 |
"""Test SSH to server public IPv4 works, verify hostname""" |
|
208 |
self._skip_if(not self._image_is(self.use_image, "linux"), |
|
209 |
"only valid for Linux servers") |
|
210 |
hostname = self._insist_get_hostname_over_ssh( |
|
211 |
self.ipv4, self.username, self.password) |
|
212 |
# The hostname must be of the form 'prefix-id' |
|
213 |
self.assertTrue(hostname.endswith("-%d" % self.server['id'])) |
|
214 |
|
|
215 |
def test_017_ssh_to_server_ipv6(self): |
|
216 |
"""Test SSH to server public IPv6 works, verify hostname""" |
|
217 |
self._skip_if(not self._image_is(self.use_image, "linux"), |
|
218 |
"only valid for Linux servers") |
|
219 |
self._skip_if(not self.use_ipv6, "--no-ipv6 flag enabled") |
|
220 |
hostname = self._insist_get_hostname_over_ssh( |
|
221 |
self.ipv6, self.username, self.password) |
|
222 |
# The hostname must be of the form 'prefix-id' |
|
223 |
self.assertTrue(hostname.endswith("-%d" % self.server['id'])) |
|
224 |
|
|
225 |
def test_018_rdp_to_server_ipv4(self): |
|
226 |
"""Test RDP connection to server public IPv4 works""" |
|
227 |
self._skip_if(not self._image_is(self.use_image, "windows"), |
|
228 |
"only valid for Windows servers") |
|
229 |
sock = self._insist_on_tcp_connection(socket.AF_INET, self.ipv4, 3389) |
|
230 |
# No actual RDP processing done. We assume the RDP server is there |
|
231 |
# if the connection to the RDP port is successful. |
|
232 |
# pylint: disable-msg=W0511 |
|
233 |
# FIXME: Use rdesktop, analyze exit code? see manpage |
|
234 |
sock.close() |
|
235 |
|
|
236 |
def test_019_rdp_to_server_ipv6(self): |
|
237 |
"""Test RDP connection to server public IPv6 works""" |
|
238 |
self._skip_if(not self._image_is(self.use_image, "windows"), |
|
239 |
"only valid for Windows servers") |
|
240 |
self._skip_if(not self.use_ipv6, "--no-ipv6 flag enabled") |
|
241 |
sock = self._insist_on_tcp_connection(socket.AF_INET, self.ipv6, 3389) |
|
242 |
# No actual RDP processing done. We assume the RDP server is there |
|
243 |
# if the connection to the RDP port is successful. |
|
244 |
# pylint: disable-msg=W0511 |
|
245 |
# FIXME: Use rdesktop, analyze exit code? see manpage |
|
246 |
sock.close() |
|
247 |
|
|
248 |
def test_020_personality(self): |
|
249 |
"""Test file injection for personality enforcement""" |
|
250 |
self._skip_if(not self._image_is(self.use_image, "linux"), |
|
251 |
"only implemented for linux servers") |
|
252 |
assert self.personality is not None, "No personality used" |
|
253 |
|
|
254 |
for inj_file in self.personality: |
|
255 |
self._check_file_through_ssh( |
|
256 |
self.ipv4, inj_file['owner'], self.password, |
|
257 |
inj_file['path'], inj_file['contents']) |
|
258 |
|
|
259 |
def test_021_submit_delete_request(self): |
|
260 |
"""Test submit request to delete server""" |
|
261 |
self.clients.cyclades.delete_server(self.server['id']) |
|
262 |
|
|
263 |
def test_022_server_becomes_deleted(self): |
|
264 |
"""Test server becomes DELETED""" |
|
265 |
self._insist_on_server_transition(self.server, "ACTIVE", "DELETED") |
|
266 |
|
|
267 |
def test_023_server_no_longer(self): |
|
268 |
"""Test server is no longer in server list""" |
|
269 |
servers = self._get_list_of_servers() |
|
270 |
self.assertNotIn(self.server['id'], [s['id'] for s in servers]) |
|
271 |
|
|
189 | 272 |
|
190 | 273 |
# -------------------------------------------------------------------- |
191 | 274 |
# The actuall test class. We use this class to dynamically create |
Also available in: Unified diff