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
163
164
165
166
167
168
169
170
171
172
173
|
from subprocess import Popen, PIPE
from testtools.content import text_content
from testtools.matchers import Contains
from functools import wraps
import platform
import errno
import os
import pwd
import re
import apt
from time import sleep, time
import signal
CLUSTER_CONTROLLER_IP = '192.168.20.5'
SQUID_DEB_PROXY_URL = 'http://10.98.3.6:8000'
# HTTP proxy used to access the external world.
HTTP_PROXY = "http://10.98.3.6:3128"
class TimeoutError(Exception):
pass
def change_logs_permissions(log_dest):
"""Change logs permissions so auto-package-testing can access them."""
ubuntu_user = pwd.getpwnam('ubuntu')
uid, gid = ubuntu_user.pw_uid, ubuntu_user.pw_gid
for root, dirs, files in os.walk(log_dest):
for d in dirs:
os.chown(os.path.join(root, d), uid, gid)
for f in files:
os.chown(os.path.join(root, f), uid, gid)
def is_juju_core():
"""Report if 'juju' is provider by juju-core."""
retcode, out, err = run_command(['juju', '--version'])
# pyjuju 0.6 has no --version option. pyjuju 0.7 returns "juju
# <version>". gojuju returns "<version>".
if out == '' or 'juju ' in out:
return False
return True
def is_dist(distname):
_, _, distid = platform.linux_distribution()
return distid == distname
def is_precise():
return is_dist('precise')
def is_saucy():
return is_dist('saucy')
# Timeout decorator from
# http://stackoverflow.com/questions/2281850/timeout-function-if-it-takes-too-long-to-finish
def timeout(seconds=10, error_message=os.strerror(errno.ETIME)):
def decorator(func):
def _handle_timeout(signum, frame):
raise TimeoutError(error_message)
def wrapper(*args, **kwargs):
signal.signal(signal.SIGALRM, _handle_timeout)
signal.alarm(seconds)
try:
result = func(*args, **kwargs)
finally:
signal.alarm(0)
return result
return wraps(func)(wrapper)
return decorator
def run_command(args, env=None):
"""A wrapper to Popen to run commands in the command-line."""
process = Popen(args, stdout=PIPE, stderr=PIPE, stdin=PIPE, env=env)
stdout, stderr = process.communicate()
return process.returncode, stdout, stderr
def update_pxe_config():
content = open('/etc/maas/import_pxe_files').read()
content = content.replace(
'#RELEASES="precise quantal raring saucy"',
'RELEASES="precise saucy"')
with open('/etc/maas/import_pxe_files', 'w') as f:
f.write(content)
def retries(timeout=30, delay=1):
"""Helper for retrying something, sleeping between attempts.
Yields ``(elapsed, remaining)`` tuples, giving times in seconds.
@param timeout: From now, how long to keep iterating, in seconds.
@param delay: The sleep between each iteration, in seconds.
"""
start = time()
end = start + timeout
for now in iter(time, None):
if now < end:
yield now - start, end - now
sleep(min(delay, end - now))
else:
break
def assertStartedUpstartService(runner, service_name, log_file=None):
"""Assert that the upstart service 'service_name' is running."""
if log_file is not None:
try:
log_file_content = file(log_file).read()
runner.addDetail(
"File %s content" % log_file,
text_content(log_file_content))
except IOError:
runner.addDetail(
"File %s content" % log_file,
text_content("could not be read"))
# The service might be running pre-start, retry a few times
# before giving up.
expected = service_name + " start/running"
for elapsed, remaining in retries(timeout=30, delay=5):
retcode, output, err = run_command(["service", service_name, "status"])
if expected in output:
break
runner.assertThat(output, Contains(expected))
def assertCommandReturnCode(
runner, cmd, expected_output, env=None, expected_retcode=0):
"""Assert that cmd returns appropriate return code."""
if env is None:
env = os.environ.copy()
retcode, output, err = run_command(cmd, env=env)
runner.addDetail('%s stdout' % cmd, text_content(str(output)))
runner.addDetail('%s stderr' % cmd, text_content(str(err)))
runner.assertThat(output, Contains(expected_output))
runner.assertIs(retcode, expected_retcode)
def get_maas_version():
"""Get the version of the installed MAAS package."""
cache = apt.Cache()
pkg = cache["maas"]
return pkg.installed.version
def get_maas_revision():
"""Get the upstream revision of the installed MAAS package.
This assumes that MAAS is installed."""
version = get_maas_version()
# Try to match all the known formats in sequence.
versions_re = [
# Daily ppa format (format 1).
'\d*\+bzr\d*(?:\+dfsg)?-0\+(\d*)\+',
# Daily ppa format (format 2).
'\d*\+bzr\d*(?:\+dfsg)?\+(\d*)\+',
# Release format.
'\d*\+bzr(\d*)(?:\+dfsg)?',
]
for version_re in versions_re:
match = re.search(version_re, version)
if match is not None:
return int(match.group(1))
return None
|