Page Menu
Home
ClusterLabs Projects
Search
Configure Global Search
Log In
Files
F2823049
boothtestenv.py
No One
Temporary
Actions
Download File
Edit File
Delete File
View Transforms
Subscribe
Mute Notifications
Flag For Later
Award Token
Size
2 KB
Referenced Files
None
Subscribers
None
boothtestenv.py
View Options
#!/usr/bin/python
import
os
import
subprocess
import
time
import
tempfile
import
unittest
from
assertions
import
BoothAssertions
from
boothrunner
import
BoothRunner
class
BoothTestEnvironment
(
unittest
.
TestCase
,
BoothAssertions
):
test_src_path
=
os
.
path
.
abspath
(
os
.
path
.
dirname
(
__file__
))
dist_path
=
os
.
path
.
join
(
test_src_path
,
'..'
)
src_path
=
os
.
path
.
join
(
dist_path
,
'src'
)
boothd_path
=
os
.
path
.
join
(
src_path
,
'boothd'
)
conf_path
=
os
.
path
.
join
(
dist_path
,
'conf'
)
example_config_path
=
os
.
path
.
join
(
conf_path
,
'booth.conf.example'
)
def
setUp
(
self
):
if
not
self
.
_testMethodName
.
startswith
(
'test_'
):
raise
RuntimeError
,
"unexpected test method name: "
+
self
.
_testMethodName
self
.
test_name
=
self
.
_testMethodName
[
5
:]
self
.
test_path
=
os
.
path
.
join
(
self
.
test_run_path
,
self
.
test_name
)
os
.
makedirs
(
self
.
test_path
)
self
.
ensure_boothd_not_running
()
def
ensure_boothd_not_running
(
self
):
# Need to redirect STDERR in case we're not root, in which
# case netstat's -p option causes a warning. However we only
# want to kill boothd processes which we own; -p will list the
# pid for those and only those, which is exactly what we want
# here.
subprocess
.
call
(
"netstat -tpln 2>&1 | perl -lne 'm,LISTEN\s+(\d+)/boothd, and kill 15, $1'"
,
shell
=
True
)
def
get_tempfile
(
self
,
identity
):
tf
=
tempfile
.
NamedTemporaryFile
(
prefix
=
'
%s
.
%d
.'
%
(
identity
,
time
.
time
()),
dir
=
self
.
test_path
,
delete
=
False
)
return
tf
.
name
def
init_log
(
self
):
self
.
log_file
=
self
.
get_tempfile
(
'log'
)
os
.
putenv
(
'HA_debugfile'
,
self
.
log_file
)
# See cluster-glue/lib/clplumbing/cl_log.c
def
read_log
(
self
):
if
not
os
.
path
.
exists
(
self
.
log_file
):
return
''
l
=
open
(
self
.
log_file
)
msgs
=
''
.
join
(
l
.
readlines
())
l
.
close
()
return
msgs
def
check_return_code
(
self
,
pid
,
return_code
,
expected_exitcode
):
if
return_code
is
None
:
print
"pid
%d
still running"
%
pid
if
expected_exitcode
is
not
None
:
self
.
fail
(
"expected exit code
%d
, not long-running process"
%
expected_exitcode
)
else
:
print
"pid
%d
exited with code
%d
"
%
(
pid
,
return_code
)
if
expected_exitcode
is
None
:
msg
=
"should not exit"
else
:
msg
=
"should exit with code
%s
"
%
expected_exitcode
msg
+=
"
\n
Log follows (see
%s
)"
%
self
.
log_file
msg
+=
"
\n
N.B. expect mlockall/setscheduler errors when running tests non-root"
msg
+=
"
\n
-----------
\n
%s
"
%
self
.
read_log
()
self
.
assertEqual
(
return_code
,
expected_exitcode
,
msg
)
File Metadata
Details
Attached
Mime Type
text/x-script.python
Expires
Sat, Jan 25, 7:07 AM (1 d, 19 h)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
1217476
Default Alt Text
boothtestenv.py (2 KB)
Attached To
Mode
rB Booth
Attached
Detach File
Event Timeline
Log In to Comment