Improved coverage and fixed bug in HTTPHandler with POST requests.

This commit is contained in:
Vinay Sajip 2011-05-05 12:59:14 +01:00
parent 9cc432d937
commit 0372e10684
2 changed files with 36 additions and 12 deletions

View File

@ -29,7 +29,7 @@ from stat import ST_DEV, ST_INO, ST_MTIME
import queue import queue
try: try:
import threading import threading
except ImportError: except ImportError: #pragma: no cover
threading = None threading = None
try: try:
@ -1044,7 +1044,9 @@ class HTTPHandler(logging.Handler):
s = ('u%s:%s' % self.credentials).encode('utf-8') s = ('u%s:%s' % self.credentials).encode('utf-8')
s = 'Basic ' + base64.b64encode(s).strip() s = 'Basic ' + base64.b64encode(s).strip()
h.putheader('Authorization', s) h.putheader('Authorization', s)
h.endheaders(data if self.method == "POST" else None) h.endheaders()
if self.method == "POST":
h.send(data.encode('utf-8'))
h.getresponse() #can't do anything with the result h.getresponse() #can't do anything with the result
except (KeyboardInterrupt, SystemExit): #pragma: no cover except (KeyboardInterrupt, SystemExit): #pragma: no cover
raise raise

View File

@ -1464,22 +1464,34 @@ class HTTPHandlerTest(BaseTest):
BaseTest.tearDown(self) BaseTest.tearDown(self)
def handle_request(self, request): def handle_request(self, request):
self.command = request.command
self.log_data = urlparse(request.path) self.log_data = urlparse(request.path)
if self.command == 'POST':
try:
rlen = int(request.headers['Content-Length'])
self.post_data = request.rfile.read(rlen)
except:
self.post_data = None
request.send_response(200) request.send_response(200)
self.handled.set() self.handled.set()
def test_output(self): def test_output(self):
# The log message sent to the SysLogHandler is properly received. # The log message sent to the SysLogHandler is properly received.
logger = logging.getLogger("http") logger = logging.getLogger("http")
msg = "sp\xe4m" for method in ('GET', 'POST'):
logger.error(msg) self.h_hdlr.method = method
self.handled.wait() msg = "sp\xe4m"
self.assertEqual(self.log_data.path, '/frob') logger.error(msg)
d = parse_qs(self.log_data.query) self.handled.wait()
self.assertEqual(d['name'], ['http']) self.assertEqual(self.log_data.path, '/frob')
self.assertEqual(d['funcName'], ['test_output']) self.assertEqual(self.command, method)
self.assertEqual(d['msg'], [msg]) if method == 'GET':
d = parse_qs(self.log_data.query)
else:
d = parse_qs(self.post_data.decode('utf-8'))
self.assertEqual(d['name'], ['http'])
self.assertEqual(d['funcName'], ['test_output'])
self.assertEqual(d['msg'], [msg])
class MemoryTest(BaseTest): class MemoryTest(BaseTest):
@ -3470,7 +3482,8 @@ class RotatingFileHandlerTest(BaseFileTest):
class TimedRotatingFileHandlerTest(BaseFileTest): class TimedRotatingFileHandlerTest(BaseFileTest):
# other test methods added below # other test methods added below
def test_rollover(self): def test_rollover(self):
fh = logging.handlers.TimedRotatingFileHandler(self.fn, 'S') fh = logging.handlers.TimedRotatingFileHandler(self.fn, 'S',
backupCount=1)
r = logging.makeLogRecord({'msg': 'testing'}) r = logging.makeLogRecord({'msg': 'testing'})
fh.emit(r) fh.emit(r)
self.assertLogFile(self.fn) self.assertLogFile(self.fn)
@ -3481,6 +3494,15 @@ class TimedRotatingFileHandlerTest(BaseFileTest):
suffix = prevsec.strftime(".%Y-%m-%d_%H-%M-%S") suffix = prevsec.strftime(".%Y-%m-%d_%H-%M-%S")
self.assertLogFile(self.fn + suffix) self.assertLogFile(self.fn + suffix)
def test_invalid(self):
assertRaises = self.assertRaises
assertRaises(ValueError, logging.handlers.TimedRotatingFileHandler,
self.fn, 'X')
assertRaises(ValueError, logging.handlers.TimedRotatingFileHandler,
self.fn, 'W')
assertRaises(ValueError, logging.handlers.TimedRotatingFileHandler,
self.fn, 'W7')
def secs(**kw): def secs(**kw):
return datetime.timedelta(**kw) // datetime.timedelta(seconds=1) return datetime.timedelta(**kw) // datetime.timedelta(seconds=1)