12825577578

Committed 17 Jan 2025 08:41AM UTC coverage: 79.914% (-0.08%) from 79.992%

Build # 12825577578

Build Type

Pull #37

github

Committed by

web-flow

Commit Message

Trigger CI

Pull Request Pull Request #37: Update Python version support.

Run Details

361 of 512 branches covered (70.51%)

Branch coverage included in aggregate %.

5 of 15 new or added lines in 5 files covered. (33.33%)

5 existing lines in 4 files now uncovered.

1676 of 2037 relevant lines covered (82.28%)

0.82 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.23

/src/zdaemon/zdoptions.py

##############################################################################
#
# Copyright (c) 2003 Zope Foundation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (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.
#
##############################################################################
"""Option processing for zdaemon and related code."""

import getopt
import os
import signal
import sys

import pkg_resources

import ZConfig


class ZDOptions:
    """a zdaemon script.

    Usage: python <script>.py [-C URL] [zdrun-options] [action [arguments]]

    Options:
    -C/--configure URL -- configuration file or URL
    -h/--help -- print usage message and exit
    --version -- print zdaemon version and exit

    Actions are commands like "start", "stop" and "status".  If -i is
    specified or no action is specified on the command line, a "shell"
    interpreting actions typed interactively is started (unless the
    configuration option default_to_interactive is set to false).  Use the
    action "help" to find out about available actions.
    """

    doc = None
    progname = None
    configfile = None
    schemadir = None
    schemafile = "schema.xml"
    schema = None
    confighandlers = None
    configroot = None

    # Class variable to control automatic processing of an <eventlog>
    # section.  This should be the (possibly dotted) name of something
    # accessible from configroot, typically "eventlog".
    logsectionname = None
    config_logger = None  # The configured event logger, if any

    # Class variable deciding whether positional arguments are allowed.
    # If you want positional arguments, set this to 1 in your subclass.
    positional_args_allowed = 0

    def __init__(self):
        self.names_list = []
        self.short_options = []
        self.long_options = []
        self.options_map = {}
        self.default_map = {}
        self.required_map = {}
        self.environ_map = {}
        self.zconfig_options = []
        self.version = pkg_resources.get_distribution("zdaemon").version
        self.add(None, None, "h", "help", self.help)
        self.add(None, None, None, "version", self.print_version)
        self.add("configfile", None, "C:", "configure=")
        self.add(None, None, "X:", handler=self.zconfig_options.append)

    def print_version(self, dummy):
        """Print zdaemon version number to stdout and exit(0)."""
        print(self.version)
        sys.exit(0)

    def help(self, dummy):
        """Print a long help message (self.doc) to stdout and exit(0).

        Occurrences of "%s" in self.doc are replaced by self.progname.
        """
        doc = self.doc
        if not doc:
            doc = "No help available."
        elif doc.find("%s") > 0:
            doc = doc.replace("%s", self.progname)
        print(doc, end='')
        sys.exit(0)

    def usage(self, msg):
        """Print a brief error message to stderr and exit(2)."""
        sys.stderr.write("Error: %s\n" % str(msg))
        sys.stderr.write("For help, use %s -h\n" % self.progname)
        sys.exit(2)

    def remove(self,
               name=None,               # attribute name on self
               confname=None,           # name in ZConfig (may be dotted)
               short=None,              # short option name
               long=None,               # long option name
               ):
        """Remove all traces of name, confname, short and/or long."""
        if name:
            for n, cn in self.names_list[:]:
                if n == name:
                    self.names_list.remove((n, cn))
            if name in self.default_map:
                del self.default_map[name]
            if name in self.required_map:
                del self.required_map[name]
        if confname:
            for n, cn in self.names_list[:]:
                if cn == confname:
                    self.names_list.remove((n, cn))
        if short:
            key = "-" + short[0]
            if key in self.options_map:
                del self.options_map[key]
        if long:
            key = "--" + long
            if key[-1] == "=":
                key = key[:-1]
            if key in self.options_map:
                del self.options_map[key]

    def add(self,
            name=None,                  # attribute name on self
            confname=None,              # name in ZConfig (may be dotted)
            short=None,                 # short option name
            long=None,                  # long option name
            handler=None,               # handler (defaults to string)
            default=None,               # default value
            required=None,              # message if not provided
            flag=None,                  # if not None, flag value
            env=None,                   # if not None, environment variable
            ):
        """Add information about a configuration option.

        This can take several forms:

        add(name, confname)
            Configuration option 'confname' maps to attribute 'name'
        add(name, None, short, long)
            Command line option '-short' or '--long' maps to 'name'
        add(None, None, short, long, handler)
            Command line option calls handler
        add(name, None, short, long, handler)
            Assign handler return value to attribute 'name'

        In addition, one of the following keyword arguments may be given:

        default=...  -- if not None, the default value
        required=... -- if nonempty, an error message if no value provided
        flag=...     -- if not None, flag value for command line option
        env=...      -- if not None, name of environment variable that
                        overrides the configuration file or default
        """

        if flag is not None:
            if handler is not None:
                raise ValueError("use at most one of flag= and handler=")
            if not long and not short:
                raise ValueError("flag= requires a command line flag")
            if short and short.endswith(":"):
                raise ValueError("flag= requires a command line flag")
            if long and long.endswith("="):
                raise ValueError("flag= requires a command line flag")

            def handler(arg, flag=flag):
                return flag

        if short and long:
            if short.endswith(":") != long.endswith("="):
                raise ValueError(
                    f"inconsistent short/long options: {short!r} {long!r}")

        if short:
            if short[0] == "-":
                raise ValueError("short option should not start with '-'")
            key, rest = short[:1], short[1:]
            if rest not in ("", ":"):
                raise ValueError("short option should be 'x' or 'x:'")
            key = "-" + key
            if key in self.options_map:
                raise ValueError("duplicate short option key '%s'" % key)
            self.options_map[key] = (name, handler)
            self.short_options.append(short)

        if long:
            if long[0] == "-":
                raise ValueError("long option should not start with '-'")
            key = long
            if key[-1] == "=":
                key = key[:-1]
            key = "--" + key
            if key in self.options_map:
                raise ValueError("duplicate long option key '%s'" % key)
            self.options_map[key] = (name, handler)
            self.long_options.append(long)

        if env:
            self.environ_map[env] = (name, handler)

        if name:
            if not hasattr(self, name):
                setattr(self, name, None)
            self.names_list.append((name, confname))
            if default is not None:
                self.default_map[name] = default
            if required:
                self.required_map[name] = required

    def realize(self, args=None, progname=None, doc=None,
                raise_getopt_errs=True):
        """Realize a configuration.

        Optional arguments:

        args     -- the command line arguments, less the program name
                    (default is sys.argv[1:])

        progname -- the program name (default is sys.argv[0])

        doc      -- usage message (default is __doc__ of the options class)
        """

        # Provide dynamic default method arguments
        if args is None:
            args = sys.argv[1:]

        if progname is None:
            progname = sys.argv[0]

        self.progname = progname
        self.doc = doc or self.__doc__

        self.options = []
        self.args = []

        # Call getopt
        try:
            self.options, self.args = getopt.getopt(
                args, "".join(self.short_options), self.long_options)
        except getopt.error as msg:
            if raise_getopt_errs:
                self.usage(msg)

        # Check for positional args
        if self.args and not self.positional_args_allowed:
            self.usage("positional arguments are not supported")

        # Process options returned by getopt
        for opt, arg in self.options:
            name, handler = self.options_map[opt]
            if handler is not None:
                try:
                    arg = handler(arg)
                except ValueError as msg:
                    self.usage(f"invalid value for {opt} {arg!r}: {msg}")
            if name and arg is not None:
                if getattr(self, name) is not None:
                    if getattr(self, name) == arg:
                        # Repeated option, but we don't mind because it
                        # just reinforces what we have.
                        continue
                    self.usage("conflicting command line option %r" % opt)
                setattr(self, name, arg)

        # Process environment variables
        for envvar in self.environ_map.keys():
            name, handler = self.environ_map[envvar]
            if name and getattr(self, name, None) is not None:
                continue
            if envvar in os.environ:
                value = os.environ[envvar]
                if handler is not None:
                    try:
                        value = handler(value)
                    except ValueError as msg:
                        self.usage("invalid environment value for %s %r: %s"
                                   % (envvar, value, msg))
                if name and value is not None:
                    setattr(self, name, value)

        if self.configfile is None:
            self.configfile = self.default_configfile()
        if self.zconfig_options and self.configfile is None:
            self.usage("configuration overrides (-X) cannot be used"
                       " without a configuration file")
        if self.configfile is not None:
            # Process config file
            self.load_schema()
            try:
                self.load_configfile()
            except ZConfig.ConfigurationError as msg:
                self.usage(str(msg))

        # Copy config options to attributes of self.  This only fills
        # in options that aren't already set from the command line.
        for name, confname in self.names_list:
            if confname and getattr(self, name) is None:
                parts = confname.split(".")
                obj = self.configroot
                for part in parts:
                    if obj is None:
                        break
                    # Here AttributeError is not a user error!
                    obj = getattr(obj, part)
                setattr(self, name, obj)

        # Process defaults
        for name, value in self.default_map.items():
            if getattr(self, name) is None:
                setattr(self, name, value)

        # Process required options
        for name, message in self.required_map.items():
            if getattr(self, name) is None:
                self.usage(message)

        if self.logsectionname:
            self.load_logconf(self.logsectionname)

    def default_configfile(self):
        """Return the name of the default config file, or None."""
        # This allows a default configuration file to be used without
        # affecting the -C command line option; setting self.configfile
        # before calling realize() makes the -C option unusable since
        # then realize() thinks it has already seen the option.  If no
        # -C is used, realize() will call this method to try to locate
        # a configuration file.
        return None

    def load_schema(self):
        if self.schema is None:
            # Load schema
            if self.schemadir is None:
                self.schemadir = os.path.dirname(__file__)
            self.schemafile = os.path.join(self.schemadir, self.schemafile)
            self.schema = ZConfig.loadSchema(self.schemafile)

    def load_configfile(self):
        self.configroot, self.confighandlers = \
            ZConfig.loadConfig(self.schema, self.configfile,
                               self.zconfig_options)

    def load_logconf(self, sectname="eventlog"):
        parts = sectname.split(".")
        obj = self.configroot
        for p in parts:
            if obj is None:
                break
            obj = getattr(obj, p)
        self.config_logger = obj
        if obj is not None:
            obj.startup()


class RunnerOptions(ZDOptions):

    uid = gid = None

    def __init__(self):
        ZDOptions.__init__(self)
        self.add("backofflimit", "runner.backoff_limit",
                 "b:", "backoff-limit=", int, default=10)
        self.add("daemon", "runner.daemon", "d", "daemon", flag=1, default=1)
        self.add("forever", "runner.forever", "f", "forever",
                 flag=1, default=0)
        self.add("sockname", "runner.socket_name", "s:", "socket-name=",
                 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=")
        self.add("umask", "runner.umask", "m:", "umask=", octal_type,
                 default=0o22)
        self.add("directory", "runner.directory", "z:", "directory=",
                 existing_parent_directory)
        self.add("transcript", "runner.transcript", "t:", "transcript=",
                 default="/dev/null")


# ZConfig datatype

def list_of_ints(arg):
    if not arg:
        return []
    else:
        return list(map(int, arg.split(",")))


def octal_type(arg):
    return int(arg, 8)


def name2signal(string):
    """Converts a signal name to canonical form.

    Signal names are recognized without regard for case:

      >>> name2signal('sighup')
      'SIGHUP'
      >>> name2signal('SigHup')
      'SIGHUP'
      >>> name2signal('SIGHUP')
      'SIGHUP'

    The leading 'SIG' is not required::

      >>> name2signal('hup')
      'SIGHUP'
      >>> name2signal('HUP')
      'SIGHUP'

    Names that are not known cause an exception to be raised::

      >>> name2signal('woohoo')
      Traceback (most recent call last):
      ValueError: could not convert 'woohoo' to signal name

      >>> name2signal('sigwoohoo')
      Traceback (most recent call last):
      ValueError: could not convert 'sigwoohoo' to signal name

    Numeric values are accepted to names as well::

      >>> name2signal(str(signal.SIGHUP))
      'SIGHUP'

    Numeric values that can't be matched to any signal known to Python
    are treated as errors::

      >>> name2signal('-234')
      Traceback (most recent call last):
      ValueError: unsupported signal on this platform: -234

      >>> name2signal(str(signal.NSIG))  #doctest: +ELLIPSIS
      Traceback (most recent call last):
      ValueError: unsupported signal on this platform: ...

    Non-signal attributes of the signal module are not mistakenly
    converted::

      >>> name2signal('_ign')
      Traceback (most recent call last):
      ValueError: could not convert '_ign' to signal name

      >>> name2signal('_DFL')
      Traceback (most recent call last):
      ValueError: could not convert '_DFL' to signal name

      >>> name2signal('sig_ign')
      Traceback (most recent call last):
      ValueError: could not convert 'sig_ign' to signal name

      >>> name2signal('SIG_DFL')
      Traceback (most recent call last):
      ValueError: could not convert 'SIG_DFL' to signal name

      >>> name2signal('getsignal')
      Traceback (most recent call last):
      ValueError: could not convert 'getsignal' to signal name

    """
    try:
        v = int(string)
    except ValueError:
        if "_" in string:
            raise ValueError("could not convert %r to signal name" % string)
        if string.startswith('Signals.'):  # py35 signals are an enum type
            string = string[len('Signals.'):]
        s = string.upper()
        if not s.startswith("SIG"):
            s = "SIG" + s
        v = getattr(signal, s, None)
        if isinstance(v, int):
            return s
        raise ValueError("could not convert %r to signal name" % string)
    if v >= signal.NSIG:
        raise ValueError("unsupported signal on this platform: %s" % string)
    for name in dir(signal):
        if "_" in name:
            continue
        if getattr(signal, name) == v:
            return name
    raise ValueError("unsupported signal on this platform: %s" % string)


def existing_parent_directory(arg):
    path = os.path.expanduser(arg)
    if os.path.isdir(path):
        # If the directory exists, that's fine.
        return path
    parent, tail = os.path.split(path)
    if os.path.isdir(parent):
        return path
    raise ValueError('%s is not an existing directory' % arg)


def existing_parent_dirpath(arg):
    path = os.path.expanduser(arg)
    dir = os.path.dirname(path)
    parent, tail = os.path.split(dir)
    if not parent:
        # relative pathname
        return path
    if os.path.isdir(parent):
        return path
    raise ValueError('The directory named as part of the path %s '
                     'does not exist.' % arg)


def _test():  # pragma: nocover
    # Stupid test program
    z = ZDOptions()
    z.add("program", "zdctl.program", "p:", "program=")
    print(z.names_list)
    z.realize()
    names = sorted(z.names_list[:])
    for name, confname in names:
        print("%-20s = %.56r" % (name, getattr(z, name)))


if __name__ == '__main__':
    __file__ = sys.argv[0]
    _test()

1	##############################################################################
2	#
3	# Copyright (c) 2003 Zope Foundation and Contributors.
4	# All Rights Reserved.
5	#
6	# This software is subject to the provisions of the Zope Public License,
7	# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
8	# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
9	# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
10	# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
11	# FOR A PARTICULAR PURPOSE.
12	#
13	##############################################################################
14	"""Option processing for zdaemon and related code."""
15
16	import getopt	1✔
17	import os	1✔
18	import signal	1✔
19	import sys	1✔
20
21	import pkg_resources	1✔
22
23	import ZConfig	1✔
24
25
26	class ZDOptions:	1✔
27	"""a zdaemon script.
28
29	Usage: python <script>.py [-C URL] [zdrun-options] [action [arguments]]
30
31	Options:
32	-C/--configure URL -- configuration file or URL
33	-h/--help -- print usage message and exit
34	--version -- print zdaemon version and exit
35
36	Actions are commands like "start", "stop" and "status". If -i is
37	specified or no action is specified on the command line, a "shell"
38	interpreting actions typed interactively is started (unless the
39	configuration option default_to_interactive is set to false). Use the
40	action "help" to find out about available actions.
41	"""
42
43	doc = None	1✔
44	progname = None	1✔
45	configfile = None	1✔
46	schemadir = None	1✔
47	schemafile = "schema.xml"	1✔
48	schema = None	1✔
49	confighandlers = None	1✔
50	configroot = None	1✔
51
52	# Class variable to control automatic processing of an <eventlog>
53	# section. This should be the (possibly dotted) name of something
54	# accessible from configroot, typically "eventlog".
55	logsectionname = None	1✔
56	config_logger = None # The configured event logger, if any	1✔
57
58	# Class variable deciding whether positional arguments are allowed.
59	# If you want positional arguments, set this to 1 in your subclass.
60	positional_args_allowed = 0	1✔
61
62	def __init__(self):	1✔
63	self.names_list = []	1✔
64	self.short_options = []	1✔
65	self.long_options = []	1✔
66	self.options_map = {}	1✔
67	self.default_map = {}	1✔
68	self.required_map = {}	1✔
69	self.environ_map = {}	1✔
70	self.zconfig_options = []	1✔
71	self.version = pkg_resources.get_distribution("zdaemon").version	1✔
72	self.add(None, None, "h", "help", self.help)	1✔
73	self.add(None, None, None, "version", self.print_version)	1✔
74	self.add("configfile", None, "C:", "configure=")	1✔
75	self.add(None, None, "X:", handler=self.zconfig_options.append)	1✔
76
77	def print_version(self, dummy):	1✔
78	"""Print zdaemon version number to stdout and exit(0)."""
79	print(self.version)	1✔
80	sys.exit(0)	1✔
81
82	def help(self, dummy):	1✔
83	"""Print a long help message (self.doc) to stdout and exit(0).
84
85	Occurrences of "%s" in self.doc are replaced by self.progname.
86	"""
87	doc = self.doc	1✔
88	if not doc:	1✔
89	doc = "No help available."	1✔
90	elif doc.find("%s") > 0:	1✔
91	doc = doc.replace("%s", self.progname)	1✔
92	print(doc, end='')	1✔
93	sys.exit(0)	1✔
94
95	def usage(self, msg):	1✔
96	"""Print a brief error message to stderr and exit(2)."""
97	sys.stderr.write("Error: %s\n" % str(msg))	1✔
98	sys.stderr.write("For help, use %s -h\n" % self.progname)	1✔
99	sys.exit(2)	1✔
100
101	def remove(self,	1✔
102	name=None, # attribute name on self
103	confname=None, # name in ZConfig (may be dotted)
104	short=None, # short option name
105	long=None, # long option name
106	):
107	"""Remove all traces of name, confname, short and/or long."""
108	if name:	×
109	for n, cn in self.names_list[:]:	×
110	if n == name:	×
111	self.names_list.remove((n, cn))	×
112	if name in self.default_map:	×
113	del self.default_map[name]	×
114	if name in self.required_map:	×
115	del self.required_map[name]	×
116	if confname:	×
117	for n, cn in self.names_list[:]:	×
118	if cn == confname:	×
119	self.names_list.remove((n, cn))	×
120	if short:	×
121	key = "-" + short[0]	×
122	if key in self.options_map:	×
123	del self.options_map[key]	×
124	if long:	×
125	key = "--" + long	×
126	if key[-1] == "=":	×
127	key = key[:-1]	×
128	if key in self.options_map:	×
129	del self.options_map[key]	×
130
131	def add(self,	1✔
132	name=None, # attribute name on self
133	confname=None, # name in ZConfig (may be dotted)
134	short=None, # short option name
135	long=None, # long option name
136	handler=None, # handler (defaults to string)
137	default=None, # default value
138	required=None, # message if not provided
139	flag=None, # if not None, flag value
140	env=None, # if not None, environment variable
141	):
142	"""Add information about a configuration option.
143
144	This can take several forms:
145
146	add(name, confname)
147	Configuration option 'confname' maps to attribute 'name'
148	add(name, None, short, long)
149	Command line option '-short' or '--long' maps to 'name'
150	add(None, None, short, long, handler)
151	Command line option calls handler
152	add(name, None, short, long, handler)
153	Assign handler return value to attribute 'name'
154
155	In addition, one of the following keyword arguments may be given:
156
157	default=... -- if not None, the default value
158	required=... -- if nonempty, an error message if no value provided
159	flag=... -- if not None, flag value for command line option
160	env=... -- if not None, name of environment variable that
161	overrides the configuration file or default
162	"""
163
164	if flag is not None:	1✔
165	if handler is not None:	1✔
166	raise ValueError("use at most one of flag= and handler=")	1✔
167	if not long and not short:	1✔
168	raise ValueError("flag= requires a command line flag")	1✔
169	if short and short.endswith(":"):	1✔
170	raise ValueError("flag= requires a command line flag")	1✔
171	if long and long.endswith("="):	1✔
172	raise ValueError("flag= requires a command line flag")	1✔
173
174	def handler(arg, flag=flag):	1✔
175	return flag	1✔
176
177	if short and long:	1✔
178	if short.endswith(":") != long.endswith("="):	1✔
179	raise ValueError(	1✔
180	f"inconsistent short/long options: {short!r} {long!r}")
181
182	if short:	1✔
183	if short[0] == "-":	1✔
184	raise ValueError("short option should not start with '-'")	1✔
185	key, rest = short[:1], short[1:]	1✔
186	if rest not in ("", ":"):	1✔
187	raise ValueError("short option should be 'x' or 'x:'")	1✔
188	key = "-" + key	1✔
189	if key in self.options_map:	1✔
190	raise ValueError("duplicate short option key '%s'" % key)	1✔
191	self.options_map[key] = (name, handler)	1✔
192	self.short_options.append(short)	1✔
193
194	if long:	1✔
195	if long[0] == "-":	1✔
196	raise ValueError("long option should not start with '-'")	1✔
197	key = long	1✔
198	if key[-1] == "=":	1✔
199	key = key[:-1]	1✔
200	key = "--" + key	1✔
201	if key in self.options_map:	1✔
202	raise ValueError("duplicate long option key '%s'" % key)	1✔
203	self.options_map[key] = (name, handler)	1✔
204	self.long_options.append(long)	1✔
205
206	if env:	1✔
207	self.environ_map[env] = (name, handler)	1✔
208
209	if name:	1✔
210	if not hasattr(self, name):	1✔
211	setattr(self, name, None)	1✔
212	self.names_list.append((name, confname))	1✔
213	if default is not None:	1✔
214	self.default_map[name] = default	1✔
215	if required:	1✔
216	self.required_map[name] = required	1✔
217
218	def realize(self, args=None, progname=None, doc=None,	1✔
219	raise_getopt_errs=True):
220	"""Realize a configuration.
221
222	Optional arguments:
223
224	args -- the command line arguments, less the program name
225	(default is sys.argv[1:])
226
227	progname -- the program name (default is sys.argv[0])
228
229	doc -- usage message (default is __doc__ of the options class)
230	"""
231
232	# Provide dynamic default method arguments
233	if args is None:	1✔
234	args = sys.argv[1:]	1✔
235
236	if progname is None:	1✔
237	progname = sys.argv[0]	1✔
238
239	self.progname = progname	1✔
240	self.doc = doc or self.__doc__	1✔
241
242	self.options = []	1✔
243	self.args = []	1✔
244
245	# Call getopt
246	try:	1✔
247	self.options, self.args = getopt.getopt(	1✔
248	args, "".join(self.short_options), self.long_options)
249	except getopt.error as msg:	1✔
250	if raise_getopt_errs:	1✔
251	self.usage(msg)	1✔
252
253	# Check for positional args
254	if self.args and not self.positional_args_allowed:	1✔
255	self.usage("positional arguments are not supported")	1✔
256
257	# Process options returned by getopt
258	for opt, arg in self.options:	1✔
259	name, handler = self.options_map[opt]	1✔
260	if handler is not None:	1✔
261	try:	1✔
262	arg = handler(arg)	1✔
263	except ValueError as msg:	1✔
264	self.usage(f"invalid value for {opt} {arg!r}: {msg}")	1✔
265	if name and arg is not None:	1✔
266	if getattr(self, name) is not None:	1✔
267	if getattr(self, name) == arg:	1✔
268	# Repeated option, but we don't mind because it
269	# just reinforces what we have.
270	continue	1✔
271	self.usage("conflicting command line option %r" % opt)	1✔
272	setattr(self, name, arg)	1✔
273
274	# Process environment variables
275	for envvar in self.environ_map.keys():	1✔
276	name, handler = self.environ_map[envvar]	1✔
277	if name and getattr(self, name, None) is not None:	1✔
278	continue	1✔
279	if envvar in os.environ:	1✔
280	value = os.environ[envvar]	1✔
281	if handler is not None:	1!
282	try:	1✔
283	value = handler(value)	1✔
284	except ValueError as msg:	1✔
285	self.usage("invalid environment value for %s %r: %s"	1✔
286	% (envvar, value, msg))
287	if name and value is not None:	1!
288	setattr(self, name, value)	1✔
289
290	if self.configfile is None:	1✔
291	self.configfile = self.default_configfile()	1✔
292	if self.zconfig_options and self.configfile is None:	1✔
293	self.usage("configuration overrides (-X) cannot be used"	1✔
294	" without a configuration file")
295	if self.configfile is not None:	1✔
296	# Process config file
297	self.load_schema()	1✔
298	try:	1✔
299	self.load_configfile()	1✔
300	except ZConfig.ConfigurationError as msg:	1✔
301	self.usage(str(msg))	1✔
302
303	# Copy config options to attributes of self. This only fills
304	# in options that aren't already set from the command line.
305	for name, confname in self.names_list:	1✔
306	if confname and getattr(self, name) is None:	1✔
307	parts = confname.split(".")	1✔
308	obj = self.configroot	1✔
309	for part in parts:	1✔
310	if obj is None:	1✔
311	break	1✔
312	# Here AttributeError is not a user error!
313	obj = getattr(obj, part)	1✔
314	setattr(self, name, obj)	1✔
315
316	# Process defaults
317	for name, value in self.default_map.items():	1✔
318	if getattr(self, name) is None:	1✔
319	setattr(self, name, value)	1✔
320
321	# Process required options
322	for name, message in self.required_map.items():	1✔
323	if getattr(self, name) is None:	1✔
324	self.usage(message)	1✔
325
326	if self.logsectionname:	1✔
327	self.load_logconf(self.logsectionname)	1✔
328
329	def default_configfile(self):	1✔
330	"""Return the name of the default config file, or None."""
331	# This allows a default configuration file to be used without
332	# affecting the -C command line option; setting self.configfile
333	# before calling realize() makes the -C option unusable since
334	# then realize() thinks it has already seen the option. If no
335	# -C is used, realize() will call this method to try to locate
336	# a configuration file.
337	return None	1✔
338
339	def load_schema(self):	1✔
340	if self.schema is None:	1!
341	# Load schema
342	if self.schemadir is None:	1!
343	self.schemadir = os.path.dirname(__file__)	1✔
344	self.schemafile = os.path.join(self.schemadir, self.schemafile)	1✔
345	self.schema = ZConfig.loadSchema(self.schemafile)	1✔
346
347	def load_configfile(self):	1✔
348	self.configroot, self.confighandlers = \	1✔
349	ZConfig.loadConfig(self.schema, self.configfile,
350	self.zconfig_options)
351
352	def load_logconf(self, sectname="eventlog"):	1✔
353	parts = sectname.split(".")	1✔
354	obj = self.configroot	1✔
355	for p in parts:	1✔
356	if obj is None:	1✔
357	break	1✔
358	obj = getattr(obj, p)	1✔
359	self.config_logger = obj	1✔
360	if obj is not None:	1✔
361	obj.startup()	1✔
362
363
364	class RunnerOptions(ZDOptions):	1✔
365
366	uid = gid = None	1✔
367
368	def __init__(self):	1✔
369	ZDOptions.__init__(self)	1✔
370	self.add("backofflimit", "runner.backoff_limit",	1✔
371	"b:", "backoff-limit=", int, default=10)
372	self.add("daemon", "runner.daemon", "d", "daemon", flag=1, default=1)	1✔
373	self.add("forever", "runner.forever", "f", "forever",	1✔
374	flag=1, default=0)
375	self.add("sockname", "runner.socket_name", "s:", "socket-name=",	1✔
376	existing_parent_dirpath, default="zdsock")
377	self.add("exitcodes", "runner.exit_codes", "x:", "exit-codes=",	1✔
378	list_of_ints, default=[0, 2])
379	self.add("user", "runner.user", "u:", "user=")	1✔
380	self.add("umask", "runner.umask", "m:", "umask=", octal_type,	1✔
381	default=0o22)
382	self.add("directory", "runner.directory", "z:", "directory=",	1✔
383	existing_parent_directory)
384	self.add("transcript", "runner.transcript", "t:", "transcript=",	1✔
385	default="/dev/null")
386
387
388	# ZConfig datatype
389
390	def list_of_ints(arg):	1✔
391	if not arg:	1✔
392	return []	1✔
393	else:
394	return list(map(int, arg.split(",")))	1✔
395
396
397	def octal_type(arg):	1✔
398	return int(arg, 8)	1✔
399
400
401	def name2signal(string):	1✔
402	"""Converts a signal name to canonical form.
403
404	Signal names are recognized without regard for case:
405
406	>>> name2signal('sighup')
407	'SIGHUP'
408	>>> name2signal('SigHup')
409	'SIGHUP'
410	>>> name2signal('SIGHUP')
411	'SIGHUP'
412
413	The leading 'SIG' is not required::
414
415	>>> name2signal('hup')
416	'SIGHUP'
417	>>> name2signal('HUP')
418	'SIGHUP'
419
420	Names that are not known cause an exception to be raised::
421
422	>>> name2signal('woohoo')
423	Traceback (most recent call last):
424	ValueError: could not convert 'woohoo' to signal name
425
426	>>> name2signal('sigwoohoo')
427	Traceback (most recent call last):
428	ValueError: could not convert 'sigwoohoo' to signal name
429
430	Numeric values are accepted to names as well::
431
432	>>> name2signal(str(signal.SIGHUP))
433	'SIGHUP'
434
435	Numeric values that can't be matched to any signal known to Python
436	are treated as errors::
437
438	>>> name2signal('-234')
439	Traceback (most recent call last):
440	ValueError: unsupported signal on this platform: -234
441
442	>>> name2signal(str(signal.NSIG)) #doctest: +ELLIPSIS
443	Traceback (most recent call last):
444	ValueError: unsupported signal on this platform: ...
445
446	Non-signal attributes of the signal module are not mistakenly
447	converted::
448
449	>>> name2signal('_ign')
450	Traceback (most recent call last):
451	ValueError: could not convert '_ign' to signal name
452
453	>>> name2signal('_DFL')
454	Traceback (most recent call last):
455	ValueError: could not convert '_DFL' to signal name
456
457	>>> name2signal('sig_ign')
458	Traceback (most recent call last):
459	ValueError: could not convert 'sig_ign' to signal name
460
461	>>> name2signal('SIG_DFL')
462	Traceback (most recent call last):
463	ValueError: could not convert 'SIG_DFL' to signal name
464
465	>>> name2signal('getsignal')
466	Traceback (most recent call last):
467	ValueError: could not convert 'getsignal' to signal name
468
469	"""
470	try:	1✔
471	v = int(string)	1✔
472	except ValueError:	1✔
473	if "_" in string:	1✔
474	raise ValueError("could not convert %r to signal name" % string)	1✔
475	if string.startswith('Signals.'): # py35 signals are an enum type	1!
UNCOV 476	string = string[len('Signals.'):]	×
477	s = string.upper()	1✔
478	if not s.startswith("SIG"):	1✔
479	s = "SIG" + s	1✔
480	v = getattr(signal, s, None)	1✔
481	if isinstance(v, int):	1✔
482	return s	1✔
483	raise ValueError("could not convert %r to signal name" % string)	1✔
484	if v >= signal.NSIG:	1✔
485	raise ValueError("unsupported signal on this platform: %s" % string)	1✔
486	for name in dir(signal):	1✔
487	if "_" in name:	1✔
488	continue	1✔
489	if getattr(signal, name) == v:	1✔
490	return name	1✔
491	raise ValueError("unsupported signal on this platform: %s" % string)	1✔
492
493
494	def existing_parent_directory(arg):	1✔
495	path = os.path.expanduser(arg)	1✔
496	if os.path.isdir(path):	1✔
497	# If the directory exists, that's fine.
498	return path	1✔
499	parent, tail = os.path.split(path)	1✔
500	if os.path.isdir(parent):	1✔
501	return path	1✔
502	raise ValueError('%s is not an existing directory' % arg)	1✔
503
504
505	def existing_parent_dirpath(arg):	1✔
506	path = os.path.expanduser(arg)	1✔
507	dir = os.path.dirname(path)	1✔
508	parent, tail = os.path.split(dir)	1✔
509	if not parent:	1✔
510	# relative pathname
511	return path	1✔
512	if os.path.isdir(parent):	1✔
513	return path	1✔
514	raise ValueError('The directory named as part of the path %s '	1✔
515	'does not exist.' % arg)
516
517
518	def _test(): # pragma: nocover
519	# Stupid test program
520	z = ZDOptions()
521	z.add("program", "zdctl.program", "p:", "program=")
522	print(z.names_list)
523	z.realize()
524	names = sorted(z.names_list[:])
525	for name, confname in names:
526	print("%-20s = %.56r" % (name, getattr(z, name)))
527
528
529	if __name__ == '__main__':
530	__file__ = sys.argv[0]
531	_test()

zopefoundation / zdaemon / 12825577578

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous