25075727545

Committed 28 Apr 2026 08:16PM UTC coverage: 74.686% (+0.004%) from 74.682%

Build # 25075727545

Build Type

push

github

Committed by

rouilj

Commit Message

chore(ruff):

hyperdb.py:

  sort imports

  collapse multiple:
       if x:
       else:
         if Y:
         else:
    to
       if x:
       elif y:
       else:

    One collapse included multiple nested levels.

  replace vars:
    id -> node_id (3 times) (shadows builtins)

    id -> item_id

    parameter name 'property' to 'property_' to match
      blobfiles::FileStorage like prototype methods.  (shadows
      builtins)

    p -> prop_name

  replace multiple set(generator) with {set comprehension}

  replace
      for k in X:
        ...(X[k])
    with
      for k, v in X.values():
         ...(v)

  remove redundant list from list(sorted(...))

  added FIXME 3.10 to uses of zip without strict=True. Can add strict
     when 3.10 is earliest supported version. (2 instances)

  replace "string with multiple words".split() into literal list of
    individual words.

  replace dict([(a,b) for x in y]) with dict comprehension.
    {a:b for x in y} (3 cases)

  replace for loop with list comprehension.

  replace "for X in d.keys()" with "for X in d"

  bytes != str -> bytes is not str (2 cases)

  whitespace fixes

Coverage Stats

61 of 63 new or added lines in 1 file covered. (96.83%)

667 existing lines in 10 files now uncovered.

19142 of 25630 relevant lines covered (74.69%)

4.39 hits per line

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

87.4

/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

try:
    from collections.abc import Callable
except ImportError:
    from collections import Callable

import os
import sys

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

roundup-tracker / roundup / 25075727545

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