(self)
| 2180 | self.handled.set() |
| 2181 | |
| 2182 | def test_output(self): |
| 2183 | # The log message sent to the HTTPHandler is properly received. |
| 2184 | logger = logging.getLogger("http") |
| 2185 | root_logger = self.root_logger |
| 2186 | root_logger.removeHandler(self.root_logger.handlers[0]) |
| 2187 | for secure in (False, True): |
| 2188 | addr = ('localhost', 0) |
| 2189 | if secure: |
| 2190 | try: |
| 2191 | import ssl |
| 2192 | except ImportError: |
| 2193 | sslctx = None |
| 2194 | else: |
| 2195 | here = os.path.dirname(__file__) |
| 2196 | localhost_cert = os.path.join(here, "certdata", "keycert.pem") |
| 2197 | sslctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) |
| 2198 | sslctx.load_cert_chain(localhost_cert) |
| 2199 | |
| 2200 | context = ssl.create_default_context(cafile=localhost_cert) |
| 2201 | else: |
| 2202 | sslctx = None |
| 2203 | context = None |
| 2204 | self.server = server = TestHTTPServer(addr, self.handle_request, |
| 2205 | 0.01, sslctx=sslctx) |
| 2206 | server.start() |
| 2207 | server.ready.wait() |
| 2208 | host = 'localhost:%d' % server.server_port |
| 2209 | secure_client = secure and sslctx |
| 2210 | self.h_hdlr = logging.handlers.HTTPHandler(host, '/frob', |
| 2211 | secure=secure_client, |
| 2212 | context=context, |
| 2213 | credentials=('foo', 'bar')) |
| 2214 | self.log_data = None |
| 2215 | root_logger.addHandler(self.h_hdlr) |
| 2216 | |
| 2217 | for method in ('GET', 'POST'): |
| 2218 | self.h_hdlr.method = method |
| 2219 | self.handled.clear() |
| 2220 | msg = "sp\xe4m" |
| 2221 | logger.error(msg) |
| 2222 | handled = self.handled.wait(support.SHORT_TIMEOUT) |
| 2223 | self.assertTrue(handled, "HTTP request timed out") |
| 2224 | self.assertEqual(self.log_data.path, '/frob') |
| 2225 | self.assertEqual(self.command, method) |
| 2226 | if method == 'GET': |
| 2227 | d = parse_qs(self.log_data.query) |
| 2228 | else: |
| 2229 | d = parse_qs(self.post_data.decode('utf-8')) |
| 2230 | self.assertEqual(d['name'], ['http']) |
| 2231 | self.assertEqual(d['funcName'], ['test_output']) |
| 2232 | self.assertEqual(d['msg'], [msg]) |
| 2233 | |
| 2234 | self.server.stop() |
| 2235 | self.root_logger.removeHandler(self.h_hdlr) |
| 2236 | self.h_hdlr.close() |
| 2237 | |
| 2238 | class MemoryTest(BaseTest): |
| 2239 |
nothing calls this directly
no test coverage detected