Back to index

python3.2  3.2.2
test_httplib.py
Go to the documentation of this file.
00001 import errno
00002 from http import client
00003 import io
00004 import os
00005 import array
00006 import socket
00007 
00008 import unittest
00009 TestCase = unittest.TestCase
00010 
00011 from test import support
00012 
00013 here = os.path.dirname(__file__)
00014 # Self-signed cert file for 'localhost'
00015 CERT_localhost = os.path.join(here, 'keycert.pem')
00016 # Self-signed cert file for 'fakehostname'
00017 CERT_fakehostname = os.path.join(here, 'keycert2.pem')
00018 # Root cert file (CA) for svn.python.org's cert
00019 CACERT_svn_python_org = os.path.join(here, 'https_svn_python_org_root.pem')
00020 
00021 HOST = support.HOST
00022 
00023 class FakeSocket:
00024     def __init__(self, text, fileclass=io.BytesIO):
00025         if isinstance(text, str):
00026             text = text.encode("ascii")
00027         self.text = text
00028         self.fileclass = fileclass
00029         self.data = b''
00030 
00031     def sendall(self, data):
00032         self.data += data
00033 
00034     def makefile(self, mode, bufsize=None):
00035         if mode != 'r' and mode != 'rb':
00036             raise client.UnimplementedFileMode()
00037         return self.fileclass(self.text)
00038 
00039 class EPipeSocket(FakeSocket):
00040 
00041     def __init__(self, text, pipe_trigger):
00042         # When sendall() is called with pipe_trigger, raise EPIPE.
00043         FakeSocket.__init__(self, text)
00044         self.pipe_trigger = pipe_trigger
00045 
00046     def sendall(self, data):
00047         if self.pipe_trigger in data:
00048             raise socket.error(errno.EPIPE, "gotcha")
00049         self.data += data
00050 
00051     def close(self):
00052         pass
00053 
00054 class NoEOFStringIO(io.BytesIO):
00055     """Like StringIO, but raises AssertionError on EOF.
00056 
00057     This is used below to test that http.client doesn't try to read
00058     more from the underlying file than it should.
00059     """
00060     def read(self, n=-1):
00061         data = io.BytesIO.read(self, n)
00062         if data == b'':
00063             raise AssertionError('caller tried to read past EOF')
00064         return data
00065 
00066     def readline(self, length=None):
00067         data = io.BytesIO.readline(self, length)
00068         if data == b'':
00069             raise AssertionError('caller tried to read past EOF')
00070         return data
00071 
00072 class HeaderTests(TestCase):
00073     def test_auto_headers(self):
00074         # Some headers are added automatically, but should not be added by
00075         # .request() if they are explicitly set.
00076 
00077         class HeaderCountingBuffer(list):
00078             def __init__(self):
00079                 self.count = {}
00080             def append(self, item):
00081                 kv = item.split(b':')
00082                 if len(kv) > 1:
00083                     # item is a 'Key: Value' header string
00084                     lcKey = kv[0].decode('ascii').lower()
00085                     self.count.setdefault(lcKey, 0)
00086                     self.count[lcKey] += 1
00087                 list.append(self, item)
00088 
00089         for explicit_header in True, False:
00090             for header in 'Content-length', 'Host', 'Accept-encoding':
00091                 conn = client.HTTPConnection('example.com')
00092                 conn.sock = FakeSocket('blahblahblah')
00093                 conn._buffer = HeaderCountingBuffer()
00094 
00095                 body = 'spamspamspam'
00096                 headers = {}
00097                 if explicit_header:
00098                     headers[header] = str(len(body))
00099                 conn.request('POST', '/', body, headers)
00100                 self.assertEqual(conn._buffer.count[header.lower()], 1)
00101 
00102     def test_putheader(self):
00103         conn = client.HTTPConnection('example.com')
00104         conn.sock = FakeSocket(None)
00105         conn.putrequest('GET','/')
00106         conn.putheader('Content-length', 42)
00107         self.assertTrue(b'Content-length: 42' in conn._buffer)
00108 
00109     def test_ipv6host_header(self):
00110         # Default host header on IPv6 transaction should wrapped by [] if
00111         # its actual IPv6 address
00112         expected = b'GET /foo HTTP/1.1\r\nHost: [2001::]:81\r\n' \
00113                    b'Accept-Encoding: identity\r\n\r\n'
00114         conn = client.HTTPConnection('[2001::]:81')
00115         sock = FakeSocket('')
00116         conn.sock = sock
00117         conn.request('GET', '/foo')
00118         self.assertTrue(sock.data.startswith(expected))
00119 
00120         expected = b'GET /foo HTTP/1.1\r\nHost: [2001:102A::]\r\n' \
00121                    b'Accept-Encoding: identity\r\n\r\n'
00122         conn = client.HTTPConnection('[2001:102A::]')
00123         sock = FakeSocket('')
00124         conn.sock = sock
00125         conn.request('GET', '/foo')
00126         self.assertTrue(sock.data.startswith(expected))
00127 
00128 
00129 class BasicTest(TestCase):
00130     def test_status_lines(self):
00131         # Test HTTP status lines
00132 
00133         body = "HTTP/1.1 200 Ok\r\n\r\nText"
00134         sock = FakeSocket(body)
00135         resp = client.HTTPResponse(sock)
00136         resp.begin()
00137         self.assertEqual(resp.read(), b"Text")
00138         self.assertTrue(resp.isclosed())
00139 
00140         body = "HTTP/1.1 400.100 Not Ok\r\n\r\nText"
00141         sock = FakeSocket(body)
00142         resp = client.HTTPResponse(sock)
00143         self.assertRaises(client.BadStatusLine, resp.begin)
00144 
00145     def test_bad_status_repr(self):
00146         exc = client.BadStatusLine('')
00147         self.assertEqual(repr(exc), '''BadStatusLine("\'\'",)''')
00148 
00149     def test_partial_reads(self):
00150         # if we have a lenght, the system knows when to close itself
00151         # same behaviour than when we read the whole thing with read()
00152         body = "HTTP/1.1 200 Ok\r\nContent-Length: 4\r\n\r\nText"
00153         sock = FakeSocket(body)
00154         resp = client.HTTPResponse(sock)
00155         resp.begin()
00156         self.assertEqual(resp.read(2), b'Te')
00157         self.assertFalse(resp.isclosed())
00158         self.assertEqual(resp.read(2), b'xt')
00159         self.assertTrue(resp.isclosed())
00160 
00161     def test_host_port(self):
00162         # Check invalid host_port
00163 
00164         for hp in ("www.python.org:abc", "www.python.org:"):
00165             self.assertRaises(client.InvalidURL, client.HTTPConnection, hp)
00166 
00167         for hp, h, p in (("[fe80::207:e9ff:fe9b]:8000",
00168                           "fe80::207:e9ff:fe9b", 8000),
00169                          ("www.python.org:80", "www.python.org", 80),
00170                          ("www.python.org", "www.python.org", 80),
00171                          ("[fe80::207:e9ff:fe9b]", "fe80::207:e9ff:fe9b", 80)):
00172             c = client.HTTPConnection(hp)
00173             self.assertEqual(h, c.host)
00174             self.assertEqual(p, c.port)
00175 
00176     def test_response_headers(self):
00177         # test response with multiple message headers with the same field name.
00178         text = ('HTTP/1.1 200 OK\r\n'
00179                 'Set-Cookie: Customer="WILE_E_COYOTE"; '
00180                 'Version="1"; Path="/acme"\r\n'
00181                 'Set-Cookie: Part_Number="Rocket_Launcher_0001"; Version="1";'
00182                 ' Path="/acme"\r\n'
00183                 '\r\n'
00184                 'No body\r\n')
00185         hdr = ('Customer="WILE_E_COYOTE"; Version="1"; Path="/acme"'
00186                ', '
00187                'Part_Number="Rocket_Launcher_0001"; Version="1"; Path="/acme"')
00188         s = FakeSocket(text)
00189         r = client.HTTPResponse(s)
00190         r.begin()
00191         cookies = r.getheader("Set-Cookie")
00192         self.assertEqual(cookies, hdr)
00193 
00194     def test_read_head(self):
00195         # Test that the library doesn't attempt to read any data
00196         # from a HEAD request.  (Tickles SF bug #622042.)
00197         sock = FakeSocket(
00198             'HTTP/1.1 200 OK\r\n'
00199             'Content-Length: 14432\r\n'
00200             '\r\n',
00201             NoEOFStringIO)
00202         resp = client.HTTPResponse(sock, method="HEAD")
00203         resp.begin()
00204         if resp.read():
00205             self.fail("Did not expect response from HEAD request")
00206 
00207     def test_send_file(self):
00208         expected = (b'GET /foo HTTP/1.1\r\nHost: example.com\r\n'
00209                     b'Accept-Encoding: identity\r\nContent-Length:')
00210 
00211         with open(__file__, 'rb') as body:
00212             conn = client.HTTPConnection('example.com')
00213             sock = FakeSocket(body)
00214             conn.sock = sock
00215             conn.request('GET', '/foo', body)
00216             self.assertTrue(sock.data.startswith(expected), '%r != %r' %
00217                     (sock.data[:len(expected)], expected))
00218 
00219     def test_send(self):
00220         expected = b'this is a test this is only a test'
00221         conn = client.HTTPConnection('example.com')
00222         sock = FakeSocket(None)
00223         conn.sock = sock
00224         conn.send(expected)
00225         self.assertEqual(expected, sock.data)
00226         sock.data = b''
00227         conn.send(array.array('b', expected))
00228         self.assertEqual(expected, sock.data)
00229         sock.data = b''
00230         conn.send(io.BytesIO(expected))
00231         self.assertEqual(expected, sock.data)
00232 
00233     def test_send_iter(self):
00234         expected = b'GET /foo HTTP/1.1\r\nHost: example.com\r\n' \
00235                    b'Accept-Encoding: identity\r\nContent-Length: 11\r\n' \
00236                    b'\r\nonetwothree'
00237 
00238         def body():
00239             yield b"one"
00240             yield b"two"
00241             yield b"three"
00242 
00243         conn = client.HTTPConnection('example.com')
00244         sock = FakeSocket("")
00245         conn.sock = sock
00246         conn.request('GET', '/foo', body(), {'Content-Length': '11'})
00247         self.assertEqual(sock.data, expected)
00248 
00249     def test_send_type_error(self):
00250         # See: Issue #12676
00251         conn = client.HTTPConnection('example.com')
00252         conn.sock = FakeSocket('')
00253         with self.assertRaises(TypeError):
00254             conn.request('POST', 'test', conn)
00255 
00256     def test_chunked(self):
00257         chunked_start = (
00258             'HTTP/1.1 200 OK\r\n'
00259             'Transfer-Encoding: chunked\r\n\r\n'
00260             'a\r\n'
00261             'hello worl\r\n'
00262             '1\r\n'
00263             'd\r\n'
00264         )
00265         sock = FakeSocket(chunked_start + '0\r\n')
00266         resp = client.HTTPResponse(sock, method="GET")
00267         resp.begin()
00268         self.assertEqual(resp.read(), b'hello world')
00269         resp.close()
00270 
00271         for x in ('', 'foo\r\n'):
00272             sock = FakeSocket(chunked_start + x)
00273             resp = client.HTTPResponse(sock, method="GET")
00274             resp.begin()
00275             try:
00276                 resp.read()
00277             except client.IncompleteRead as i:
00278                 self.assertEqual(i.partial, b'hello world')
00279                 self.assertEqual(repr(i),'IncompleteRead(11 bytes read)')
00280                 self.assertEqual(str(i),'IncompleteRead(11 bytes read)')
00281             else:
00282                 self.fail('IncompleteRead expected')
00283             finally:
00284                 resp.close()
00285 
00286     def test_chunked_head(self):
00287         chunked_start = (
00288             'HTTP/1.1 200 OK\r\n'
00289             'Transfer-Encoding: chunked\r\n\r\n'
00290             'a\r\n'
00291             'hello world\r\n'
00292             '1\r\n'
00293             'd\r\n'
00294         )
00295         sock = FakeSocket(chunked_start + '0\r\n')
00296         resp = client.HTTPResponse(sock, method="HEAD")
00297         resp.begin()
00298         self.assertEqual(resp.read(), b'')
00299         self.assertEqual(resp.status, 200)
00300         self.assertEqual(resp.reason, 'OK')
00301         self.assertTrue(resp.isclosed())
00302 
00303     def test_negative_content_length(self):
00304         sock = FakeSocket(
00305             'HTTP/1.1 200 OK\r\nContent-Length: -1\r\n\r\nHello\r\n')
00306         resp = client.HTTPResponse(sock, method="GET")
00307         resp.begin()
00308         self.assertEqual(resp.read(), b'Hello\r\n')
00309         resp.close()
00310 
00311     def test_incomplete_read(self):
00312         sock = FakeSocket('HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\nHello\r\n')
00313         resp = client.HTTPResponse(sock, method="GET")
00314         resp.begin()
00315         try:
00316             resp.read()
00317         except client.IncompleteRead as i:
00318             self.assertEqual(i.partial, b'Hello\r\n')
00319             self.assertEqual(repr(i),
00320                              "IncompleteRead(7 bytes read, 3 more expected)")
00321             self.assertEqual(str(i),
00322                              "IncompleteRead(7 bytes read, 3 more expected)")
00323         else:
00324             self.fail('IncompleteRead expected')
00325         finally:
00326             resp.close()
00327 
00328     def test_epipe(self):
00329         sock = EPipeSocket(
00330             "HTTP/1.0 401 Authorization Required\r\n"
00331             "Content-type: text/html\r\n"
00332             "WWW-Authenticate: Basic realm=\"example\"\r\n",
00333             b"Content-Length")
00334         conn = client.HTTPConnection("example.com")
00335         conn.sock = sock
00336         self.assertRaises(socket.error,
00337                           lambda: conn.request("PUT", "/url", "body"))
00338         resp = conn.getresponse()
00339         self.assertEqual(401, resp.status)
00340         self.assertEqual("Basic realm=\"example\"",
00341                          resp.getheader("www-authenticate"))
00342 
00343     # Test lines overflowing the max line size (_MAXLINE in http.client)
00344 
00345     def test_overflowing_status_line(self):
00346         body = "HTTP/1.1 200 Ok" + "k" * 65536 + "\r\n"
00347         resp = client.HTTPResponse(FakeSocket(body))
00348         self.assertRaises((client.LineTooLong, client.BadStatusLine), resp.begin)
00349 
00350     def test_overflowing_header_line(self):
00351         body = (
00352             'HTTP/1.1 200 OK\r\n'
00353             'X-Foo: bar' + 'r' * 65536 + '\r\n\r\n'
00354         )
00355         resp = client.HTTPResponse(FakeSocket(body))
00356         self.assertRaises(client.LineTooLong, resp.begin)
00357 
00358     def test_overflowing_chunked_line(self):
00359         body = (
00360             'HTTP/1.1 200 OK\r\n'
00361             'Transfer-Encoding: chunked\r\n\r\n'
00362             + '0' * 65536 + 'a\r\n'
00363             'hello world\r\n'
00364             '0\r\n'
00365         )
00366         resp = client.HTTPResponse(FakeSocket(body))
00367         resp.begin()
00368         self.assertRaises(client.LineTooLong, resp.read)
00369 
00370 class OfflineTest(TestCase):
00371     def test_responses(self):
00372         self.assertEqual(client.responses[client.NOT_FOUND], "Not Found")
00373 
00374 
00375 class SourceAddressTest(TestCase):
00376     def setUp(self):
00377         self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
00378         self.port = support.bind_port(self.serv)
00379         self.source_port = support.find_unused_port()
00380         self.serv.listen(5)
00381         self.conn = None
00382 
00383     def tearDown(self):
00384         if self.conn:
00385             self.conn.close()
00386             self.conn = None
00387         self.serv.close()
00388         self.serv = None
00389 
00390     def testHTTPConnectionSourceAddress(self):
00391         self.conn = client.HTTPConnection(HOST, self.port,
00392                 source_address=('', self.source_port))
00393         self.conn.connect()
00394         self.assertEqual(self.conn.sock.getsockname()[1], self.source_port)
00395 
00396     @unittest.skipIf(not hasattr(client, 'HTTPSConnection'),
00397                      'http.client.HTTPSConnection not defined')
00398     def testHTTPSConnectionSourceAddress(self):
00399         self.conn = client.HTTPSConnection(HOST, self.port,
00400                 source_address=('', self.source_port))
00401         # We don't test anything here other the constructor not barfing as
00402         # this code doesn't deal with setting up an active running SSL server
00403         # for an ssl_wrapped connect() to actually return from.
00404 
00405 
00406 class TimeoutTest(TestCase):
00407     PORT = None
00408 
00409     def setUp(self):
00410         self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
00411         TimeoutTest.PORT = support.bind_port(self.serv)
00412         self.serv.listen(5)
00413 
00414     def tearDown(self):
00415         self.serv.close()
00416         self.serv = None
00417 
00418     def testTimeoutAttribute(self):
00419         # This will prove that the timeout gets through HTTPConnection
00420         # and into the socket.
00421 
00422         # default -- use global socket timeout
00423         self.assertTrue(socket.getdefaulttimeout() is None)
00424         socket.setdefaulttimeout(30)
00425         try:
00426             httpConn = client.HTTPConnection(HOST, TimeoutTest.PORT)
00427             httpConn.connect()
00428         finally:
00429             socket.setdefaulttimeout(None)
00430         self.assertEqual(httpConn.sock.gettimeout(), 30)
00431         httpConn.close()
00432 
00433         # no timeout -- do not use global socket default
00434         self.assertTrue(socket.getdefaulttimeout() is None)
00435         socket.setdefaulttimeout(30)
00436         try:
00437             httpConn = client.HTTPConnection(HOST, TimeoutTest.PORT,
00438                                               timeout=None)
00439             httpConn.connect()
00440         finally:
00441             socket.setdefaulttimeout(None)
00442         self.assertEqual(httpConn.sock.gettimeout(), None)
00443         httpConn.close()
00444 
00445         # a value
00446         httpConn = client.HTTPConnection(HOST, TimeoutTest.PORT, timeout=30)
00447         httpConn.connect()
00448         self.assertEqual(httpConn.sock.gettimeout(), 30)
00449         httpConn.close()
00450 
00451 
00452 class HTTPSTest(TestCase):
00453 
00454     def setUp(self):
00455         if not hasattr(client, 'HTTPSConnection'):
00456             self.skipTest('ssl support required')
00457 
00458     def make_server(self, certfile):
00459         from test.ssl_servers import make_https_server
00460         return make_https_server(self, certfile)
00461 
00462     def test_attributes(self):
00463         # simple test to check it's storing the timeout
00464         h = client.HTTPSConnection(HOST, TimeoutTest.PORT, timeout=30)
00465         self.assertEqual(h.timeout, 30)
00466 
00467     def _check_svn_python_org(self, resp):
00468         # Just a simple check that everything went fine
00469         server_string = resp.getheader('server')
00470         self.assertIn('Apache', server_string)
00471 
00472     def test_networked(self):
00473         # Default settings: no cert verification is done
00474         support.requires('network')
00475         with support.transient_internet('svn.python.org'):
00476             h = client.HTTPSConnection('svn.python.org', 443)
00477             h.request('GET', '/')
00478             resp = h.getresponse()
00479             self._check_svn_python_org(resp)
00480 
00481     def test_networked_good_cert(self):
00482         # We feed a CA cert that validates the server's cert
00483         import ssl
00484         support.requires('network')
00485         with support.transient_internet('svn.python.org'):
00486             context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
00487             context.verify_mode = ssl.CERT_REQUIRED
00488             context.load_verify_locations(CACERT_svn_python_org)
00489             h = client.HTTPSConnection('svn.python.org', 443, context=context)
00490             h.request('GET', '/')
00491             resp = h.getresponse()
00492             self._check_svn_python_org(resp)
00493 
00494     def test_networked_bad_cert(self):
00495         # We feed a "CA" cert that is unrelated to the server's cert
00496         import ssl
00497         support.requires('network')
00498         with support.transient_internet('svn.python.org'):
00499             context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
00500             context.verify_mode = ssl.CERT_REQUIRED
00501             context.load_verify_locations(CERT_localhost)
00502             h = client.HTTPSConnection('svn.python.org', 443, context=context)
00503             with self.assertRaises(ssl.SSLError):
00504                 h.request('GET', '/')
00505 
00506     def test_local_good_hostname(self):
00507         # The (valid) cert validates the HTTP hostname
00508         import ssl
00509         from test.ssl_servers import make_https_server
00510         server = make_https_server(self, CERT_localhost)
00511         context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
00512         context.verify_mode = ssl.CERT_REQUIRED
00513         context.load_verify_locations(CERT_localhost)
00514         h = client.HTTPSConnection('localhost', server.port, context=context)
00515         h.request('GET', '/nonexistent')
00516         resp = h.getresponse()
00517         self.assertEqual(resp.status, 404)
00518 
00519     def test_local_bad_hostname(self):
00520         # The (valid) cert doesn't validate the HTTP hostname
00521         import ssl
00522         from test.ssl_servers import make_https_server
00523         server = make_https_server(self, CERT_fakehostname)
00524         context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
00525         context.verify_mode = ssl.CERT_REQUIRED
00526         context.load_verify_locations(CERT_fakehostname)
00527         h = client.HTTPSConnection('localhost', server.port, context=context)
00528         with self.assertRaises(ssl.CertificateError):
00529             h.request('GET', '/')
00530         # Same with explicit check_hostname=True
00531         h = client.HTTPSConnection('localhost', server.port, context=context,
00532                                    check_hostname=True)
00533         with self.assertRaises(ssl.CertificateError):
00534             h.request('GET', '/')
00535         # With check_hostname=False, the mismatching is ignored
00536         h = client.HTTPSConnection('localhost', server.port, context=context,
00537                                    check_hostname=False)
00538         h.request('GET', '/nonexistent')
00539         resp = h.getresponse()
00540         self.assertEqual(resp.status, 404)
00541 
00542 
00543 class RequestBodyTest(TestCase):
00544     """Test cases where a request includes a message body."""
00545 
00546     def setUp(self):
00547         self.conn = client.HTTPConnection('example.com')
00548         self.conn.sock = self.sock = FakeSocket("")
00549         self.conn.sock = self.sock
00550 
00551     def get_headers_and_fp(self):
00552         f = io.BytesIO(self.sock.data)
00553         f.readline()  # read the request line
00554         message = client.parse_headers(f)
00555         return message, f
00556 
00557     def test_manual_content_length(self):
00558         # Set an incorrect content-length so that we can verify that
00559         # it will not be over-ridden by the library.
00560         self.conn.request("PUT", "/url", "body",
00561                           {"Content-Length": "42"})
00562         message, f = self.get_headers_and_fp()
00563         self.assertEqual("42", message.get("content-length"))
00564         self.assertEqual(4, len(f.read()))
00565 
00566     def test_ascii_body(self):
00567         self.conn.request("PUT", "/url", "body")
00568         message, f = self.get_headers_and_fp()
00569         self.assertEqual("text/plain", message.get_content_type())
00570         self.assertIsNone(message.get_charset())
00571         self.assertEqual("4", message.get("content-length"))
00572         self.assertEqual(b'body', f.read())
00573 
00574     def test_latin1_body(self):
00575         self.conn.request("PUT", "/url", "body\xc1")
00576         message, f = self.get_headers_and_fp()
00577         self.assertEqual("text/plain", message.get_content_type())
00578         self.assertIsNone(message.get_charset())
00579         self.assertEqual("5", message.get("content-length"))
00580         self.assertEqual(b'body\xc1', f.read())
00581 
00582     def test_bytes_body(self):
00583         self.conn.request("PUT", "/url", b"body\xc1")
00584         message, f = self.get_headers_and_fp()
00585         self.assertEqual("text/plain", message.get_content_type())
00586         self.assertIsNone(message.get_charset())
00587         self.assertEqual("5", message.get("content-length"))
00588         self.assertEqual(b'body\xc1', f.read())
00589 
00590     def test_file_body(self):
00591         with open(support.TESTFN, "w") as f:
00592             f.write("body")
00593         with open(support.TESTFN) as f:
00594             self.conn.request("PUT", "/url", f)
00595             message, f = self.get_headers_and_fp()
00596             self.assertEqual("text/plain", message.get_content_type())
00597             self.assertIsNone(message.get_charset())
00598             self.assertEqual("4", message.get("content-length"))
00599             self.assertEqual(b'body', f.read())
00600 
00601     def test_binary_file_body(self):
00602         with open(support.TESTFN, "wb") as f:
00603             f.write(b"body\xc1")
00604         with open(support.TESTFN, "rb") as f:
00605             self.conn.request("PUT", "/url", f)
00606             message, f = self.get_headers_and_fp()
00607             self.assertEqual("text/plain", message.get_content_type())
00608             self.assertIsNone(message.get_charset())
00609             self.assertEqual("5", message.get("content-length"))
00610             self.assertEqual(b'body\xc1', f.read())
00611 
00612 
00613 class HTTPResponseTest(TestCase):
00614 
00615     def setUp(self):
00616         body = "HTTP/1.1 200 Ok\r\nMy-Header: first-value\r\nMy-Header: \
00617                 second-value\r\n\r\nText"
00618         sock = FakeSocket(body)
00619         self.resp = client.HTTPResponse(sock)
00620         self.resp.begin()
00621 
00622     def test_getting_header(self):
00623         header = self.resp.getheader('My-Header')
00624         self.assertEqual(header, 'first-value, second-value')
00625 
00626         header = self.resp.getheader('My-Header', 'some default')
00627         self.assertEqual(header, 'first-value, second-value')
00628 
00629     def test_getting_nonexistent_header_with_string_default(self):
00630         header = self.resp.getheader('No-Such-Header', 'default-value')
00631         self.assertEqual(header, 'default-value')
00632 
00633     def test_getting_nonexistent_header_with_iterable_default(self):
00634         header = self.resp.getheader('No-Such-Header', ['default', 'values'])
00635         self.assertEqual(header, 'default, values')
00636 
00637         header = self.resp.getheader('No-Such-Header', ('default', 'values'))
00638         self.assertEqual(header, 'default, values')
00639 
00640     def test_getting_nonexistent_header_without_default(self):
00641         header = self.resp.getheader('No-Such-Header')
00642         self.assertEqual(header, None)
00643 
00644     def test_getting_header_defaultint(self):
00645         header = self.resp.getheader('No-Such-Header',default=42)
00646         self.assertEqual(header, 42)
00647 
00648 def test_main(verbose=None):
00649     support.run_unittest(HeaderTests, OfflineTest, BasicTest, TimeoutTest,
00650                          HTTPSTest, RequestBodyTest, SourceAddressTest,
00651                          HTTPResponseTest)
00652 
00653 if __name__ == '__main__':
00654     test_main()