1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
|
# Copyright 2010 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
import cStringIO
import errno
import logging
import unittest
import urllib
import socket
import re
import lazr.uri
import wsgi_intercept
from wsgi_intercept.urllib2_intercept import install_opener, uninstall_opener
import wsgi_intercept.zope_testbrowser
from paste import httpserver
from paste.httpexceptions import HTTPExceptionHandler
import zope.event
from canonical.config import config
from canonical.launchpad.webapp.errorlog import ErrorReport, ErrorReportEvent
from canonical.launchpad.webapp.vhosts import allvhosts
from canonical.testing.layers import DatabaseFunctionalLayer
from launchpad_loggerhead.app import (
_oops_html_template,
oops_middleware,
RootApp,
)
from launchpad_loggerhead.session import SessionHandler
from lp.testing import TestCase
SESSION_VAR = 'lh.session'
# See sourcecode/launchpad-loggerhead/start-loggerhead.py for the production
# mechanism for getting the secret.
SECRET = 'secret'
def session_scribbler(app, test):
"""Squirrel away the session variable."""
def scribble(environ, start_response):
test.session = environ[SESSION_VAR] # Yay for mutables.
return app(environ, start_response)
return scribble
def dummy_destination(environ, start_response):
"""Return a fake response."""
start_response('200 OK', [('Content-type','text/plain')])
return ['This is a dummy destination.\n']
class SimpleLogInRootApp(RootApp):
"""A mock root app that doesn't require open id."""
def _complete_login(self, environ, start_response):
environ[SESSION_VAR]['user'] = 'bob'
start_response('200 OK', [('Content-type','text/plain')])
return ['\n']
class TestLogout(TestCase):
layer = DatabaseFunctionalLayer
def intercept(self, uri, app):
"""Install wsgi interceptors for the uri, app tuple."""
if isinstance(uri, basestring):
uri = lazr.uri.URI(uri)
port = uri.port
if port is None:
if uri.scheme == 'http':
port = 80
elif uri.scheme == 'https':
port = 443
else:
raise NotImplementedError(uri.scheme)
else:
port = int(port)
wsgi_intercept.add_wsgi_intercept(uri.host, port, lambda: app)
self.intercepted.append((uri.host, port))
def setUp(self):
TestCase.setUp(self)
self.intercepted = []
self.session = None
self.root = app = SimpleLogInRootApp(SESSION_VAR)
app = session_scribbler(app, self)
app = HTTPExceptionHandler(app)
app = SessionHandler(app, SESSION_VAR, SECRET)
self.cookie_name = app.cookie_handler.cookie_name
self.intercept(config.codehosting.codebrowse_root, app)
self.intercept(config.codehosting.secure_codebrowse_root, app)
self.intercept(allvhosts.configs['mainsite'].rooturl,
dummy_destination)
install_opener()
self.browser = wsgi_intercept.zope_testbrowser.WSGI_Browser()
# We want to pretend we are not a robot, or else mechanize will honor
# robots.txt.
self.browser.mech_browser.set_handle_robots(False)
self.browser.open(
config.codehosting.secure_codebrowse_root + '+login')
def tearDown(self):
uninstall_opener()
for host, port in self.intercepted:
wsgi_intercept.remove_wsgi_intercept(host, port)
TestCase.tearDown(self)
def testLoggerheadLogout(self):
# We start logged in as 'bob'.
self.assertEqual(self.session['user'], 'bob')
self.browser.open(
config.codehosting.secure_codebrowse_root + 'favicon.ico')
self.assertEqual(self.session['user'], 'bob')
self.failUnless(self.browser.cookies.get(self.cookie_name))
# When we visit +logout, our session is gone.
self.browser.open(
config.codehosting.secure_codebrowse_root + '+logout')
self.assertEqual(self.session, {})
# By default, we have been redirected to the Launchpad root.
self.assertEqual(
self.browser.url, allvhosts.configs['mainsite'].rooturl)
# The session cookie still exists, because of how
# paste.auth.cookie works (see
# http://trac.pythonpaste.org/pythonpaste/ticket/139 ) but the user
# does in fact have an empty session now.
self.browser.open(
config.codehosting.secure_codebrowse_root + 'favicon.ico')
self.assertEqual(self.session, {})
def testLoggerheadLogoutRedirect(self):
# When we visit +logout with a 'next_to' value in the query string,
# the logout page will redirect to the given URI. As of this
# writing, this is used by Launchpad to redirect to our OpenId
# provider (see canonical.launchpad.tests.test_login.
# TestLoginAndLogout.test_CookieLogoutPage).
# Here, we will have a more useless example of the basic machinery.
dummy_root = 'http://dummy.dev/'
self.intercept(dummy_root, dummy_destination)
self.browser.open(
config.codehosting.secure_codebrowse_root +
'+logout?' +
urllib.urlencode(dict(next_to=dummy_root + '+logout')))
# We are logged out, as before.
self.assertEqual(self.session, {})
# Now, though, we are redirected to the ``next_to`` destination.
self.assertEqual(self.browser.url, dummy_root + '+logout')
self.assertEqual(self.browser.contents,
'This is a dummy destination.\n')
class TestOopsMiddleware(TestCase):
def setUp(self):
super(TestOopsMiddleware, self).setUp()
self.start_response_called = False
def assertContainsRe(self, haystack, needle_re, flags=0):
"""Assert that a contains something matching a regular expression."""
# There is: self.assertTextMatchesExpressionIgnoreWhitespace
# but it does weird things with whitespace, and gives
# unhelpful error messages when it fails, so this is copied
# from bzrlib
if not re.search(needle_re, haystack, flags):
if '\n' in haystack or len(haystack) > 60:
# a long string, format it in a more readable way
raise AssertionError(
'pattern "%s" not found in\n"""\\\n%s"""\n'
% (needle_re, haystack))
else:
raise AssertionError('pattern "%s" not found in "%s"'
% (needle_re, haystack))
def catchLogEvents(self):
"""Any log events that are triggered get written to self.log_stream"""
logger = logging.getLogger('lp-loggerhead')
logger.setLevel(logging.DEBUG)
self.log_stream = cStringIO.StringIO()
handler = logging.StreamHandler(self.log_stream)
handler.setLevel(logging.DEBUG)
logger.addHandler(handler)
self.addCleanup(logger.removeHandler, handler)
def runtime_failing_app(self, environ, start_response):
if False:
yield None
raise RuntimeError('just a generic runtime error.')
def socket_failing_app(self, environ, start_response):
if False:
yield None
raise socket.error(errno.EPIPE, 'Connection closed')
def logging_start_response(self, status, response_headers, exc_info=None):
self._response_chunks = []
def _write(chunk):
self._response_chunks.append(chunk)
self.start_response_called = True
return _write
def success_app(self, environ, start_response):
writer = start_response('200 OK', {})
writer('Successfull\n')
return []
def failing_start_response(self, status, response_headers, exc_info=None):
def fail_write(chunk):
raise socket.error(errno.EPIPE, 'Connection closed')
self.start_response_called = True
return fail_write
def multi_yielding_app(self, environ, start_response):
writer = start_response('200 OK', {})
yield 'content\n'
yield 'I want\n'
yield 'to give to the user\n'
def no_body_app(self, environ, start_response):
writer = start_response('200 OK', {})
return []
def _get_default_environ(self):
return {'wsgi.version': (1, 0),
'wsgi.url_scheme': 'http',
'PATH_INFO': '/test/path',
'REQUEST_METHOD': 'GET',
'SERVER_NAME': 'localhost',
'SERVER_PORT': '8080',
}
def wrap_and_run(self, app, failing_write=False):
app = oops_middleware(app)
# Just random env data, rather than setting up a whole wsgi stack just
# to pass in values for this dict
environ = self._get_default_environ()
if failing_write:
result = list(app(environ, self.failing_start_response))
else:
result = list(app(environ, self.logging_start_response))
return result
def test_exception_triggers_oops(self):
res = self.wrap_and_run(self.runtime_failing_app)
# After the exception was raised, we should also have gotten an oops
# event
self.assertEqual(1, len(self.oopses))
oops = self.oopses[0]
self.assertEqual('RuntimeError', oops.type)
# runtime_failing_app doesn't call start_response, but oops_middleware
# does because it tries to send the OOPS information to the user.
self.assertTrue(self.start_response_called)
self.assertEqual(_oops_html_template % {'oopsid': oops.id},
''.join(self._response_chunks))
def test_ignores_socket_exceptions(self):
self.catchLogEvents()
res = self.wrap_and_run(self.socket_failing_app)
self.assertEqual(0, len(self.oopses))
self.assertContainsRe(self.log_stream.getvalue(),
'Caught socket exception from <unknown>:.*Connection closed')
# start_response doesn't get called because the app fails first,
# and oops_middleware knows not to do anything with a closed socket.
self.assertFalse(self.start_response_called)
def test_ignores_writer_failures(self):
self.catchLogEvents()
res = self.wrap_and_run(self.success_app, failing_write=True)
self.assertEqual(0, len(self.oopses))
self.assertContainsRe(self.log_stream.getvalue(),
'Caught socket exception from <unknown>:.*Connection closed')
# success_app calls start_response, so this should get passed on.
self.assertTrue(self.start_response_called)
def test_stopping_early_no_oops(self):
# See bug #726985.
# If content is being streamed, and the pipe closes, we'll get a
# 'GeneratorExit', because it is closed before finishing. This doesn't
# need to become an OOPS.
self.catchLogEvents()
app = oops_middleware(self.multi_yielding_app)
environ = self._get_default_environ()
result = app(environ, self.logging_start_response)
self.assertEqual('content\n', result.next())
# At this point, we intentionally kill the app and the response, so
# that they will get GeneratorExit
del app, result
self.assertEqual([], self.oopses)
self.assertContainsRe(self.log_stream.getvalue(),
'Caught GeneratorExit from <unknown>')
# Body content was yielded, we must have called start_response
self.assertTrue(self.start_response_called)
def test_no_body_calls_start_response(self):
# See bug #732481, even if we don't have a body, if we have headers to
# send, we must call start_response
result = self.wrap_and_run(self.no_body_app)
self.assertEqual([], result)
self.assertTrue(self.start_response_called)
# Output content is empty because of no_body_app
self.assertEqual('', ''.join(self._response_chunks))
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)
|