25117017919

Committed 29 Apr 2026 04:31AM UTC coverage: 74.699% (+0.01%) from 74.686%

Build # 25117017919

Build Type

push

github

Committed by

rouilj

Commit Message

refactor: use open context managers.

Coverage Stats

0 of 3 new or added lines in 1 file covered. (0.0%)

56 existing lines in 3 files now uncovered.

19138 of 25620 relevant lines covered (74.7%)

4.4 hits per line

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

88.52

/roundup/instance.py

#
# Copyright (c) 2001 Bizar Software Pty Ltd (http://www.bizarsoftware.com.au/)
# This module is free software, and you may redistribute it and/or modify
# under the same terms as Python, so long as this copyright message and
# disclaimer are retained in their original form.
#
# IN NO EVENT SHALL BIZAR SOFTWARE PTY LTD BE LIABLE TO ANY PARTY FOR
# DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES ARISING
# OUT OF THE USE OF THIS CODE, EVEN IF THE AUTHOR HAS BEEN ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
# BIZAR SOFTWARE PTY LTD SPECIFICALLY DISCLAIMS ANY WARRANTIES, INCLUDING,
# BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
# FOR A PARTICULAR PURPOSE.  THE CODE PROVIDED HEREUNDER IS ON AN "AS IS"
# BASIS, AND THERE IS NO OBLIGATION WHATSOEVER TO PROVIDE MAINTENANCE,
# SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
#

"""Top-level tracker interface.

Open a tracker with:

    >>> from roundup import instance
    >>> db = instance.open('path to tracker home')

The "db" handle you get back is the tracker's hyperdb which has the interface
described in `roundup.hyperdb.Database`.
"""
__docformat__ = 'restructuredtext'

try:
    import builtins
except ImportError:
    import __builtin__ as builtins

import os
import sys

from roundup import actions, backends, configuration, hyperdb, mailgw
from roundup.cgi import actions as cgi_actions
from roundup.cgi import client, templating
from roundup.exceptions import RoundupException


class Tracker:
    def __init__(self, tracker_home, optimize=0):
        """New-style tracker instance constructor

        Parameters:
            tracker_home:
                tracker home directory
            optimize:
                if set, precompile html templates

        """
        self.tracker_home = tracker_home
        self.optimize = optimize
        self.config = configuration.CoreConfig(tracker_home)
        self.actions = {}
        self.cgi_actions = {}
        self.templating_utils = {}

        libdir = os.path.join(self.tracker_home, 'lib')
        self.libdir = (os.path.isdir(libdir) and libdir) or ''

        self.load_interfaces()
        self.templates = templating.get_loader(self.config["TEMPLATES"],
                                               self.config["TEMPLATE_ENGINE"])

        rdbms_backend = self.config.RDBMS_BACKEND

        self.backend = backends.get_backend(rdbms_backend)

        if self.optimize:
            self.templates.precompile()
            # initialize tracker extensions
            for extension in self.get_extensions('extensions'):
                extension(self)
            # load database schema
            self.schema = self._compile('schema.py')
            # load database detectors
            self.detectors = self.get_extensions('detectors')
            # db_open is set to True after first open()
            self.db_open = 0

    def open(self, name=None):
        # load the database schema
        # we cannot skip this part even if self.optimize is set
        # because the schema has security settings that must be
        # applied to each database instance
        backend = self.backend
        env = {
            'Class': backend.Class,
            'FileClass': backend.FileClass,
            'IssueClass': backend.IssueClass,
            'String': hyperdb.String,
            'Password': hyperdb.Password,
            'Date': hyperdb.Date,
            'Link': hyperdb.Link,
            'Multilink': hyperdb.Multilink,
            'Interval': hyperdb.Interval,
            'Boolean': hyperdb.Boolean,
            'Number': hyperdb.Number,
            'Integer': hyperdb.Integer,
            'db': backend.Database(self.config, name)
        }

        if self.optimize:
            # execute preloaded schema object
            self._exec(self.schema, env)
            # use preloaded detectors
            detectors = self.detectors
        else:
            # execute the schema file
            self._execfile('schema.py', env)
            # reload extensions and detectors
            for extension in self.get_extensions('extensions'):
                extension(self)
            detectors = self.get_extensions('detectors')
        db = env['db']
        db.tx_Source = None
        # Useful for script when multiple open calls happen. Scripts have
        # to inject the i18n object, there is currently no support for this
        if hasattr(self, 'i18n'):
            db.i18n = self.i18n

        # apply the detectors
        for detector in detectors:
            detector(db)
        # if we are running in debug mode
        # or this is the first time the database is opened,
        # do database upgrade checks
        if not (self.optimize and self.db_open):
            # As a consistency check, ensure that every link property is
            # pointing at a defined class.  Otherwise, the schema is
            # internally inconsistent.  This is an important safety
            # measure as it protects against an accidental schema change
            # dropping a table while there are still links to the table;
            # once the table has been dropped, there is no way to get it
            # back, so it is important to drop it only if we are as sure
            # as possible that it is no longer needed.
            classes = db.getclasses()
            for classname in classes:
                cl = db.getclass(classname)
                for propname, prop in cl.getprops().items():
                    if not isinstance(prop, (hyperdb.Link,
                                             hyperdb.Multilink)):
                        continue
                    linkto = prop.classname
                    if linkto not in classes:
                        raise ValueError("property %s.%s links to "
                                         "non-existent class %s"
                                         % (classname, propname, linkto))

            self.db_open = 1
        # *Must* call post_init! It is not an error if called multiple times.
        db.post_init()
        return db

    def load_interfaces(self):
        """load interfaces.py (if any), initialize Client and MailGW attrs"""
        env = {}
        if os.path.isfile(os.path.join(self.tracker_home, 'interfaces.py')):
            self._execfile('interfaces.py', env)
        self.Client = env.get('Client', client.Client)
        self.MailGW = env.get('MailGW', mailgw.MailGW)
        self.TemplatingUtils = env.get('TemplatingUtils',
                                       templating.TemplatingUtils)

    def get_extensions(self, dirname):
        """Load python extensions

        Parameters:
            dirname:
                extension directory name relative to tracker home

        Return value:
            list of init() functions for each extension

        """
        extensions = []
        dirpath = os.path.join(self.tracker_home, dirname)
        if os.path.isdir(dirpath):
            sys.path.insert(1, dirpath)
            for dir_entry in os.scandir(dirpath):
                name = dir_entry.name
                if not name.endswith('.py'):
                    continue
                env = {}
                self._execfile(os.path.join(dirname, name), env)
                extensions.append(env['init'])
            sys.path.remove(dirpath)
        return extensions

    def init(self, adminpw, tx_Source=None):
        db = self.open('admin')
        db.tx_Source = tx_Source
        self._execfile('initial_data.py',
                       {'db': db, 'adminpw': adminpw,
                        'admin_email': self.config['ADMIN_EMAIL']})
        db.commit()
        db.close()

    def exists(self):
        return self.backend.db_exists(self.config)

    def nuke(self):
        self.backend.db_nuke(self.config)

    def _compile(self, fname):
        fname = os.path.join(self.tracker_home, fname)
        with builtins.open(fname) as fnamed:
            return compile(fnamed.read(), fname, 'exec')

    def _exec(self, obj, env):
        if self.libdir:
            sys.path.insert(1, self.libdir)
        exec(obj, env)
        if self.libdir:
            sys.path.remove(self.libdir)
        return env

    def _execfile(self, fname, env):
        self._exec(self._compile(fname), env)

    def registerAction(self, name, action):

        # The logic here is this:
        # * if `action` derives from actions.Action,
        #   it is executable as a generic action.
        # * if, moreover, it also derives from cgi.actions.Bridge,
        #   it may in addition be called via CGI
        # * in all other cases we register it as a CGI action, without
        #   any check (for backward compatibility).
        if issubclass(action, actions.Action):
            self.actions[name] = action
            if issubclass(action, cgi_actions.Bridge):
                self.cgi_actions[name] = action
        else:
            self.cgi_actions[name] = action

    def registerUtil(self, name, function):
        """Register a function that can be called using:
           `utils.<name>(...)`.

           The function is defined as:

               def function(...):

           If you need access to the client, database, form or other
           item, you have to pass it explicitly::

               utils.name(request.client, ...)

           If you need client access, consider using registerUtilMethod()
           instead.

        """
        self.templating_utils[name] = function

    def registerUtilMethod(self, name, function):
        """Register a method that can be called using:
           `utils.<name>(...)`.

           Unlike registerUtil, the method is defined as:

               def function(self, ...):

           `self` is a TemplatingUtils object. You can use self.client
           to access the client object for your request.
        """
        setattr(self.TemplatingUtils,
                name,
                function)


class TrackerError(RoundupException):
    pass


def open(tracker_home, optimize=0):
    if os.path.exists(os.path.join(tracker_home, 'dbinit.py')):
        # user should upgrade...
        raise TrackerError("Old style trackers using dbinit.py "
                           "are not supported after release 2.0")

    return Tracker(tracker_home, optimize=optimize)

# vim: set filetype=python sts=4 sw=4 et si :

1	#
2	# Copyright (c) 2001 Bizar Software Pty Ltd (http://www.bizarsoftware.com.au/)
3	# This module is free software, and you may redistribute it and/or modify
4	# under the same terms as Python, so long as this copyright message and
5	# disclaimer are retained in their original form.
6	#
7	# IN NO EVENT SHALL BIZAR SOFTWARE PTY LTD BE LIABLE TO ANY PARTY FOR
8	# DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES ARISING
9	# OUT OF THE USE OF THIS CODE, EVEN IF THE AUTHOR HAS BEEN ADVISED OF THE
10	# POSSIBILITY OF SUCH DAMAGE.
11	#
12	# BIZAR SOFTWARE PTY LTD SPECIFICALLY DISCLAIMS ANY WARRANTIES, INCLUDING,
13	# BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
14	# FOR A PARTICULAR PURPOSE. THE CODE PROVIDED HEREUNDER IS ON AN "AS IS"
15	# BASIS, AND THERE IS NO OBLIGATION WHATSOEVER TO PROVIDE MAINTENANCE,
16	# SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
17	#
18
19	"""Top-level tracker interface.	2✔
20
21	Open a tracker with:
22
23	>>> from roundup import instance
24	>>> db = instance.open('path to tracker home')
25
26	The "db" handle you get back is the tracker's hyperdb which has the interface
27	described in `roundup.hyperdb.Database`.
28	"""
29	__docformat__ = 'restructuredtext'	6✔
30
31	try:	6✔
32	import builtins	6✔
33	except ImportError:	×
34	import __builtin__ as builtins	×
35
36	import os	6✔
37	import sys	6✔
38
39	from roundup import actions, backends, configuration, hyperdb, mailgw	6✔
40	from roundup.cgi import actions as cgi_actions	6✔
41	from roundup.cgi import client, templating	6✔
42	from roundup.exceptions import RoundupException	6✔
43
44
45	class Tracker:	6✔
46	def __init__(self, tracker_home, optimize=0):	6✔
47	"""New-style tracker instance constructor
48
49	Parameters:
50	tracker_home:
51	tracker home directory
52	optimize:
53	if set, precompile html templates
54
55	"""
56	self.tracker_home = tracker_home	6✔
57	self.optimize = optimize	6✔
58	self.config = configuration.CoreConfig(tracker_home)	6✔
59	self.actions = {}	6✔
60	self.cgi_actions = {}	6✔
61	self.templating_utils = {}	6✔
62
63	libdir = os.path.join(self.tracker_home, 'lib')	6✔
64	self.libdir = (os.path.isdir(libdir) and libdir) or ''	6✔
65
66	self.load_interfaces()	6✔
67	self.templates = templating.get_loader(self.config["TEMPLATES"],	6✔
68	self.config["TEMPLATE_ENGINE"])
69
70	rdbms_backend = self.config.RDBMS_BACKEND	6✔
71
72	self.backend = backends.get_backend(rdbms_backend)	6✔
73
74	if self.optimize:	6✔
75	self.templates.precompile()	6✔
76	# initialize tracker extensions
77	for extension in self.get_extensions('extensions'):	6✔
UNCOV 78	extension(self)	×
79	# load database schema
80	self.schema = self._compile('schema.py')	6✔
81	# load database detectors
82	self.detectors = self.get_extensions('detectors')	6✔
83	# db_open is set to True after first open()
84	self.db_open = 0	6✔
85
86	def open(self, name=None):	6✔
87	# load the database schema
88	# we cannot skip this part even if self.optimize is set
89	# because the schema has security settings that must be
90	# applied to each database instance
91	backend = self.backend	6✔
92	env = {	6✔
93	'Class': backend.Class,
94	'FileClass': backend.FileClass,
95	'IssueClass': backend.IssueClass,
96	'String': hyperdb.String,
97	'Password': hyperdb.Password,
98	'Date': hyperdb.Date,
99	'Link': hyperdb.Link,
100	'Multilink': hyperdb.Multilink,
101	'Interval': hyperdb.Interval,
102	'Boolean': hyperdb.Boolean,
103	'Number': hyperdb.Number,
104	'Integer': hyperdb.Integer,
105	'db': backend.Database(self.config, name)
106	}
107
108	if self.optimize:	6✔
109	# execute preloaded schema object
110	self._exec(self.schema, env)	6✔
111	# use preloaded detectors
112	detectors = self.detectors	6✔
113	else:
114	# execute the schema file
115	self._execfile('schema.py', env)	6✔
116	# reload extensions and detectors
117	for extension in self.get_extensions('extensions'):	6✔
UNCOV 118	extension(self)	×
119	detectors = self.get_extensions('detectors')	6✔
120	db = env['db']	6✔
121	db.tx_Source = None	6✔
122	# Useful for script when multiple open calls happen. Scripts have
123	# to inject the i18n object, there is currently no support for this
124	if hasattr(self, 'i18n'):	6✔
UNCOV 125	db.i18n = self.i18n	×
126
127	# apply the detectors
128	for detector in detectors:	6✔
129	detector(db)	6✔
130	# if we are running in debug mode
131	# or this is the first time the database is opened,
132	# do database upgrade checks
133	if not (self.optimize and self.db_open):	6✔
134	# As a consistency check, ensure that every link property is
135	# pointing at a defined class. Otherwise, the schema is
136	# internally inconsistent. This is an important safety
137	# measure as it protects against an accidental schema change
138	# dropping a table while there are still links to the table;
139	# once the table has been dropped, there is no way to get it
140	# back, so it is important to drop it only if we are as sure
141	# as possible that it is no longer needed.
142	classes = db.getclasses()	6✔
143	for classname in classes:	6✔
144	cl = db.getclass(classname)	6✔
145	for propname, prop in cl.getprops().items():	6✔
146	if not isinstance(prop, (hyperdb.Link,	6✔
147	hyperdb.Multilink)):
148	continue	6✔
149	linkto = prop.classname	6✔
150	if linkto not in classes:	6✔
UNCOV 151	raise ValueError("property %s.%s links to "	×
152	"non-existent class %s"
153	% (classname, propname, linkto))
154
155	self.db_open = 1	6✔
156	# Must call post_init! It is not an error if called multiple times.
157	db.post_init()	6✔
158	return db	6✔
159
160	def load_interfaces(self):	6✔
161	"""load interfaces.py (if any), initialize Client and MailGW attrs"""
162	env = {}	6✔
163	if os.path.isfile(os.path.join(self.tracker_home, 'interfaces.py')):	6✔
UNCOV 164	self._execfile('interfaces.py', env)	×
165	self.Client = env.get('Client', client.Client)	6✔
166	self.MailGW = env.get('MailGW', mailgw.MailGW)	6✔
167	self.TemplatingUtils = env.get('TemplatingUtils',	6✔
168	templating.TemplatingUtils)
169
170	def get_extensions(self, dirname):	6✔
171	"""Load python extensions
172
173	Parameters:
174	dirname:
175	extension directory name relative to tracker home
176
177	Return value:
178	list of init() functions for each extension
179
180	"""
181	extensions = []	6✔
182	dirpath = os.path.join(self.tracker_home, dirname)	6✔
183	if os.path.isdir(dirpath):	6✔
184	sys.path.insert(1, dirpath)	6✔
185	for dir_entry in os.scandir(dirpath):	6✔
186	name = dir_entry.name	6✔
187	if not name.endswith('.py'):	6✔
188	continue	6✔
189	env = {}	6✔
190	self._execfile(os.path.join(dirname, name), env)	6✔
191	extensions.append(env['init'])	6✔
192	sys.path.remove(dirpath)	6✔
193	return extensions	6✔
194
195	def init(self, adminpw, tx_Source=None):	6✔
196	db = self.open('admin')	6✔
197	db.tx_Source = tx_Source	6✔
198	self._execfile('initial_data.py',	6✔
199	{'db': db, 'adminpw': adminpw,
200	'admin_email': self.config['ADMIN_EMAIL']})
201	db.commit()	6✔
202	db.close()	6✔
203
204	def exists(self):	6✔
205	return self.backend.db_exists(self.config)	6✔
206
207	def nuke(self):	6✔
208	self.backend.db_nuke(self.config)	6✔
209
210	def _compile(self, fname):	6✔
211	fname = os.path.join(self.tracker_home, fname)	6✔
212	with builtins.open(fname) as fnamed:	6✔
213	return compile(fnamed.read(), fname, 'exec')	6✔
214
215	def _exec(self, obj, env):	6✔
216	if self.libdir:	6✔
UNCOV 217	sys.path.insert(1, self.libdir)	×
218	exec(obj, env)	6✔
219	if self.libdir:	6✔
UNCOV 220	sys.path.remove(self.libdir)	×
221	return env	6✔
222
223	def _execfile(self, fname, env):	6✔
224	self._exec(self._compile(fname), env)	6✔
225
226	def registerAction(self, name, action):	6✔
227
228	# The logic here is this:
229	# * if `action` derives from actions.Action,
230	# it is executable as a generic action.
231	# * if, moreover, it also derives from cgi.actions.Bridge,
232	# it may in addition be called via CGI
233	# * in all other cases we register it as a CGI action, without
234	# any check (for backward compatibility).
235	if issubclass(action, actions.Action):	6✔
UNCOV 236	self.actions[name] = action	×
UNCOV 237	if issubclass(action, cgi_actions.Bridge):	×
UNCOV 238	self.cgi_actions[name] = action	×
239	else:
240	self.cgi_actions[name] = action	6✔
241
242	def registerUtil(self, name, function):	6✔
243	"""Register a function that can be called using:
244	`utils.<name>(...)`.
245
246	The function is defined as:
247
248	def function(...):
249
250	If you need access to the client, database, form or other
251	item, you have to pass it explicitly::
252
253	utils.name(request.client, ...)
254
255	If you need client access, consider using registerUtilMethod()
256	instead.
257
258	"""
UNCOV 259	self.templating_utils[name] = function	×
260
261	def registerUtilMethod(self, name, function):	6✔
262	"""Register a method that can be called using:
263	`utils.<name>(...)`.
264
265	Unlike registerUtil, the method is defined as:
266
267	def function(self, ...):
268
269	`self` is a TemplatingUtils object. You can use self.client
270	to access the client object for your request.
271	"""
UNCOV 272	setattr(self.TemplatingUtils,	×
273	name,
274	function)
275
276
277	class TrackerError(RoundupException):	6✔
278	pass	6✔
279
280
281	def open(tracker_home, optimize=0):	6✔
282	if os.path.exists(os.path.join(tracker_home, 'dbinit.py')):	6✔
283	# user should upgrade...
284	raise TrackerError("Old style trackers using dbinit.py "	6✔
285	"are not supported after release 2.0")
286
287	return Tracker(tracker_home, optimize=optimize)	6✔
288
289	# vim: set filetype=python sts=4 sw=4 et si :

roundup-tracker / roundup / 25117017919

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