(self)
| 2154 | self.handled.set() |
| 2155 | |
| 2156 | def test_output(self): |
| 2157 | # The log message sent to the HTTPHandler is properly received. |
| 2158 | logger = logging.getLogger("http") |
| 2159 | root_logger = self.root_logger |
| 2160 | root_logger.removeHandler(self.root_logger.handlers[0]) |
| 2161 | for secure in (False, True): |
| 2162 | addr = ('localhost', 0) |
| 2163 | if secure: |
| 2164 | try: |
| 2165 | import ssl |
| 2166 | except ImportError: |
| 2167 | sslctx = None |
| 2168 | else: |
| 2169 | here = os.path.dirname(__file__) |
| 2170 | localhost_cert = os.path.join(here, "certdata", "keycert.pem") |
| 2171 | sslctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) |
| 2172 | sslctx.load_cert_chain(localhost_cert) |
| 2173 | |
| 2174 | context = ssl.create_default_context(cafile=localhost_cert) |
| 2175 | else: |
| 2176 | sslctx = None |
| 2177 | context = None |
| 2178 | self.server = server = TestHTTPServer(addr, self.handle_request, |
| 2179 | 0.01, sslctx=sslctx) |
| 2180 | server.start() |
| 2181 | server.ready.wait() |
| 2182 | host = 'localhost:%d' % server.server_port |
| 2183 | secure_client = secure and sslctx |
| 2184 | self.h_hdlr = logging.handlers.HTTPHandler(host, '/frob', |
| 2185 | secure=secure_client, |
| 2186 | context=context, |
| 2187 | credentials=('foo', 'bar')) |
| 2188 | self.log_data = None |
| 2189 | root_logger.addHandler(self.h_hdlr) |
| 2190 | |
| 2191 | for method in ('GET', 'POST'): |
| 2192 | self.h_hdlr.method = method |
| 2193 | self.handled.clear() |
| 2194 | msg = "sp\xe4m" |
| 2195 | logger.error(msg) |
| 2196 | handled = self.handled.wait(support.SHORT_TIMEOUT) |
| 2197 | self.assertTrue(handled, "HTTP request timed out") |
| 2198 | self.assertEqual(self.log_data.path, '/frob') |
| 2199 | self.assertEqual(self.command, method) |
| 2200 | if method == 'GET': |
| 2201 | d = parse_qs(self.log_data.query) |
| 2202 | else: |
| 2203 | d = parse_qs(self.post_data.decode('utf-8')) |
| 2204 | self.assertEqual(d['name'], ['http']) |
| 2205 | self.assertEqual(d['funcName'], ['test_output']) |
| 2206 | self.assertEqual(d['msg'], [msg]) |
| 2207 | |
| 2208 | self.server.stop() |
| 2209 | self.root_logger.removeHandler(self.h_hdlr) |
| 2210 | self.h_hdlr.close() |
| 2211 | |
| 2212 | class MemoryTest(BaseTest): |
| 2213 |
nothing calls this directly
no test coverage detected