17787786822

Committed 17 Sep 2025 05:33AM UTC coverage: 74.642% (+0.1%) from 74.519%

Build # 17787786822

Build Type

push

github

Committed by

rouilj

Commit Message

fix: use getloing value --unknown-- on error.

In case there is a real user with name unknown, Unlikely for account to
to have dashes in name.

Run Details

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

559 existing lines in 6 files now uncovered.

19024 of 25487 relevant lines covered (74.64%)

4.39 hits per line

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

94.12

/roundup/cgi/wsgi_handler.py

# WSGI interface for Roundup Issue Tracker
#
# This module is free software, you may redistribute it
# and/or modify under the same terms as Python.
#

import os
from contextlib import contextmanager

import roundup.instance
from roundup.anypy import http_
from roundup.anypy.html import html_escape
from roundup.anypy.strings import s2b
from roundup.cgi import TranslationService
from roundup.cgi.client import BinaryFieldStorage
from roundup.logcontext import gen_trace_id, store_trace_reason

BaseHTTPRequestHandler = http_.server.BaseHTTPRequestHandler
DEFAULT_ERROR_MESSAGE = http_.server.DEFAULT_ERROR_MESSAGE

try:
    # python2 is missing this definition
    http_.server.BaseHTTPRequestHandler.responses[429]
except KeyError:
    http_.server.BaseHTTPRequestHandler.responses[429] = (
         'Too Many Requests',
        'The user has sent too many requests in '
        'a given amount of time ("rate limiting")',
    )


class Headers(object):
    """ Idea more or less stolen from the 'apache.py' in same directory.
        Except that wsgi stores http headers in environment.
    """
    def __init__(self, environ):
        self.environ = environ

    def mangle_name(self, name):
        """ Content-Type is handled specially, it doesn't have a HTTP_
            prefix in cgi.
        """
        n = name.replace('-', '_').upper()
        if n == 'CONTENT_TYPE':
            return n
        return 'HTTP_' + n

    def get(self, name, default=None):
        return self.environ.get(self.mangle_name(name), default)
    getheader = get


class Writer(object):
    '''Perform a start_response if need be when we start writing.'''
    def __init__(self, request):
        self.request = request  # weakref.ref(request)

    def write(self, data):
        f = self.request.get_wfile()
        self.write = f
        return self.write(data)


class RequestHandler(object):
    def __init__(self, environ, start_response):
        self.__start_response = start_response
        self.__wfile = None
        self.headers = Headers(environ)
        self.rfile, self.wfile = None, Writer(self)

    def start_response(self, headers, response_code):
        """Set HTTP response code"""
        message, _explain = BaseHTTPRequestHandler.responses[response_code]
        self.__wfile = self.__start_response('%d %s' % (response_code,
                                                        message), headers)

    def get_wfile(self):
        if self.__wfile is None:
            raise ValueError('start_response() not called')
        return self.__wfile


class RequestDispatcher(object):
    def __init__(self, home, debug=False, timing=False, lang=None,
                 feature_flags=None):
        if not os.path.isdir(home):
            raise ValueError('%r is not a directory' % (home,))
        self.home = home
        self.debug = debug
        self.timing = timing
        self.feature_flags = feature_flags or {}
        self.tracker = None
        if lang:
            self.translator = TranslationService.get_translation(
                lang,
                tracker_home=home)
        else:
            self.translator = None

        if "cache_tracker" not in self.feature_flags or \
           self.feature_flags["cache_tracker"] is not False:
            self.tracker = roundup.instance.open(self.home, not self.debug)
        else:
            self.preload()

    @gen_trace_id()
    @store_trace_reason("wsgi")
    def __call__(self, environ, start_response):
        """Initialize with `apache.Request` object"""
        request = RequestHandler(environ, start_response)

        if environ['REQUEST_METHOD'] == 'OPTIONS':
            if environ["PATH_INFO"][:5] == "/rest":
                # rest does support options
                # This I hope will result in self.form=None
                environ['CONTENT_LENGTH'] = 0
            else:
                code = 501
                message, explain = BaseHTTPRequestHandler.responses[code]
                request.start_response([('Content-Type', 'text/html')],
                                       code)
                request.wfile.write(s2b(DEFAULT_ERROR_MESSAGE % locals()))
                return []

        # need to strip the leading '/'
        environ["PATH_INFO"] = environ["PATH_INFO"][1:]
        if self.timing:
            environ["CGI_SHOW_TIMING"] = self.timing

        if environ['REQUEST_METHOD'] in ("OPTIONS", "DELETE"):
            # these methods have no data. When we init tracker.Client
            # set form to None to get a properly initialized empty
            # form.
            form = None
        else:
            form = BinaryFieldStorage(fp=environ['wsgi.input'], environ=environ)

        if "cache_tracker" not in self.feature_flags or \
           self.feature_flags["cache_tracker"] is not False:
            client = self.tracker.Client(self.tracker, request, environ, form,
                                         self.translator)
            try:
                client.main()
            except roundup.cgi.client.NotFound:
                request.start_response([('Content-Type', 'text/html')], 404)
                request.wfile.write(s2b('Not found: %s' %
                                        html_escape(client.path)))
        else:
            with self.get_tracker() as tracker:
                client = tracker.Client(tracker, request, environ, form,
                                        self.translator)
                try:
                    client.main()
                except roundup.cgi.client.NotFound:
                    request.start_response([('Content-Type', 'text/html')], 404)
                    request.wfile.write(s2b('Not found: %s' %
                                            html_escape(client.path)))

        # all body data has been written using wfile
        return []

    def preload(self):
        """ Trigger pre-loading of imports and templates """
        with self.get_tracker():
            pass

    @contextmanager
    def get_tracker(self):
        # get a new instance for each request
        yield roundup.instance.open(self.home, not self.debug)

1	# WSGI interface for Roundup Issue Tracker
2	#
3	# This module is free software, you may redistribute it
4	# and/or modify under the same terms as Python.
5	#
6
7	import os	6✔
8	from contextlib import contextmanager	6✔
9
10	import roundup.instance	6✔
11	from roundup.anypy import http_	6✔
12	from roundup.anypy.html import html_escape	6✔
13	from roundup.anypy.strings import s2b	6✔
14	from roundup.cgi import TranslationService	6✔
15	from roundup.cgi.client import BinaryFieldStorage	6✔
16	from roundup.logcontext import gen_trace_id, store_trace_reason	6✔
17
18	BaseHTTPRequestHandler = http_.server.BaseHTTPRequestHandler	6✔
19	DEFAULT_ERROR_MESSAGE = http_.server.DEFAULT_ERROR_MESSAGE	6✔
20
21	try:	6✔
22	# python2 is missing this definition
23	http_.server.BaseHTTPRequestHandler.responses[429]	6✔
24	except KeyError:	×
UNCOV 25	http_.server.BaseHTTPRequestHandler.responses[429] = (	×
26	'Too Many Requests',
27	'The user has sent too many requests in '
28	'a given amount of time ("rate limiting")',
29	)
30
31
32	class Headers(object):	6✔
33	""" Idea more or less stolen from the 'apache.py' in same directory.	2✔
34	Except that wsgi stores http headers in environment.
35	"""
36	def __init__(self, environ):	6✔
37	self.environ = environ	6✔
38
39	def mangle_name(self, name):	6✔
40	""" Content-Type is handled specially, it doesn't have a HTTP_
41	prefix in cgi.
42	"""
43	n = name.replace('-', '_').upper()	6✔
44	if n == 'CONTENT_TYPE':	6✔
45	return n	6✔
46	return 'HTTP_' + n	6✔
47
48	def get(self, name, default=None):	6✔
49	return self.environ.get(self.mangle_name(name), default)	6✔
50	getheader = get	6✔
51
52
53	class Writer(object):	6✔
54	'''Perform a start_response if need be when we start writing.'''	2✔
55	def __init__(self, request):	6✔
56	self.request = request # weakref.ref(request)	6✔
57
58	def write(self, data):	6✔
59	f = self.request.get_wfile()	6✔
60	self.write = f	6✔
61	return self.write(data)	6✔
62
63
64	class RequestHandler(object):	6✔
65	def __init__(self, environ, start_response):	6✔
66	self.__start_response = start_response	6✔
67	self.__wfile = None	6✔
68	self.headers = Headers(environ)	6✔
69	self.rfile, self.wfile = None, Writer(self)	6✔
70
71	def start_response(self, headers, response_code):	6✔
72	"""Set HTTP response code"""
73	message, _explain = BaseHTTPRequestHandler.responses[response_code]	6✔
74	self.__wfile = self.__start_response('%d %s' % (response_code,	6✔
75	message), headers)
76
77	def get_wfile(self):	6✔
78	if self.__wfile is None:	6✔
UNCOV 79	raise ValueError('start_response() not called')	×
80	return self.__wfile	6✔
81
82
83	class RequestDispatcher(object):	6✔
84	def __init__(self, home, debug=False, timing=False, lang=None,	6✔
85	feature_flags=None):
86	if not os.path.isdir(home):	6✔
UNCOV 87	raise ValueError('%r is not a directory' % (home,))	×
88	self.home = home	6✔
89	self.debug = debug	6✔
90	self.timing = timing	6✔
91	self.feature_flags = feature_flags or {}	6✔
92	self.tracker = None	6✔
93	if lang:	6✔
UNCOV 94	self.translator = TranslationService.get_translation(	×
95	lang,
96	tracker_home=home)
97	else:
98	self.translator = None	6✔
99
100	if "cache_tracker" not in self.feature_flags or \	6✔
101	self.feature_flags["cache_tracker"] is not False:
102	self.tracker = roundup.instance.open(self.home, not self.debug)	6✔
103	else:
104	self.preload()	6✔
105
106	@gen_trace_id()	6✔
107	@store_trace_reason("wsgi")	6✔
108	def __call__(self, environ, start_response):	5✔
109	"""Initialize with `apache.Request` object"""
110	request = RequestHandler(environ, start_response)	6✔
111
112	if environ['REQUEST_METHOD'] == 'OPTIONS':	6✔
113	if environ["PATH_INFO"][:5] == "/rest":	6✔
114	# rest does support options
115	# This I hope will result in self.form=None
116	environ['CONTENT_LENGTH'] = 0	6✔
117	else:
118	code = 501	6✔
119	message, explain = BaseHTTPRequestHandler.responses[code]	6✔
120	request.start_response([('Content-Type', 'text/html')],	6✔
121	code)
122	request.wfile.write(s2b(DEFAULT_ERROR_MESSAGE % locals()))	6✔
123	return []	6✔
124
125	# need to strip the leading '/'
126	environ["PATH_INFO"] = environ["PATH_INFO"][1:]	6✔
127	if self.timing:	6✔
UNCOV 128	environ["CGI_SHOW_TIMING"] = self.timing	×
129
130	if environ['REQUEST_METHOD'] in ("OPTIONS", "DELETE"):	6✔
131	# these methods have no data. When we init tracker.Client
132	# set form to None to get a properly initialized empty
133	# form.
134	form = None	6✔
135	else:
136	form = BinaryFieldStorage(fp=environ['wsgi.input'], environ=environ)	6✔
137
138	if "cache_tracker" not in self.feature_flags or \	6✔
139	self.feature_flags["cache_tracker"] is not False:
140	client = self.tracker.Client(self.tracker, request, environ, form,	6✔
141	self.translator)
142	try:	6✔
143	client.main()	6✔
144	except roundup.cgi.client.NotFound:	6✔
145	request.start_response([('Content-Type', 'text/html')], 404)	6✔
146	request.wfile.write(s2b('Not found: %s' %	6✔
147	html_escape(client.path)))
148	else:
149	with self.get_tracker() as tracker:	6✔
150	client = tracker.Client(tracker, request, environ, form,	6✔
151	self.translator)
152	try:	6✔
153	client.main()	6✔
154	except roundup.cgi.client.NotFound:	6✔
155	request.start_response([('Content-Type', 'text/html')], 404)	6✔
156	request.wfile.write(s2b('Not found: %s' %	6✔
157	html_escape(client.path)))
158
159	# all body data has been written using wfile
160	return []	6✔
161
162	def preload(self):	6✔
163	""" Trigger pre-loading of imports and templates """
164	with self.get_tracker():	6✔
165	pass	6✔
166
167	@contextmanager	6✔
168	def get_tracker(self):	5✔
169	# get a new instance for each request
170	yield roundup.instance.open(self.home, not self.debug)	6✔

roundup-tracker / roundup / 17787786822

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