[Zope3-checkins] SVN: zdaemon/trunk/src/zdaemon/ finished creation of rundirectory
Christian Zagrodnick
cz at gocept.com
Wed Jan 28 11:48:12 EST 2009
Log message for revision 95349:
finished creation of rundirectory
Changed:
U zdaemon/trunk/src/zdaemon/tests/testzdoptions.py
U zdaemon/trunk/src/zdaemon/tests/testzdrun.py
U zdaemon/trunk/src/zdaemon/zdctl.py
U zdaemon/trunk/src/zdaemon/zdoptions.py
-=-
Modified: zdaemon/trunk/src/zdaemon/tests/testzdoptions.py
===================================================================
--- zdaemon/trunk/src/zdaemon/tests/testzdoptions.py 2009-01-28 16:39:33 UTC (rev 95348)
+++ zdaemon/trunk/src/zdaemon/tests/testzdoptions.py 2009-01-28 16:48:11 UTC (rev 95349)
@@ -24,7 +24,8 @@
import ZConfig
import zdaemon
from zdaemon.zdoptions import (
- ZDOptions, RunnerOptions, existing_parent_directory)
+ ZDOptions, RunnerOptions,
+ existing_parent_directory, existing_parent_dirpath)
class ZDOptionsTestBase(unittest.TestCase):
@@ -340,16 +341,23 @@
options = self.OptionsClass()
path = os.path.join(self.root, 'does-not-exist', 'really-not')
self.check_exit_code(options, ["-z", path])
+ socket = os.path.join(path, 'socket')
+ self.check_exit_code(options, ["-s", socket])
def test_existing_directory(self):
options = self.OptionsClass()
options.realize(["-z", self.root])
+ socket = os.path.join(self.root, 'socket')
+ self.check_exit_code(options, ["-s", socket])
def test_parent_is_created(self):
options = self.OptionsClass()
path = os.path.join(self.root, 'will-be-created')
options.realize(["-z", path])
self.assertEquals(path, options.directory)
+ socket = os.path.join(path, 'socket')
+ options = self.OptionsClass()
+ options.realize(["-s", socket])
# Directory will be created when zdaemon runs, not when the
# configuration is read
self.assertFalse(os.path.exists(path))
@@ -362,7 +370,16 @@
ValueError, existing_parent_directory,
os.path.join(self.root, 'not-there', 'this-also-not'))
+ def test_existing_parent_dirpath(self):
+ self.assertTrue(existing_parent_dirpath(
+ os.path.join(self.root, 'sock')))
+ self.assertTrue(existing_parent_dirpath(
+ os.path.join(self.root, 'not-there', 'sock')))
+ self.assertRaises(
+ ValueError, existing_parent_dirpath,
+ os.path.join(self.root, 'not-there', 'this-also-not', 'sock'))
+
def test_suite():
suite = unittest.TestSuite()
for cls in [TestBasicFunctionality,
Modified: zdaemon/trunk/src/zdaemon/tests/testzdrun.py
===================================================================
--- zdaemon/trunk/src/zdaemon/tests/testzdrun.py 2009-01-28 16:39:33 UTC (rev 95348)
+++ zdaemon/trunk/src/zdaemon/tests/testzdrun.py 2009-01-28 16:48:11 UTC (rev 95349)
@@ -315,39 +315,60 @@
super(TestRunnerDirectory, self).setUp()
self.root = tempfile.mkdtemp()
self.save_stdout = sys.stdout
- self.save_stderr = sys.stderr
+ self.save_stderr = sys.stdout
sys.stdout = StringIO()
sys.stderr = StringIO()
+ self.expect = ''
def tearDown(self):
shutil.rmtree(self.root)
+ got = sys.stdout.getvalue()
+ err = sys.stderr.getvalue()
sys.stdout = self.save_stdout
sys.stderr = self.save_stderr
+ if err:
+ print >>sys.stderr, err,
+ self.assertEqual(self.expect, got)
super(TestRunnerDirectory, self).tearDown()
- def run_ctl(self, path):
+ def run_ctl(self, opts):
true_cmd = "/bin/true"
if not os.path.exists(true_cmd):
true_cmd = "/usr/bin/true" # Mac OS X
options = zdctl.ZDCtlOptions()
- options.realize(['-z', path, '-p', 'sleep 1', 'fg'])
+ options.realize(opts + ['-p', 'sleep 1', 'fg'])
self.expect = 'sleep 1\n'
proc = zdctl.ZDCmd(options)
proc.onecmd(" ".join(options.args))
def testCtlRunDirectoryCreation(self):
path = os.path.join(self.root, 'rundir')
- self.run_ctl(path)
+ self.run_ctl(['-z', path])
self.assert_(os.path.exists(path))
def testCtlRunDirectoryCreationOnlyOne(self):
path = os.path.join(self.root, 'rundir', 'not-created')
- self.run_ctl(path)
+ self.assertRaises(SystemExit, self.run_ctl, ['-z', path])
self.assertFalse(os.path.exists(path))
- self.assertTrue(sys.stdout.getvalue().startswith(
- 'Error: invalid value for -z'))
+ got = sys.stderr.getvalue().strip()
+ sys.stderr = StringIO()
+ self.assertTrue(got.startswith('Error: invalid value for -z'))
+ def testCtlSocketDirectoryCreation(self):
+ path = os.path.join(self.root, 'rundir', 'sock')
+ self.run_ctl(['-s', path])
+ self.assert_(os.path.exists(os.path.dirname(path)))
+
+ def testCtlSocketDirectoryCreationOnlyOne(self):
+ path = os.path.join(self.root, 'rundir', 'not-created', 'sock')
+ self.assertRaises(SystemExit, self.run_ctl, ['-s', path])
+ self.assertFalse(os.path.exists(path))
+ got = sys.stderr.getvalue().strip()
+ sys.stderr = StringIO()
+ self.assertTrue(got.startswith('Error: invalid value for -s'))
+
+
def send_action(action, sockname):
"""Send an action to the zdrun server and return the response.
Modified: zdaemon/trunk/src/zdaemon/zdctl.py
===================================================================
--- zdaemon/trunk/src/zdaemon/zdctl.py 2009-01-28 16:39:33 UTC (rev 95348)
+++ zdaemon/trunk/src/zdaemon/zdctl.py 2009-01-28 16:48:11 UTC (rev 95349)
@@ -43,6 +43,7 @@
"""
import os
+import os.path
import re
import cmd
import sys
@@ -145,14 +146,24 @@
os.environ[k] = v
self.create_rundir()
+ self.create_socket_dir()
self.set_uid()
def create_rundir(self):
if self.options.directory is None:
return
- if os.path.isdir(self.options.directory):
+ self.create_directory(self.options.directory)
+
+ def create_socket_dir(self):
+ dir = os.path.dirname(self.options.sockname)
+ if not dir:
return
- os.mkdir(self.options.directory)
+ self.create_directory(dir)
+
+ def create_directory(self, directory):
+ if os.path.isdir(directory):
+ return
+ os.mkdir(directory)
uid = os.geteuid()
if uid == 0 and uid != self.options.uid:
# Change owner of directory to target
Modified: zdaemon/trunk/src/zdaemon/zdoptions.py
===================================================================
--- zdaemon/trunk/src/zdaemon/zdoptions.py 2009-01-28 16:39:33 UTC (rev 95348)
+++ zdaemon/trunk/src/zdaemon/zdoptions.py 2009-01-28 16:48:11 UTC (rev 95349)
@@ -355,7 +355,7 @@
self.add("forever", "runner.forever", "f", "forever",
flag=1, default=0)
self.add("sockname", "runner.socket_name", "s:", "socket-name=",
- ZConfig.datatypes.existing_dirpath, default="zdsock")
+ existing_parent_dirpath, default="zdsock")
self.add("exitcodes", "runner.exit_codes", "x:", "exit-codes=",
list_of_ints, default=[0, 2])
self.add("user", "runner.user", "u:", "user=")
@@ -409,6 +409,19 @@
raise ValueError('%s is not an existing directory' % arg)
+def existing_parent_dirpath(arg):
+ path = os.path.expanduser(arg)
+ dir = os.path.dirname(path)
+ if not dir:
+ # relative pathname with no directory component
+ return path
+ parent, tail = os.path.split(dir)
+ if os.path.isdir(parent):
+ return path
+ raise ValueError('The directory named as part of the path %s '
+ 'does not exist.' % arg)
+
+
def _test():
# Stupid test program
z = ZDOptions()
More information about the Zope3-Checkins
mailing list