Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in
Toggle navigation
Menu
Open sidebar
itminedu
synnefo
Commits
b0a039fe
Commit
b0a039fe
authored
Nov 19, 2013
by
Ilias Tsitsimpis
Browse files
burnin: Write ServerTestSuite
parent
befc1d2c
Changes
6
Hide whitespace changes
Inline
Side-by-side
snf-tools/synnefo_tools/burnin/__init__.py
View file @
b0a039fe
...
...
@@ -45,6 +45,7 @@ from synnefo_tools.burnin.astakos_tests import AstakosTestSuite
from
synnefo_tools.burnin.images_tests
import
\
FlavorsTestSuite
,
ImagesTestSuite
from
synnefo_tools.burnin.pithos_tests
import
PithosTestSuite
from
synnefo_tools.burnin.server_tests
import
ServerTestSuite
# --------------------------------------------------------------------
...
...
@@ -54,6 +55,7 @@ TESTSUITES = [
FlavorsTestSuite
,
ImagesTestSuite
,
PithosTestSuite
,
ServerTestSuite
,
]
TSUITES_NAMES
=
[
tsuite
.
__name__
for
tsuite
in
TESTSUITES
]
...
...
@@ -68,13 +70,8 @@ def string_to_class(names):
# Parse arguments
def
parse_comma
(
option
,
_
,
value
,
parser
):
"""Parse comma separated arguments"""
tests
=
set
(
TSUITES_NAMES
)
parse_input
=
value
.
split
(
','
)
if
not
(
set
(
parse_input
)).
issubset
(
tests
):
raise
optparse
.
OptionValueError
(
"The selected set of tests is invalid"
)
setattr
(
parser
.
values
,
option
.
dest
,
value
.
split
(
','
))
parse_input
=
[
p
.
strip
()
for
p
in
value
.
split
(
','
)]
setattr
(
parser
.
values
,
option
.
dest
,
parse_input
)
def
parse_arguments
(
args
):
...
...
@@ -120,18 +117,18 @@ def parse_arguments(args):
help
=
"Query server status when requests are pending "
"every INTERVAL seconds"
)
parser
.
add_option
(
"--
force-
flavor"
,
action
=
"
store"
,
type
=
"string"
,
default
=
None
,
dest
=
"
force_
flavor"
,
metavar
=
"FLAVOR"
,
help
=
"Force all server creations to use the specified FLAVOR "
"--flavor
s
"
,
action
=
"
callback"
,
callback
=
parse_comma
,
type
=
"string"
,
default
=
None
,
dest
=
"flavor
s
"
,
metavar
=
"FLAVOR
S
"
,
help
=
"Force all server creations to use
one of
the specified FLAVOR
S
"
"instead of a randomly chosen one. Supports both search by name "
"(reg expression) with
\"
name:flavor name
\"
or by id with "
"
\"
id:flavor id
\"
"
)
parser
.
add_option
(
"--
force-
image"
,
action
=
"
store"
,
type
=
"string"
,
default
=
None
,
dest
=
"
force_
image"
,
metavar
=
"IMAGE"
,
help
=
"Force all server creations to use the specified IMAGE "
"--image
s
"
,
action
=
"
callback"
,
callback
=
parse_comma
,
type
=
"string"
,
default
=
None
,
dest
=
"image
s
"
,
metavar
=
"IMAGE
S
"
,
help
=
"Force all server creations to use
one of
the specified IMAGE
S
"
"instead of the default one (a Debian Base image). Just like the "
"--
force-
flavor option, it supports both search by name and id"
)
"--flavor
s
option, it supports both search by name and id"
)
parser
.
add_option
(
"--system-user"
,
action
=
"store"
,
type
=
"string"
,
default
=
None
,
dest
=
"system_user"
,
...
...
@@ -203,6 +200,14 @@ def parse_arguments(args):
if
opts
.
final_report
:
opts
.
quiet
=
True
# Check `--set-tests' and `--exclude-tests' options
if
opts
.
tests
!=
"all"
and
\
not
(
set
(
opts
.
tests
)).
issubset
(
set
(
TSUITES_NAMES
)):
raise
optparse
.
OptionValueError
(
"The selected set of tests is invalid"
)
if
opts
.
exclude_tests
is
not
None
and
\
not
(
set
(
opts
.
exclude_tests
)).
issubset
(
set
(
TSUITES_NAMES
)):
raise
optparse
.
OptionValueError
(
"The selected set of tests is invalid"
)
# `token' is mandatory
mandatory_argument
(
opts
.
token
,
"--token"
)
# `auth_url' is mandatory
...
...
@@ -245,8 +250,8 @@ def main():
# Run burnin
# The return value denotes the success status
return
common
.
run
(
testsuites
,
failfast
=
opts
.
failfast
,
final_report
=
opts
.
final_report
)
return
common
.
run
_burnin
(
testsuites
,
failfast
=
opts
.
failfast
,
final_report
=
opts
.
final_report
)
if
__name__
==
"__main__"
:
...
...
snf-tools/synnefo_tools/burnin/common.py
View file @
b0a039fe
...
...
@@ -38,12 +38,17 @@ Common utils for burnin tests
import
os
import
re
import
time
import
shutil
import
socket
import
random
import
unittest
import
datetime
import
tempfile
import
traceback
import
subprocess
from
kamaki.clients.cyclades
import
CycladesClient
from
kamaki.clients.astakos
import
AstakosClient
from
kamaki.clients.compute
import
ComputeClient
from
kamaki.clients.pithos
import
PithosClient
...
...
@@ -104,7 +109,8 @@ class BurninTestResult(unittest.TestResult):
# --------------------------------------------------------------------
# BurninTests class
# Too few public methods (0/2). pylint: disable-msg=R0903
# Too few public methods. pylint: disable-msg=R0903
# Too many instance attributes. pylint: disable-msg=R0902
class
Clients
(
object
):
"""Our kamaki clients"""
auth_url
=
None
...
...
@@ -135,6 +141,8 @@ class BurninTests(unittest.TestCase):
action_warning
=
None
query_interval
=
None
system_user
=
None
images
=
None
flavors
=
None
@
classmethod
def
setUpClass
(
cls
):
# noqa
...
...
@@ -160,6 +168,10 @@ class BurninTests(unittest.TestCase):
self
.
clients
.
compute_url
,
self
.
clients
.
token
)
self
.
clients
.
compute
.
CONNECTION_RETRY_LIMIT
=
self
.
clients
.
retry
self
.
clients
.
cyclades
=
CycladesClient
(
self
.
clients
.
compute_url
,
self
.
clients
.
token
)
self
.
clients
.
cyclades
.
CONNECTION_RETRY_LIMIT
=
self
.
clients
.
retry
self
.
clients
.
pithos_url
=
self
.
clients
.
astakos
.
\
get_service_endpoints
(
'object-store'
)[
'publicURL'
]
self
.
info
(
"Pithos url is %s"
,
self
.
clients
.
pithos_url
)
...
...
@@ -268,6 +280,36 @@ class BurninTests(unittest.TestCase):
self
.
warning
(
"No system user found"
)
return
None
def
_try_until_timeout_expires
(
self
,
opmsg
,
check_fun
):
"""Try to perform an action until timeout expires"""
assert
callable
(
check_fun
),
"Not a function"
action_timeout
=
self
.
action_timeout
action_warning
=
self
.
action_warning
if
action_warning
>
action_timeout
:
action_warning
=
action_timeout
start_time
=
time
.
time
()
while
(
start_time
+
action_warning
)
>
time
.
time
():
try
:
return
check_fun
()
except
Retry
:
time
.
sleep
(
self
.
query_interval
)
self
.
warning
(
"Operation `%s' is taking too long"
,
opmsg
)
while
(
start_time
+
action_timeout
)
>
time
.
time
():
try
:
return
check_fun
()
except
Retry
:
time
.
sleep
(
self
.
query_interval
)
self
.
error
(
"Operation `%s' timed out"
,
opmsg
)
self
.
fail
(
"time out"
)
def
_skip_if
(
self
,
condition
,
msg
):
"""Skip tests"""
if
condition
:
self
.
info
(
"Test skipped: %s"
%
msg
)
self
.
skipTest
(
msg
)
# ----------------------------------
# Flavors
def
_get_list_of_flavors
(
self
,
detail
=
False
):
...
...
@@ -279,6 +321,48 @@ class BurninTests(unittest.TestCase):
flavors
=
self
.
clients
.
compute
.
list_flavors
(
detail
=
detail
)
return
flavors
def
_find_flavors
(
self
,
patterns
,
flavors
=
None
):
"""Find a list of suitable flavors to use
The patterns is a list of `typed_options'. A list of all flavors
matching this patterns will be returned.
"""
if
flavors
is
None
:
flavors
=
self
.
_get_list_of_flavors
(
detail
=
True
)
ret_flavors
=
[]
for
ptrn
in
patterns
:
parsed_ptrn
=
parse_typed_option
(
ptrn
)
if
parsed_ptrn
is
None
:
msg
=
"Invalid flavor format: %s. Must be [id|name]:.+"
self
.
warning
(
msg
,
ptrn
)
continue
flv_type
,
flv_value
=
parsed_ptrn
if
flv_type
==
"name"
:
# Filter flavor by name
msg
=
"Trying to find a flavor with name %s"
self
.
info
(
msg
,
flv_value
)
filtered_flvs
=
\
[
f
for
f
in
flavors
if
re
.
search
(
flv_value
,
f
[
'name'
],
flags
=
re
.
I
)
is
not
None
]
elif
flv_type
==
"id"
:
# Filter flavors by id
msg
=
"Trying to find a flavor with id %s"
self
.
info
(
msg
,
flv_value
)
filtered_flvs
=
\
[
f
for
f
in
flavors
if
str
(
f
[
'id'
])
==
flv_value
]
else
:
self
.
error
(
"Unrecognized flavor type %s"
,
flv_type
)
self
.
fail
(
"Unrecognized flavor type"
)
# Append and continue
ret_flavors
.
extend
(
filtered_flvs
)
self
.
assertGreater
(
len
(
ret_flavors
),
0
,
"No matching flavors found"
)
return
ret_flavors
# ----------------------------------
# Images
def
_get_list_of_images
(
self
,
detail
=
False
):
...
...
@@ -306,16 +390,17 @@ class BurninTests(unittest.TestCase):
return
ret_images
def
_find_image
(
self
,
patterns
,
images
=
None
):
"""Find a suitable image to use
def
_find_image
s
(
self
,
patterns
,
images
=
None
):
"""Find a
list of
suitable image
s
to use
The patterns is a list of `typed_options'.
The first pattern to
match
an image will be the one that
will be returned.
The patterns is a list of `typed_options'.
A list of all images
match
ing this patterns
will be returned.
"""
if
images
is
None
:
images
=
self
.
_get_list_of_sys_images
()
ret_images
=
[]
for
ptrn
in
patterns
:
parsed_ptrn
=
parse_typed_option
(
ptrn
)
if
parsed_ptrn
is
None
:
...
...
@@ -341,16 +426,12 @@ class BurninTests(unittest.TestCase):
self
.
error
(
"Unrecognized image type %s"
,
img_type
)
self
.
fail
(
"Unrecognized image type"
)
# Check if we found one
if
filtered_imgs
:
img
=
filtered_imgs
[
0
]
self
.
info
(
"Will use %s with id %s"
,
img
[
'name'
],
img
[
'id'
])
return
img
# Append and continue
ret_images
.
extend
(
filtered_imgs
)
# We didn't found one
err
=
"No matching image found"
self
.
error
(
err
)
self
.
fail
(
err
)
self
.
assertGreater
(
len
(
ret_images
),
0
,
"No matching images found"
)
return
ret_images
# ----------------------------------
# Pithos
...
...
@@ -387,6 +468,148 @@ class BurninTests(unittest.TestCase):
self
.
clients
.
pithos
.
container
=
container
self
.
clients
.
pithos
.
container_put
()
# ----------------------------------
# Servers
def
_get_list_of_servers
(
self
,
detail
=
False
):
"""Get (detailed) list of servers"""
if
detail
:
self
.
info
(
"Getting detailed list of servers"
)
else
:
self
.
info
(
"Getting simple list of servers"
)
return
self
.
clients
.
cyclades
.
list_servers
(
detail
=
detail
)
def
_get_server_details
(
self
,
server
):
"""Get details for a server"""
self
.
info
(
"Getting details for server %s with id %s"
,
server
[
'name'
],
server
[
'id'
])
return
self
.
clients
.
cyclades
.
get_server_details
(
server
[
'id'
])
def
_create_server
(
self
,
name
,
image
,
flavor
):
"""Create a new server"""
self
.
info
(
"Creating a server with name %s"
,
name
)
self
.
info
(
"Using image %s with id %s"
,
image
[
'name'
],
image
[
'id'
])
self
.
info
(
"Using flavor %s with id %s"
,
flavor
[
'name'
],
flavor
[
'id'
])
server
=
self
.
clients
.
cyclades
.
create_server
(
name
,
flavor
[
'id'
],
image
[
'id'
])
self
.
info
(
"Server id: %s"
,
server
[
'id'
])
self
.
info
(
"Server password: %s"
,
server
[
'adminPass'
])
self
.
assertEqual
(
server
[
'name'
],
name
)
self
.
assertEqual
(
server
[
'flavor'
][
'id'
],
flavor
[
'id'
])
self
.
assertEqual
(
server
[
'image'
][
'id'
],
image
[
'id'
])
self
.
assertEqual
(
server
[
'status'
],
"BUILD"
)
return
server
def
_get_connection_username
(
self
,
server
):
"""Determine the username to use to connect to the server"""
users
=
server
[
'metadata'
].
get
(
"users"
,
None
)
ret_user
=
None
if
users
is
not
None
:
user_list
=
users
.
split
()
if
"root"
in
user_list
:
ret_user
=
"root"
else
:
ret_user
=
random
.
choice
(
user_list
)
else
:
# Return the login name for connections based on the server OS
self
.
info
(
"Could not find `users' metadata in server. Let's guess"
)
os_value
=
server
[
'metadata'
].
get
(
"os"
)
if
os_value
in
(
"Ubuntu"
,
"Kubuntu"
,
"Fedora"
):
ret_user
=
"user"
elif
os_value
in
(
"windows"
,
"windows_alpha1"
):
ret_user
=
"Administrator"
else
:
ret_user
=
"root"
self
.
assertIsNotNone
(
ret_user
)
self
.
info
(
"User's login name: %s"
,
ret_user
)
return
ret_user
def
_insist_on_server_transition
(
self
,
server
,
curr_status
,
new_status
):
"""Insist on server transiting from curr_status to new_status"""
def
check_fun
():
"""Check server status"""
srv
=
self
.
clients
.
cyclades
.
get_server_details
(
server
[
'id'
])
if
srv
[
'status'
]
==
curr_status
:
raise
Retry
()
elif
srv
[
'status'
]
==
new_status
:
return
else
:
msg
=
"Server %s went to unexpected status %s"
self
.
error
(
msg
,
server
[
'name'
],
srv
[
'status'
])
self
.
fail
(
msg
%
(
server
[
'name'
],
srv
[
'status'
]))
opmsg
=
"Waiting for server %s to transit from %s to %s"
self
.
info
(
opmsg
,
server
[
'name'
],
curr_status
,
new_status
)
opmsg
=
opmsg
%
(
server
[
'name'
],
curr_status
,
new_status
)
self
.
_try_until_timeout_expires
(
opmsg
,
check_fun
)
def
_insist_on_tcp_connection
(
self
,
family
,
host
,
port
):
"""Insist on tcp connection"""
def
check_fun
():
"""Get a connected socket from the specified family to host:port"""
sock
=
None
for
res
in
socket
.
getaddrinfo
(
host
,
port
,
family
,
socket
.
SOCK_STREAM
,
0
,
socket
.
AI_PASSIVE
):
fam
,
socktype
,
proto
,
_
,
saddr
=
res
try
:
sock
=
socket
.
socket
(
fam
,
socktype
,
proto
)
except
socket
.
error
:
sock
=
None
continue
try
:
sock
.
connect
(
saddr
)
except
socket
.
error
:
sock
.
close
()
sock
=
None
continue
if
sock
is
None
:
raise
Retry
return
sock
familystr
=
{
socket
.
AF_INET
:
"IPv4"
,
socket
.
AF_INET6
:
"IPv6"
,
socket
.
AF_UNSPEC
:
"Unspecified-IPv4/6"
}
opmsg
=
"Connecting over %s to %s:%s"
self
.
info
(
opmsg
,
familystr
.
get
(
family
,
"Unknown"
),
host
,
port
)
opmsg
=
opmsg
%
(
familystr
.
get
(
family
,
"Unknown"
),
host
,
port
)
return
self
.
_try_until_timeout_expires
(
opmsg
,
check_fun
)
def
_get_ip
(
self
,
server
,
version
):
"""Get the public IP of a server from the detailed server info"""
assert
version
in
(
4
,
6
)
nics
=
server
[
'attachments'
]
public_addrs
=
None
for
nic
in
nics
:
net_id
=
nic
[
'network_id'
]
if
self
.
clients
.
cyclades
.
get_network_details
(
net_id
)[
'public'
]:
public_addrs
=
nic
[
'ipv'
+
str
(
version
)]
self
.
assertIsNotNone
(
public_addrs
)
msg
=
"Servers %s public IPv%s is %s"
self
.
info
(
msg
,
server
[
'name'
],
version
,
public_addrs
)
return
public_addrs
def
_insist_on_ping
(
self
,
ip_addr
,
version
):
"""Test server responds to a single IPv4 of IPv6 ping"""
def
check_fun
():
"""Ping to server"""
assert
version
in
(
4
,
6
)
cmd
=
(
"ping%s -c 3 -w 20 %s"
%
(
"6"
if
version
==
6
else
""
,
ip_addr
))
ping
=
subprocess
.
Popen
(
cmd
,
shell
=
True
,
stdout
=
subprocess
.
PIPE
,
stderr
=
subprocess
.
PIPE
)
ping
.
communicate
()
ret
=
ping
.
wait
()
if
ret
!=
0
:
raise
Retry
opmsg
=
"Sent IPv%s ping requests to %s"
self
.
info
(
opmsg
,
version
,
ip_addr
)
opmsg
=
opmsg
%
(
version
,
ip_addr
)
self
.
_try_until_timeout_expires
(
opmsg
,
check_fun
)
# --------------------------------------------------------------------
# Initialize Burnin
...
...
@@ -412,6 +635,8 @@ def initialize(opts, testsuites):
BurninTests
.
action_warning
=
opts
.
action_warning
BurninTests
.
query_interval
=
opts
.
query_interval
BurninTests
.
system_user
=
opts
.
system_user
BurninTests
.
flavors
=
opts
.
flavors
BurninTests
.
images
=
opts
.
images
BurninTests
.
run_id
=
SNF_TEST_PREFIX
+
\
datetime
.
datetime
.
strftime
(
datetime
.
datetime
.
now
(),
"%Y%m%d%H%M%S"
)
...
...
@@ -427,16 +652,13 @@ def initialize(opts, testsuites):
# --------------------------------------------------------------------
# Run Burnin
def
run
(
testsuites
,
failfast
=
False
,
final_report
=
False
):
def
run
_burnin
(
testsuites
,
failfast
=
False
,
final_report
=
False
):
"""Run burnin testsuites"""
global
logger
# Using global. pylint: disable-msg=C0103,W0603,W0602
success
=
True
for
tcase
in
testsuites
:
tsuite
=
unittest
.
TestLoader
().
loadTestsFromTestCase
(
tcase
)
results
=
tsuite
.
run
(
BurninTestResult
())
was_success
=
was_successful
(
tcase
.
__name__
,
results
.
wasSuccessful
())
was_success
=
run_test
(
tcase
)
success
=
success
and
was_success
if
failfast
and
not
success
:
break
...
...
@@ -451,6 +673,14 @@ def run(testsuites, failfast=False, final_report=False):
return
0
if
success
else
1
def
run_test
(
tcase
):
"""Run a testcase"""
tsuite
=
unittest
.
TestLoader
().
loadTestsFromTestCase
(
tcase
)
results
=
tsuite
.
run
(
BurninTestResult
())
return
was_successful
(
tcase
.
__name__
,
results
.
wasSuccessful
())
# --------------------------------------------------------------------
# Helper functions
def
was_successful
(
tsuite
,
success
):
...
...
@@ -494,3 +724,11 @@ class Proper(object):
def
__set__
(
self
,
obj
,
value
):
self
.
val
=
value
class
Retry
(
Exception
):
"""Retry the action
This is used by _try_unit_timeout_expires method.
"""
snf-tools/synnefo_tools/burnin/cyclades_tests.py
deleted
100644 → 0
View file @
befc1d2c
# Copyright 2013 GRNET S.A. All rights reserved.
#
# Redistribution and use in source and binary forms, with or
# without modification, are permitted provided that the following
# conditions are met:
#
# 1. Redistributions of source code must retain the above
# copyright notice, this list of conditions and the following
# disclaimer.
#
# 2. Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following
# disclaimer in the documentation and/or other materials
# provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
# The views and conclusions contained in the software and
# documentation are those of the authors and should not be
# interpreted as representing official policies, either expressed
# or implied, of GRNET S.A.
"""
This is the burnin class that tests the Cyclades functionality
"""
from
synnefo_tools.burnin.common
import
BurninTests
,
Proper
# Too many public methods. pylint: disable-msg=R0904
class
FlavorsTestSuite
(
BurninTests
):
"""Test flavor lists for consistency"""
simple_flavors
=
Proper
(
value
=
None
)
detailed_flavors
=
Proper
(
value
=
None
)
simple_names
=
Proper
(
value
=
None
)
def
test_001_simple_flavors
(
self
):
"""Test flavor list actually returns flavors"""
self
.
simple_flavors
=
self
.
_get_list_of_flavors
(
detail
=
False
)
self
.
assertGreater
(
len
(
self
.
simple_flavors
),
0
)
def
test_002_get_detailed_flavors
(
self
):
"""Test detailed flavor list is the same length as list"""
self
.
detailed_flavors
=
self
.
_get_list_of_flavors
(
detail
=
True
)
self
.
assertEquals
(
len
(
self
.
simple_flavors
),
len
(
self
.
detailed_flavors
))
def
test_003_same_flavor_names
(
self
):
"""Test detailed and simple flavor list contain same names"""
names
=
sorted
([
flv
[
'name'
]
for
flv
in
self
.
simple_flavors
])
self
.
simple_names
=
names
detailed_names
=
sorted
([
flv
[
'name'
]
for
flv
in
self
.
detailed_flavors
])
self
.
assertEqual
(
self
.
simple_names
,
detailed_names
)
def
test_004_unique_flavor_names
(
self
):
"""Test flavors have unique names"""
self
.
assertEqual
(
sorted
(
list
(
set
(
self
.
simple_names
))),
self
.
simple_names
)
def
test_005_well_formed_names
(
self
):
"""Test flavors have well formed names
Test flavors have names of the form CxxRyyDzz, where xx is vCPU count,
yy is RAM in MiB, zz is Disk in GiB
"""
for
flv
in
self
.
detailed_flavors
:
flavor
=
(
flv
[
'vcpus'
],
flv
[
'ram'
],
flv
[
'disk'
],
flv
[
'SNF:disk_template'
])
self
.
assertEqual
(
"C%dR%dD%d%s"
%
flavor
,
flv
[
'name'
],
"Flavor %s doesn't match its specs"
%
flv
[
'name'
])
snf-tools/synnefo_tools/burnin/images_tests.py
View file @
b0a039fe
...
...
@@ -135,8 +135,10 @@ class ImagesTestSuite(BurninTests):
def
test_007_download_image
(
self
):
"""Download image from Pithos"""
# Find the 'Debian Base' image
image
=
self
.
_find_image
([
"name:^Debian Base$"
],
images
=
self
.
system_images
)
images
=
self
.
_find_images
([
"name:^Debian Base$"
],
images
=
self
.
system_images
)
image
=
images
[
0
]
self
.
info
(
"Will use %s with id %s"
,
image
[
'name'
],
image
[
'id'
])
image_location
=
\
image
[
'location'
].
replace
(
"://"
,
" "
).
replace
(
"/"
,
" "
).
split
()
image_owner
=
image_location
[
1
]
...
...
snf-tools/synnefo_tools/burnin/logger.py
View file @
b0a039fe
...
...
@@ -99,7 +99,10 @@ def _green(msg):
def
_format_message
(
msg
,
*
args
):
"""Format the message using the args"""
return
(
msg
%
args
)
+
"
\n
"
if
args
:
return
(
msg
%
args
)
+
"
\n
"
else
:
return
msg
+
"
\n
"
def
_list_to_string
(
lst
,
append
=
""
):
...
...
@@ -439,7 +442,10 @@ class Log(object):
"""
if
self
.
use_colors
:
if
callable
(
color_fun
):
return
color_fun
((
msg
%
args
))
+
"
\n
"
if
args
:
return
color_fun
((
msg
%
args
))
+
"
\n
"
else
:
return
color_fun
(
msg
)
+
"
\n
"
else
:
args
=
tuple
([
_blue
(
arg
)
for
arg
in
args
])
return
_format_message
(
msg
,
*
args
)
...
...
snf-tools/synnefo_tools/burnin/server_tests.py
0 → 100644
View file @
b0a039fe
# Copyright 2013 GRNET S.A. All rights reserved.
#
# Redistribution and use in source and binary forms, with or
# without modification, are permitted provided that the following
# conditions are met:
#
# 1. Redistributions of source code must retain the above
# copyright notice, this list of conditions and the following
# disclaimer.
#
# 2. Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following
# disclaimer in the documentation and/or other materials
# provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED