[Zope3-checkins] CVS: Packages3/zdaemon/tests - testzdrun.py:1.1 testzdoptions.py:1.1 testDaemon.py:1.6 __init__.py:1.2
Jeremy Hylton
jeremy@zope.com
Thu, 19 Jun 2003 11:48:54 -0400
Update of /cvs-repository/Packages3/zdaemon/tests
In directory cvs.zope.org:/tmp/cvs-serv28675/tests
Modified Files:
__init__.py
Added Files:
testzdrun.py testzdoptions.py testDaemon.py
Log Message:
Merge the zdaemon code from Zope2.
=== Added File Packages3/zdaemon/tests/testzdrun.py ===
"""Test suite for zdrun.py."""
import os
import sys
import time
import signal
import tempfile
import unittest
from StringIO import StringIO
from zdaemon import zdrun, zdctl
class ZDaemonTests(unittest.TestCase):
python = os.path.abspath(sys.executable)
assert os.path.exists(python)
here = os.path.abspath(os.path.dirname(__file__))
assert os.path.isdir(here)
nokill = os.path.join(here, "nokill.py")
assert os.path.exists(nokill)
parent = os.path.dirname(here)
zdrun = os.path.join(parent, "zdrun.py")
assert os.path.exists(zdrun)
ppath = os.pathsep.join(sys.path)
def setUp(self):
self.zdsock = tempfile.mktemp()
self.new_stdout = StringIO()
self.save_stdout = sys.stdout
sys.stdout = self.new_stdout
self.expect = ""
def tearDown(self):
sys.stdout = self.save_stdout
for sig in (signal.SIGTERM,
signal.SIGHUP,
signal.SIGINT,
signal.SIGCHLD):
signal.signal(sig, signal.SIG_DFL)
try:
os.unlink(self.zdsock)
except os.error:
pass
output = self.new_stdout.getvalue()
self.assertEqual(self.expect, output)
def quoteargs(self, args):
for i in range(len(args)):
if " " in args[i]:
args[i] = '"%s"' % args[i]
return " ".join(args)
def rundaemon(self, args):
# Add quotes, in case some pathname contains spaces (e.g. Mac OS X)
args = self.quoteargs(args)
cmd = ('PYTHONPATH="%s" "%s" "%s" -d -s "%s" %s' %
(self.ppath, self.python, self.zdrun, self.zdsock, args))
os.system(cmd)
# When the daemon crashes, the following may help debug it:
##os.system("PYTHONPATH=%s %s %s -s %s %s &" %
## (self.ppath, self.python, self.zdrun, self.zdsock, args))
def run(self, args):
if type(args) is type(""):
args = args.split()
try:
zdctl.main(["-s", self.zdsock] + args)
except SystemExit:
pass
def testSystem(self):
self.rundaemon(["echo", "-n"])
self.expect = ""
## def testInvoke(self):
## self.run("echo -n")
## self.expect = ""
## def testControl(self):
## self.rundaemon(["sleep", "1000"])
## time.sleep(1)
## self.run("stop")
## time.sleep(1)
## self.run("exit")
## self.expect = "Sent SIGTERM\nExiting now\n"
## def testStop(self):
## self.rundaemon([self.python, self.nokill])
## time.sleep(1)
## self.run("stop")
## time.sleep(1)
## self.run("exit")
## self.expect = "Sent SIGTERM\nSent SIGTERM; will exit later\n"
def testHelp(self):
self.run("-h")
import __main__
self.expect = __main__.__doc__
def testOptionsSysArgv(self):
# Check that options are parsed from sys.argv by default
options = zdrun.ZDRunOptions()
save_sys_argv = sys.argv
try:
sys.argv = ["A", "B", "C"]
options.realize()
finally:
sys.argv = save_sys_argv
self.assertEqual(options.options, [])
self.assertEqual(options.args, ["B", "C"])
def testOptionsBasic(self):
# Check basic option parsing
options = zdrun.ZDRunOptions()
options.realize(["B", "C"], "foo")
self.assertEqual(options.options, [])
self.assertEqual(options.args, ["B", "C"])
self.assertEqual(options.progname, "foo")
def testOptionsHelp(self):
# Check that -h behaves properly
options = zdrun.ZDRunOptions()
try:
options.realize(["-h"], doc=zdrun.__doc__)
except SystemExit, err:
self.failIf(err.code)
else:
self.fail("SystemExit expected")
self.expect = zdrun.__doc__
def testSubprocessBasic(self):
# Check basic subprocess management: spawn, kill, wait
options = zdrun.ZDRunOptions()
options.realize(["sleep", "100"])
proc = zdrun.Subprocess(options)
self.assertEqual(proc.pid, 0)
pid = proc.spawn()
self.assertEqual(proc.pid, pid)
msg = proc.kill(signal.SIGTERM)
self.assertEqual(msg, None)
wpid, wsts = os.waitpid(pid, 0)
self.assertEqual(wpid, pid)
self.assertEqual(os.WIFSIGNALED(wsts), 1)
self.assertEqual(os.WTERMSIG(wsts), signal.SIGTERM)
proc.setstatus(wsts)
self.assertEqual(proc.pid, 0)
def test_suite():
suite = unittest.TestSuite()
if os.name == "posix":
suite.addTest(unittest.makeSuite(ZDaemonTests))
return suite
if __name__ == '__main__':
__file__ = sys.argv[0]
unittest.main(defaultTest='test_suite')
=== Added File Packages3/zdaemon/tests/testzdoptions.py ===
##############################################################################
#
# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE
#
##############################################################################
"""Test suite for zdaemon.zdoptions."""
import os
import sys
import tempfile
import unittest
from StringIO import StringIO
import ZConfig
import zdaemon
from zdaemon.zdoptions import ZDOptions
class ZDOptionsTestBase(unittest.TestCase):
OptionsClass = ZDOptions
def save_streams(self):
self.save_stdout = sys.stdout
self.save_stderr = sys.stderr
sys.stdout = self.stdout = StringIO()
sys.stderr = self.stderr = StringIO()
def restore_streams(self):
sys.stdout = self.save_stdout
sys.stderr = self.save_stderr
def check_exit_code(self, options, args):
save_sys_stderr = sys.stderr
try:
sys.stderr = StringIO()
try:
options.realize(args)
except SystemExit, err:
self.assertEqual(err.code, 2)
else:
self.fail("SystemExit expected")
finally:
sys.stderr = save_sys_stderr
class TestZDOptions(ZDOptionsTestBase):
input_args = ["arg1", "arg2"]
output_opts = []
output_args = ["arg1", "arg2"]
def test_basic(self):
progname = "progname"
doc = "doc"
options = self.OptionsClass()
options.positional_args_allowed = 1
options.schemadir = os.path.dirname(zdaemon.__file__)
options.realize(self.input_args, progname, doc)
self.assertEqual(options.progname, "progname")
self.assertEqual(options.doc, "doc")
self.assertEqual(options.options, self.output_opts)
self.assertEqual(options.args, self.output_args)
def test_configure(self):
configfile = os.path.join(os.path.dirname(zdaemon.__file__),
"sample.conf")
for arg in "-C", "--c", "--configure":
options = self.OptionsClass()
options.realize([arg, configfile])
self.assertEqual(options.configfile, configfile)
def test_help(self):
for arg in "-h", "--h", "--help":
options = self.OptionsClass()
try:
self.save_streams()
try:
options.realize([arg])
finally:
self.restore_streams()
except SystemExit, err:
self.assertEqual(err.code, 0)
else:
self.fail("%s didn't call sys.exit()" % repr(arg))
def test_unrecognized(self):
# Check that we get an error for an unrecognized option
self.check_exit_code(self.OptionsClass(), ["-/"])
class TestBasicFunctionality(TestZDOptions):
def test_no_positional_args(self):
# Check that we get an error for positional args when they
# haven't been enabled.
self.check_exit_code(self.OptionsClass(), ["A"])
def test_positional_args(self):
options = self.OptionsClass()
options.positional_args_allowed = 1
options.realize(["A", "B"])
self.assertEqual(options.args, ["A", "B"])
def test_positional_args_empty(self):
options = self.OptionsClass()
options.positional_args_allowed = 1
options.realize([])
self.assertEqual(options.args, [])
def test_positional_args_unknown_option(self):
# Make sure an unknown option doesn't become a positional arg.
options = self.OptionsClass()
options.positional_args_allowed = 1
self.check_exit_code(options, ["-o", "A", "B"])
def test_conflicting_flags(self):
# Check that we get an error for flags which compete over the
# same option setting.
options = self.OptionsClass()
options.add("setting", None, "a", flag=1)
options.add("setting", None, "b", flag=2)
self.check_exit_code(options, ["-a", "-b"])
def test_handler_simple(self):
# Test that a handler is called; use one that doesn't return None.
options = self.OptionsClass()
options.add("setting", None, "a:", handler=int)
options.realize(["-a2"])
self.assertEqual(options.setting, 2)
def test_handler_side_effect(self):
# Test that a handler is called and conflicts are not
# signalled when it returns None.
options = self.OptionsClass()
L = []
options.add("setting", None, "a:", "append=", handler=L.append)
options.realize(["-a2", "--append", "3"])
self.assert_(options.setting is None)
self.assertEqual(L, ["2", "3"])
def test_handler_with_bad_value(self):
options = self.OptionsClass()
options.add("setting", None, "a:", handler=int)
self.check_exit_code(options, ["-afoo"])
class EnvironmentOptions(ZDOptionsTestBase):
saved_schema = None
class OptionsClass(ZDOptions):
def __init__(self):
ZDOptions.__init__(self)
self.add("opt", "opt", "o:", "opt=",
default=42, handler=int, env="OPT")
def load_schema(self):
# Doing this here avoids needing a separate file for the schema:
if self.schema is None:
if EnvironmentOptions.saved_schema is None:
schema = ZConfig.loadSchemaFile(StringIO("""\
<schema>
<key name='opt' datatype='integer' default='12'/>
</schema>
"""))
EnvironmentOptions.saved_schema = schema
self.schema = EnvironmentOptions.saved_schema
def load_configfile(self):
if getattr(self, "configtext", None):
self.configfile = tempfile.mktemp()
f = open(self.configfile, 'w')
f.write(self.configtext)
f.close()
try:
ZDOptions.load_configfile(self)
finally:
os.unlink(self.configfile)
else:
ZDOptions.load_configfile(self)
# Save and restore the environment around each test:
def setUp(self):
self._oldenv = os.environ
env = {}
for k, v in os.environ.items():
env[k] = v
os.environ = env
def tearDown(self):
os.environ = self._oldenv
def create_with_config(self, text):
options = self.OptionsClass()
zdpkgdir = os.path.dirname(os.path.abspath(zdaemon.__file__))
options.schemadir = os.path.join(zdpkgdir, 'tests')
options.schemafile = "envtest.xml"
# configfile must be set for ZDOptions to use ZConfig:
if text:
options.configfile = "not used"
options.configtext = text
return options
class TestZDOptionsEnvironment(EnvironmentOptions):
def test_with_environment(self):
os.environ["OPT"] = "2"
self.check_from_command_line()
options = self.OptionsClass()
options.realize([])
self.assertEqual(options.opt, 2)
def test_without_environment(self):
self.check_from_command_line()
options = self.OptionsClass()
options.realize([])
self.assertEqual(options.opt, 42)
def check_from_command_line(self):
for args in (["-o1"], ["--opt", "1"]):
options = self.OptionsClass()
options.realize(args)
self.assertEqual(options.opt, 1)
def test_with_bad_environment(self):
os.environ["OPT"] = "Spooge!"
# make sure the bad value is ignored if the command-line is used:
self.check_from_command_line()
options = self.OptionsClass()
try:
self.save_streams()
try:
options.realize([])
finally:
self.restore_streams()
except SystemExit, e:
self.assertEqual(e.code, 2)
else:
self.fail("expected SystemExit")
def test_environment_overrides_configfile(self):
options = self.create_with_config("opt 3")
options.realize([])
self.assertEqual(options.opt, 3)
os.environ["OPT"] = "2"
options = self.create_with_config("opt 3")
options.realize([])
self.assertEqual(options.opt, 2)
class TestCommandLineOverrides(EnvironmentOptions):
def test_simple_override(self):
options = self.create_with_config("# empty config")
options.realize(["-X", "opt=-2"])
self.assertEqual(options.opt, -2)
def test_error_propogation(self):
self.check_exit_code(self.create_with_config("# empty"),
["-Xopt=1", "-Xopt=2"])
self.check_exit_code(self.create_with_config("# empty"),
["-Xunknown=foo"])
def test_suite():
suite = unittest.TestSuite()
for cls in [TestBasicFunctionality,
TestZDOptionsEnvironment,
TestCommandLineOverrides]:
suite.addTest(unittest.makeSuite(cls))
return suite
if __name__ == "__main__":
unittest.main(defaultTest='test_suite')
=== Packages3/zdaemon/tests/testDaemon.py 1.5 => 1.6 ===
--- /dev/null Thu Jun 19 11:48:54 2003
+++ Packages3/zdaemon/tests/testDaemon.py Thu Jun 19 11:48:54 2003
@@ -0,0 +1,68 @@
+# This file contains unit tests and a simple script that is explicitly
+# invoked by the tests. The script block goes first so that I don't
+# have to bother getting the imports right.
+
+# The script just kills itself or exits in a variety of ways.
+import os
+import sys
+
+if __name__ == "__main__":
+
+ arg = sys.argv[1]
+ if arg == "signal":
+ import signal
+ os.kill(os.getpid(), signal.SIGKILL)
+ elif arg == "exit":
+ os._exit(2)
+ os._exit(0)
+
+# The rest is unittest stuff that can be run by testrunner.py.
+
+import unittest
+import zdaemon
+import zdaemon.tests
+import zdaemon.Daemon
+
+class TestDoneError(RuntimeError):
+ pass
+
+class DaemonTest(unittest.TestCase):
+
+ dir, file = os.path.split(zdaemon.tests.__file__)
+ script = os.path.join(dir, "testDaemon.py")
+
+ def setUp(self):
+ os.environ["Z_DEBUG_MODE"] = ""
+ if os.environ.has_key("ZDAEMON_MANAGED"):
+ del os.environ["ZDAEMON_MANAGED"]
+
+ def tearDown(self):
+ pass
+
+ def run(self, arg):
+ try:
+ zdaemon.Daemon.run((self.script, arg))
+ except SystemExit:
+ pass
+
+ def testDeathBySignal(self):
+ try:
+ self.run("signal")
+ except TestDoneError, msg:
+ self.assert_(str(msg).find("terminated by signal") != -1)
+ else:
+ self.fail("signal was not caught")
+
+ def testDeathByExit(self):
+ try:
+ self.run("exit")
+ except TestDoneError, msg:
+ self.assert_(str(msg).find("exit status") != -1)
+ else:
+ self.fail("exit didn't raise an exception")
+
+def test_suite():
+ if hasattr(os, 'setsid'):
+ return unittest.makeSuite(DaemonTest)
+ else:
+ return None
=== Packages3/zdaemon/tests/__init__.py 1.1 => 1.2 ===
--- Packages3/zdaemon/tests/__init__.py:1.1 Mon Apr 22 18:09:27 2002
+++ Packages3/zdaemon/tests/__init__.py Thu Jun 19 11:48:54 2003
@@ -0,0 +1 @@
+# This file is needed to make this a package.