9243068082

Committed 29 Nov 2023 11:25AM CUT coverage: 88.492% (-0.09%) from 88.583%

Build # 9243068082

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #9 from zopefoundation/close-files-immediately

Replace file reads and writes utilizing Path (fixes #8)

Run Details

57 of 72 branches covered (79.17%)

Branch coverage included in aggregate %.

6 of 6 new or added lines in 1 file covered. (100.0%)

166 of 180 relevant lines covered (92.22%)

0.92 hits per line

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

86.82

/src/zc/zodbrecipes/__init__.py

##############################################################################
#
# Copyright (c) 2006 Zope Corporation 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.
#
##############################################################################

import logging
import os
from io import StringIO
from pathlib import Path

import zc.buildout
import zc.recipe.egg
import ZConfig.schemaless
from ZConfig import ConfigurationSyntaxError


logger = logging.getLogger('zc.zodbrecipes')


class StorageServer:

    def __init__(self, buildout, name, options):
        self.name, self.options = options.get('name', name), options

        deployment = self.deployment = options.get('deployment')
        if deployment:
            options['deployment-name'] = buildout[deployment].get(
                'name', deployment)
            options['rc-directory'] = buildout[deployment]['rc-directory']
            options['run-directory'] = buildout[deployment]['run-directory']
            options['log-directory'] = buildout[deployment]['log-directory']
            options['etc-directory'] = buildout[deployment]['etc-directory']
            logrotate = options.get('logrotate',
                                    buildout[deployment].get('logrotate', ''))
            if logrotate.lower() == 'false':
                options['logrotate'] = ''
            else:
                options['logrotate'] = os.path.join(
                    buildout[deployment]['logrotate-directory'],
                    options['deployment-name'] + '-' + self.name)
            options['crontab-directory'] = buildout[
                deployment]['crontab-directory']
            options['user'] = buildout[deployment]['user']
        else:
            options['rc-directory'] = buildout['buildout']['bin-directory']
            options['run-directory'] = os.path.join(
                buildout['buildout']['parts-directory'],
                self.name,
            )

        options['scripts'] = ''
        options['eggs'] = options.get('eggs', '') + '\nzdaemon\nsetuptools'
        self.egg = zc.recipe.egg.Egg(buildout, name, options)

        options['runzeo'] = os.path.join(
            buildout['buildout']['bin-directory'],
            options.get('runzeo', 'runzeo'),
        )

        options['zdaemon'] = os.path.join(
            buildout['buildout']['bin-directory'],
            options.get('zdaemon', 'zdaemon'),
        )

        options['zeopack'] = os.path.join(
            buildout['buildout']['bin-directory'],
            options.get('zeopack', 'zeopack'),
        )

        if options.get('shell-script', '') not in ('true', 'false', ''):
            raise zc.buildout.UserError(
                'The shell-script option value must be "true", "false" or "".')

    def install(self):
        options = self.options

        if not os.path.exists(options['runzeo']):
            logger.warn(no_runzeo % options['runzeo'])

        run_directory = options['run-directory']
        deployment = self.deployment
        if deployment:
            zeo_conf_path = os.path.join(options['etc-directory'],
                                         self.name + '-zeo.conf')
            zdaemon_conf_path = os.path.join(options['etc-directory'],
                                             self.name + '-zdaemon.conf')
            event_log_path = os.path.join(options['log-directory'],
                                          self.name + '-zeo.log')
            socket_path = os.path.join(run_directory,
                                       self.name + '-zdaemon.sock')
            rc = options['deployment-name'] + '-' + self.name

            options.created(
                zeo_conf_path,
                zdaemon_conf_path,
                os.path.join(options['rc-directory'], rc),
            )

            logrotate = options['logrotate']
            if logrotate:
                Path(logrotate).write_text(logrotate_template % dict(
                    logfile=event_log_path,
                    rc=os.path.join(options['rc-directory'], rc),
                    conf=zdaemon_conf_path,
                ))
                options.created(logrotate)

            pack = options.get('pack')
            if pack:
                pack = pack.split()
                if len(pack) < 5:
                    raise zc.buildout.UserError(
                        'Too few crontab fields in pack specification')
                if len(pack) > 7:
                    raise zc.buildout.UserError(
                        'Too many values in pack option')
                pack_path = os.path.join(
                    options['crontab-directory'],
                    "pack-{}-{}".format(options['deployment-name'], self.name),
                )
                if not os.path.exists(options['zeopack']):
                    logger.warn("Couln'e find zeopack script, %r",
                                options['zeopack'])
        else:
            zeo_conf_path = os.path.join(run_directory, 'zeo.conf')
            zdaemon_conf_path = os.path.join(run_directory, 'zdaemon.conf')
            event_log_path = os.path.join(run_directory, 'zeo.log')
            socket_path = os.path.join(run_directory, 'zdaemon.sock')
            rc = self.name
            options.created(run_directory,
                            os.path.join(options['rc-directory'], rc),
                            )
            if not os.path.exists(run_directory):
                os.mkdir(run_directory)
            pack = pack_path = None

        zeo_conf = options.get('zeo.conf', '') + '\n'
        try:
            zeo_conf = ZConfig.schemaless.loadConfigFile(StringIO(zeo_conf))
        except ConfigurationSyntaxError as e:
            raise zc.buildout.UserError(
                '{} in:\n{}'.format(e, zeo_conf)
            )

        zeo_section = [s for s in zeo_conf.sections if s.type == 'zeo']
        if not zeo_section:
            raise zc.buildout.UserError('No zeo section was defined.')
        if len(zeo_section) > 1:
            raise zc.buildout.UserError('Too many zeo sections.')
        zeo_section = zeo_section[0]
        if 'address' not in zeo_section:
            raise zc.buildout.UserError('No ZEO address was specified.')

        storages = [s.name or '1' for s in zeo_conf.sections
                    if s.type not in ('zeo', 'eventlog', 'runner')
                    ]

        if not storages:
            raise zc.buildout.UserError('No storages were defined.')

        if not [s for s in zeo_conf.sections if s.type == 'eventlog']:
            zeo_conf.sections.append(event_log('STDOUT'))

        zdaemon_conf = options.get('zdaemon.conf', '') + '\n'
        zdaemon_conf = ZConfig.schemaless.loadConfigFile(
            StringIO(zdaemon_conf))

        defaults = {
            'program': "{} -C {}".format(options['runzeo'], zeo_conf_path),
            'daemon': 'on',
            'transcript': event_log_path,
            'socket-name': socket_path,
            'directory': run_directory,
        }
        if deployment:
            defaults['user'] = options['user']
        runner = [s for s in zdaemon_conf.sections
                  if s.type == 'runner']
        if runner:
            runner = runner[0]
        else:
            runner = ZConfig.schemaless.Section('runner')
            zdaemon_conf.sections.insert(0, runner)
        for name, value in defaults.items():
            if name not in runner:
                runner[name] = [value]

        if not [s for s in zdaemon_conf.sections
                if s.type == 'eventlog']:
            zdaemon_conf.sections.append(event_log(event_log_path))

        zdaemon_conf = str(zdaemon_conf)

        self.egg.install()
        requirements, ws = self.egg.working_set()

        Path(zeo_conf_path).write_text(str(zeo_conf))
        Path(zdaemon_conf_path).write_text(str(zdaemon_conf))

        if options.get('shell-script') == 'true':
            if not os.path.exists(options['zdaemon']):
                logger.warn(no_zdaemon % options['zdaemon'])

            contents = "%(zdaemon)s -C '%(conf)s' $*" % dict(
                zdaemon=options['zdaemon'],
                conf=zdaemon_conf_path,
            )
            if options.get('user'):
                contents = 'su %(user)s -c \\\n  "%(contents)s"' % dict(
                    user=options['user'],
                    contents=contents,
                )
            contents = "#!/bin/sh\n%s\n" % contents

            dest = os.path.join(options['rc-directory'], rc)
            if not (os.path.exists(dest) and
                    Path(dest).read_text() == contents):
                Path(dest).write_text(contents)
                os.chmod(dest, 0o755)
                logger.info("Generated shell script %r.", dest)

        else:
            self.egg.install()
            requirements, ws = self.egg.working_set()
            zc.buildout.easy_install.scripts(
                [(rc, 'zdaemon.zdctl', 'main')],
                ws, options['executable'], options['rc-directory'],
                arguments=('['
                           '\n        %r, %r,'
                           '\n        ]+sys.argv[1:]'
                           '\n        '
                           % ('-C', zdaemon_conf_path,
                              )
                           ),
            )

        if pack:
            address, = zeo_section['address']
            if ':' in address:
                host, port = address.split(':')
                address = '-h {} -p {}'.format(host, port)
            else:
                try:
                    port = int(address)
                except ValueError:
                    address = '-U ' + address
                else:
                    address = '-p ' + address
            f = open(pack_path, 'w')
            if len(pack) == 7:
                assert '@' in pack[6]
                f.write("MAILTO=%s\n" % pack.pop())

            if len(pack) == 6:
                days = pack.pop()
            else:
                days = 1

            for storage in storages:
                f.write("{} {} {} {} -S {} -d {}\n".format(
                        ' '.join(pack), options['user'],
                        options['zeopack'], address, storage, days,
                        ))
            f.close()
            options.created(pack_path)

        return options.created()

    update = install


no_runzeo = """
A runzeo script couldn't be found at:

  %r

You may need to generate a runzeo script using the
zc.recipe.eggs:script recipe and the ZEO egg, or you may need
to specify the location of a script using the runzeo option.
"""

no_zdaemon = """
A zdaemon script couldn't be found at:

  %r

You may need to generate a zdaemon script using the
zc.recipe.eggs:script recipe and the zdaemon egg.
"""


def event_log(path, *data):
    return ZConfig.schemaless.Section(
        'eventlog', '', None,
        [ZConfig.schemaless.Section('logfile', '', dict(path=[path]))])


logrotate_template = """%(logfile)s {
  rotate 5
  weekly
  postrotate
    %(rc)s -C %(conf)s reopen_transcript
  endscript
}
"""

1	##############################################################################
2	#
3	# Copyright (c) 2006 Zope Corporation 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
15	import logging	1✔
16	import os	1✔
17	from io import StringIO	1✔
18	from pathlib import Path	1✔
19
20	import zc.buildout	1✔
21	import zc.recipe.egg	1✔
22	import ZConfig.schemaless	1✔
23	from ZConfig import ConfigurationSyntaxError	1✔
24
25
26	logger = logging.getLogger('zc.zodbrecipes')	1✔
27
28
29	class StorageServer:	1✔
30
31	def __init__(self, buildout, name, options):	1✔
32	self.name, self.options = options.get('name', name), options	1✔
33
34	deployment = self.deployment = options.get('deployment')	1✔
35	if deployment:	1✔
36	options['deployment-name'] = buildout[deployment].get(	1✔
37	'name', deployment)
38	options['rc-directory'] = buildout[deployment]['rc-directory']	1✔
39	options['run-directory'] = buildout[deployment]['run-directory']	1✔
40	options['log-directory'] = buildout[deployment]['log-directory']	1✔
41	options['etc-directory'] = buildout[deployment]['etc-directory']	1✔
42	logrotate = options.get('logrotate',	1✔
43	buildout[deployment].get('logrotate', ''))
44	if logrotate.lower() == 'false':	1✔
45	options['logrotate'] = ''	1✔
46	else:
47	options['logrotate'] = os.path.join(	1✔
48	buildout[deployment]['logrotate-directory'],
49	options['deployment-name'] + '-' + self.name)
50	options['crontab-directory'] = buildout[	1✔
51	deployment]['crontab-directory']
52	options['user'] = buildout[deployment]['user']	1✔
53	else:
54	options['rc-directory'] = buildout['buildout']['bin-directory']	1✔
55	options['run-directory'] = os.path.join(	1✔
56	buildout['buildout']['parts-directory'],
57	self.name,
58	)
59
60	options['scripts'] = ''	1✔
61	options['eggs'] = options.get('eggs', '') + '\nzdaemon\nsetuptools'	1✔
62	self.egg = zc.recipe.egg.Egg(buildout, name, options)	1✔
63
64	options['runzeo'] = os.path.join(	1✔
65	buildout['buildout']['bin-directory'],
66	options.get('runzeo', 'runzeo'),
67	)
68
69	options['zdaemon'] = os.path.join(	1✔
70	buildout['buildout']['bin-directory'],
71	options.get('zdaemon', 'zdaemon'),
72	)
73
74	options['zeopack'] = os.path.join(	1✔
75	buildout['buildout']['bin-directory'],
76	options.get('zeopack', 'zeopack'),
77	)
78
79	if options.get('shell-script', '') not in ('true', 'false', ''):	1!
80	raise zc.buildout.UserError(	×
81	'The shell-script option value must be "true", "false" or "".')
82
83	def install(self):	1✔
84	options = self.options	1✔
85
86	if not os.path.exists(options['runzeo']):	1✔
87	logger.warn(no_runzeo % options['runzeo'])	1✔
88
89	run_directory = options['run-directory']	1✔
90	deployment = self.deployment	1✔
91	if deployment:	1✔
92	zeo_conf_path = os.path.join(options['etc-directory'],	1✔
93	self.name + '-zeo.conf')
94	zdaemon_conf_path = os.path.join(options['etc-directory'],	1✔
95	self.name + '-zdaemon.conf')
96	event_log_path = os.path.join(options['log-directory'],	1✔
97	self.name + '-zeo.log')
98	socket_path = os.path.join(run_directory,	1✔
99	self.name + '-zdaemon.sock')
100	rc = options['deployment-name'] + '-' + self.name	1✔
101
102	options.created(	1✔
103	zeo_conf_path,
104	zdaemon_conf_path,
105	os.path.join(options['rc-directory'], rc),
106	)
107
108	logrotate = options['logrotate']	1✔
109	if logrotate:	1✔
110	Path(logrotate).write_text(logrotate_template % dict(	1✔
111	logfile=event_log_path,
112	rc=os.path.join(options['rc-directory'], rc),
113	conf=zdaemon_conf_path,
114	))
115	options.created(logrotate)	1✔
116
117	pack = options.get('pack')	1✔
118	if pack:	1✔
119	pack = pack.split()	1✔
120	if len(pack) < 5:	1!
121	raise zc.buildout.UserError(	×
122	'Too few crontab fields in pack specification')
123	if len(pack) > 7:	1!
124	raise zc.buildout.UserError(	×
125	'Too many values in pack option')
126	pack_path = os.path.join(	1✔
127	options['crontab-directory'],
128	"pack-{}-{}".format(options['deployment-name'], self.name),
129	)
130	if not os.path.exists(options['zeopack']):	1!
131	logger.warn("Couln'e find zeopack script, %r",	×
132	options['zeopack'])
133	else:
134	zeo_conf_path = os.path.join(run_directory, 'zeo.conf')	1✔
135	zdaemon_conf_path = os.path.join(run_directory, 'zdaemon.conf')	1✔
136	event_log_path = os.path.join(run_directory, 'zeo.log')	1✔
137	socket_path = os.path.join(run_directory, 'zdaemon.sock')	1✔
138	rc = self.name	1✔
139	options.created(run_directory,	1✔
140	os.path.join(options['rc-directory'], rc),
141	)
142	if not os.path.exists(run_directory):	1✔
143	os.mkdir(run_directory)	1✔
144	pack = pack_path = None	1✔
145
146	zeo_conf = options.get('zeo.conf', '') + '\n'	1✔
147	try:	1✔
148	zeo_conf = ZConfig.schemaless.loadConfigFile(StringIO(zeo_conf))	1✔
149	except ConfigurationSyntaxError as e:	1✔
150	raise zc.buildout.UserError(	1✔
151	'{} in:\n{}'.format(e, zeo_conf)
152	)
153
154	zeo_section = [s for s in zeo_conf.sections if s.type == 'zeo']	1✔
155	if not zeo_section:	1✔
156	raise zc.buildout.UserError('No zeo section was defined.')	1✔
157	if len(zeo_section) > 1:	1!
158	raise zc.buildout.UserError('Too many zeo sections.')	×
159	zeo_section = zeo_section[0]	1✔
160	if 'address' not in zeo_section:	1!
161	raise zc.buildout.UserError('No ZEO address was specified.')	×
162
163	storages = [s.name or '1' for s in zeo_conf.sections	1✔
164	if s.type not in ('zeo', 'eventlog', 'runner')
165	]
166
167	if not storages:	1!
168	raise zc.buildout.UserError('No storages were defined.')	×
169
170	if not [s for s in zeo_conf.sections if s.type == 'eventlog']:	1✔
171	zeo_conf.sections.append(event_log('STDOUT'))	1✔
172
173	zdaemon_conf = options.get('zdaemon.conf', '') + '\n'	1✔
174	zdaemon_conf = ZConfig.schemaless.loadConfigFile(	1✔
175	StringIO(zdaemon_conf))
176
177	defaults = {	1✔
178	'program': "{} -C {}".format(options['runzeo'], zeo_conf_path),
179	'daemon': 'on',
180	'transcript': event_log_path,
181	'socket-name': socket_path,
182	'directory': run_directory,
183	}
184	if deployment:	1✔
185	defaults['user'] = options['user']	1✔
186	runner = [s for s in zdaemon_conf.sections	1✔
187	if s.type == 'runner']
188	if runner:	1!
189	runner = runner[0]	×
190	else:
191	runner = ZConfig.schemaless.Section('runner')	1✔
192	zdaemon_conf.sections.insert(0, runner)	1✔
193	for name, value in defaults.items():	1✔
194	if name not in runner:	1!
195	runner[name] = [value]	1✔
196
197	if not [s for s in zdaemon_conf.sections	1!
198	if s.type == 'eventlog']:
199	zdaemon_conf.sections.append(event_log(event_log_path))	1✔
200
201	zdaemon_conf = str(zdaemon_conf)	1✔
202
203	self.egg.install()	1✔
204	requirements, ws = self.egg.working_set()	1✔
205
206	Path(zeo_conf_path).write_text(str(zeo_conf))	1✔
207	Path(zdaemon_conf_path).write_text(str(zdaemon_conf))	1✔
208
209	if options.get('shell-script') == 'true':	1✔
210	if not os.path.exists(options['zdaemon']):	1!
211	logger.warn(no_zdaemon % options['zdaemon'])	×
212
213	contents = "%(zdaemon)s -C '%(conf)s' $*" % dict(	1✔
214	zdaemon=options['zdaemon'],
215	conf=zdaemon_conf_path,
216	)
217	if options.get('user'):	1!
218	contents = 'su %(user)s -c \\\n "%(contents)s"' % dict(	1✔
219	user=options['user'],
220	contents=contents,
221	)
222	contents = "#!/bin/sh\n%s\n" % contents	1✔
223
224	dest = os.path.join(options['rc-directory'], rc)	1✔
225	if not (os.path.exists(dest) and	1!
226	Path(dest).read_text() == contents):
227	Path(dest).write_text(contents)	1✔
228	os.chmod(dest, 0o755)	1✔
229	logger.info("Generated shell script %r.", dest)	1✔
230
231	else:
232	self.egg.install()	1✔
233	requirements, ws = self.egg.working_set()	1✔
234	zc.buildout.easy_install.scripts(	1✔
235	[(rc, 'zdaemon.zdctl', 'main')],
236	ws, options['executable'], options['rc-directory'],
237	arguments=('['
238	'\n %r, %r,'
239	'\n ]+sys.argv[1:]'
240	'\n '
241	% ('-C', zdaemon_conf_path,
242	)
243	),
244	)
245
246	if pack:	1✔
247	address, = zeo_section['address']	1✔
248	if ':' in address:	1!
249	host, port = address.split(':')	×
250	address = '-h {} -p {}'.format(host, port)	×
251	else:
252	try:	1✔
253	port = int(address)	1✔
254	except ValueError:	×
255	address = '-U ' + address	×
256	else:
257	address = '-p ' + address	1✔
258	f = open(pack_path, 'w')	1✔
259	if len(pack) == 7:	1✔
260	assert '@' in pack[6]	1✔
261	f.write("MAILTO=%s\n" % pack.pop())	1✔
262
263	if len(pack) == 6:	1!
264	days = pack.pop()	1✔
265	else:
266	days = 1	×
267
268	for storage in storages:	1✔
269	f.write("{} {} {} {} -S {} -d {}\n".format(	1✔
270	' '.join(pack), options['user'],
271	options['zeopack'], address, storage, days,
272	))
273	f.close()	1✔
274	options.created(pack_path)	1✔
275
276	return options.created()	1✔
277
278	update = install	1✔
279
280
281	no_runzeo = """	1✔
282	A runzeo script couldn't be found at:
283
284	%r
285
286	You may need to generate a runzeo script using the
287	zc.recipe.eggs:script recipe and the ZEO egg, or you may need
288	to specify the location of a script using the runzeo option.
289	"""
290
291	no_zdaemon = """	1✔
292	A zdaemon script couldn't be found at:
293
294	%r
295
296	You may need to generate a zdaemon script using the
297	zc.recipe.eggs:script recipe and the zdaemon egg.
298	"""
299
300
301	def event_log(path, *data):	1✔
302	return ZConfig.schemaless.Section(	1✔
303	'eventlog', '', None,
304	[ZConfig.schemaless.Section('logfile', '', dict(path=[path]))])
305
306
307	logrotate_template = """%(logfile)s {	1✔
308	rotate 5
309	weekly
310	postrotate
311	%(rc)s -C %(conf)s reopen_transcript
312	endscript
313	}
314	"""

zopefoundation / zc.zodbrecipes / 9243068082

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