import atexit
from cStringIO import StringIO
-import datetime
-import iso8601
import os
import sys
import signal
import subprocess
-import subunit
+from samba import subunit
import traceback
import warnings
)
from selftest.client import write_clientconf
from selftest.run import (
- expand_environment_strings,
expand_command_list,
expand_command_run,
exported_envvars_str,
+ now,
+ run_testsuite_command,
)
from selftest.target import (
EnvironmentManager,
includes = ()
excludes = ()
-def now():
- return datetime.datetime.utcnow().replace(tzinfo=iso8601.iso8601.Utc())
-
def read_excludes(fn):
excludes.extend(testlist.read_test_regexes(fn))
parser.add_option("--include", action="callback", help="Add file to include files", callback=read_includes)
parser.add_option("--testenv", help="run a shell in the requested test environment", action="store_true", default=False)
parser.add_option("--resetup-environment", help="Re-setup environment", action="store_true", default=False)
-parser.add_option("--binary-mapping", help="Map binaries to use", type=str)
parser.add_option("--load-list", help="Load list of tests to load from a file", type=str)
parser.add_option("--prefix", help="prefix to run tests in", type=str, default="./st")
parser.add_option("--srcdir", type=str, default=".", help="source directory")
sys.stderr.write("Exiting early because of signal %s.\n" % sig)
sys.exit(1)
-for sig in (signal.SIGINT, signal.SIGQUIT, signal.SIGTERm, signal.SIGPIPE):
+for sig in (signal.SIGINT, signal.SIGQUIT, signal.SIGTERM, signal.SIGPIPE):
signal.signal(sig, handle_signal)
def skip(name):
os.unlink(pcap_file)
-def run_testsuite(envname, name, cmd):
+def run_testsuite(name, cmd, subunit_ops, env=None):
"""Run a single testsuite.
- :param envname: Name of the environment to ru nin
+ :param env: Environment to run in
:param name: Name of the testsuite
:param cmd: Name of the (fully expanded) command to run
:return: exitcode of the command
"""
pcap_file = setup_pcap(name)
- subunit_ops.start_testsuite(name)
- subunit_ops.progress(None, subunit.PROGRESS_PUSH)
- subunit_ops.time(now())
- try:
- exitcode = subprocess.call(cmd, shell=True)
- except Exception, e:
- subunit_ops.time(now())
- subunit_ops.progress(None, subunit.PROGRESS_POP)
- subunit_ops.end_testsuite(name, "error", "Unable to run %r: %s" % (cmd, e))
+ exitcode = run_testsuite_command(name, cmd, subunit_ops, env)
+ if exitcode is None:
sys.exit(1)
- subunit_ops.time(now())
- subunit_ops.progress(None, subunit.PROGRESS_POP)
-
- envlog = env_manager.getlog_env(envname)
- if envlog != "":
- sys.stdout.write("envlog: %s\n" % envlog)
-
- sys.stdout.write("command: %s\n" % cmd)
- sys.stdout.write("expanded command: %s\n" % expand_environment_strings(cmd, os.environ))
-
- if exitcode == 0:
- subunit_ops.end_testsuite(name, "success")
- else:
- subunit_ops.end_testsuite(name, "failure", "Exit code was %d" % exitcode)
-
cleanup_pcap(pcap_file, exitcode)
if not opts.socket_wrapper_keep_pcap and pcap_file is not None:
return exitcode
+
if opts.list and opts.testenv:
sys.stderr.write("--list and --testenv are mutually exclusive\n")
sys.exit(1)
# quick hack to disable rpc validation when using valgrind - it is way too slow
if not os.environ.get("VALGRIND"):
os.environ["VALIDATE"] = "validate"
- os.environ["MALLOC_CHECK_"] = "2"
+ os.environ["MALLOC_CHECK_"] = "3"
# make all our python scripts unbuffered
os.environ["PYTHONUNBUFFERED"] = "1"
prefix = os.path.normpath(opts.prefix)
-if prefix == "":
- raise Exception("using an empty prefix isn't allowed")
-
# Ensure we have the test prefix around.
#
# We need restrictive permissions on this as some subdirectories in this tree
srcdir_abs = os.path.abspath(opts.srcdir)
-if prefix_abs == "":
- raise Exception("using an empty absolute prefix isn't allowed")
if prefix_abs == "/":
- raise Exception("using '/' as absolute prefix isn't allowed")
+ raise Exception("using '/' as absolute prefix is a bad idea")
os.environ["PREFIX"] = prefix
os.environ["KRB5CCNAME"] = os.path.join(prefix, "krb5ticket")
testenv_default = "none"
-if opts.binary_mapping:
- binary_mapping = dict([l.split(":") for l in opts.binary_mapping.split(",")])
- os.environ["BINARY_MAPPING"] = opts.binary_mapping
-else:
- binary_mapping = {}
- os.environ["BINARY_MAPPING"] = ""
-
# After this many seconds, the server will self-terminate. All tests
# must terminate in this time, and testenv will only stay alive this
# long
-server_maxtime = 7500
if os.environ.get("SMBD_MAXTIME", ""):
server_maxtime = int(os.environ["SMBD_MAXTIME"])
+else:
+ server_maxtime = 7500
def has_socket_wrapper(bindir):
if opts.socket_wrapper and not has_socket_wrapper(opts.bindir):
sys.stderr.write("You must include --enable-socket-wrapper when compiling Samba in order to execute 'make test'. Exiting....\n")
sys.exit(1)
- testenv_default = "dc"
+ testenv_default = "ad_dc_ntvfs"
from selftest.target.samba import Samba
- target = Samba(opts.bindir, binary_mapping, ldap, opts.srcdir, server_maxtime)
+ target = Samba(opts.bindir, ldap, opts.srcdir, server_maxtime)
elif opts.target == "samba3":
if opts.socket_wrapper and not has_socket_wrapper(opts.bindir):
sys.stderr.write("You must include --enable-socket-wrapper when compiling Samba in order to execute 'make test'. Exiting....\n")
sys.exit(1)
testenv_default = "member"
from selftest.target.samba3 import Samba3
- target = Samba3(opts.bindir, binary_mapping, srcdir_abs, server_maxtime)
+ target = Samba3(opts.bindir, srcdir_abs, server_maxtime)
elif opts.target == "none":
testenv_default = "none"
target = NoneTarget()
"VAMPIRE_DC_NETBIOSNAME",
"VAMPIRE_DC_NETBIOSALIAS",
+ # domain controller stuff for Vampired DC
+ "PROMOTED_DC_SERVER",
+ "PROMOTED_DC_SERVER_IP",
+ "PROMOTED_DC_NETBIOSNAME",
+ "PROMOTED_DC_NETBIOSALIAS",
+
# server stuff
"SERVER",
"SERVER_IP",
elif name in os.environ:
del os.environ[name]
- return testenv_vars
+ return env
# This 'global' file needs to be empty when we start
dns_host_file_path = os.path.join(prefix_abs, "dns_host_file")
if opts.testenv:
testenv_name = os.environ.get("SELFTEST_TESTENV", testenv_default)
- testenv_vars = switch_env(testenv_name, prefix)
+ env = switch_env(testenv_name, prefix)
+ testenv_vars = env.get_vars()
os.environ["PIDDIR"] = testenv_vars["PIDDIR"]
os.environ["ENVNAME"] = testenv_name
else:
for (name, envname, cmd, supports_loadfile, supports_idlist, subtests) in todo:
try:
- envvars = switch_env(envname, prefix)
+ env = switch_env(envname, prefix)
except UnsupportedEnvironment:
subunit_ops.start_testsuite(name)
subunit_ops.end_testsuite(name, "skip",
cmd, tmpf = expand_command_run(cmd, supports_loadfile, supports_idlist,
subtests)
- run_testsuite(envname, name, cmd)
+ run_testsuite(name, cmd, subunit_ops, env=env)
if tmpf is not None:
os.remove(tmpf)
if opts.resetup_environment:
env_manager.teardown_env(envname)
+ env_manager.teardown_all()
sys.stdout.write("\n")
-if not opts.list:
- env_manager.teardown_all()
-
# if there were any valgrind failures, show them
for fn in os.listdir(prefix):
if fn.startswith("valgrind.log"):
git.samba.org / obnox / samba / samba-obnox.git / blobdiff