def __init__(self, text, fileclass=io.BytesIO):

if isinstance(text, str):

text = str(text).encode('ascii')

self.text = text

self.fileclass = fileclass

self.data = bytes(b'')

def sendall(self, data):

# self.data += bytes(data)

olddata = self.data

assert isinstance(olddata, bytes)

if utils.PY3:

self.data += data

else:

if isinstance(data, type(u'')): # i.e. unicode

newdata = data.encode('ascii')

elif isinstance(data, type(b'')): # native string type. FIXME!

newdata = bytes(data)

elif isinstance(data, bytes):

newdata = data

elif isinstance(data, array.array):

newdata = data.tostring()

else:

newdata = bytes(b'').join(chr(d) for d in bytes(data))

self.data += newdata

def makefile(self, mode, bufsize=None):

if mode != 'r' and mode != 'rb':

raise client.UnimplementedFileMode()

def __init__(self, text, pipe_trigger):

# When sendall() is called with pipe_trigger, raise EPIPE.

FakeSocket.__init__(self, text)

self.pipe_trigger = pipe_trigger

def sendall(self, data):

if self.pipe_trigger in data:

raise socket.error(errno.EPIPE, "gotcha")

self.data += data

def close(self):

"""Like BytesIO, but raises AssertionError on EOF.

def read(self, n=-1):

data = io.BytesIO.read(self, n)

if data == b'':

raise AssertionError('caller tried to read past EOF')

return data

def readline(self, length=None):

data = io.BytesIO.readline(self, length)

if data == b'':

raise AssertionError('caller tried to read past EOF')

def test_auto_headers(self):

# Some headers are added automatically, but should not be added by

# .request() if they are explicitly set.

class HeaderCountingBuffer(list):

def __init__(self):

self.count = {}

def append(self, item):

kv = item.split(b':')

if len(kv) > 1:

# item is a 'Key: Value' header string

lcKey = kv[0].decode('ascii').lower()

self.count.setdefault(lcKey, 0)

self.count[lcKey] += 1

list.append(self, item)

for explicit_header in True, False:

for header in 'Content-length', 'Host', 'Accept-encoding':

conn = client.HTTPConnection('example.com')

conn.sock = FakeSocket('blahblahblah')

conn._buffer = HeaderCountingBuffer()

body = 'spamspamspam'

headers = {}

if explicit_header:

headers[header] = str(len(body))

conn.request('POST', '/', body, headers)

self.assertEqual(conn._buffer.count[header.lower()], 1)

def test_content_length_0(self):

class ContentLengthChecker(list):

def __init__(self):

list.__init__(self)

self.content_length = None

def append(self, item):

kv = item.split(b':', 1)

if len(kv) > 1 and kv[0].lower() == b'content-length':

self.content_length = kv[1].strip()

list.append(self, item)

# POST with empty body

conn = client.HTTPConnection('example.com')

conn.sock = FakeSocket(None)

conn._buffer = ContentLengthChecker()

conn.request('POST', '/', '')

self.assertEqual(conn._buffer.content_length, b'0',

'Header Content-Length not set')

# PUT request with empty body

conn = client.HTTPConnection('example.com')

conn.sock = FakeSocket(None)

conn._buffer = ContentLengthChecker()

conn.request('PUT', '/', '')

self.assertEqual(conn._buffer.content_length, b'0',

'Header Content-Length not set')

def test_putheader(self):

conn = client.HTTPConnection('example.com')

conn.sock = FakeSocket(None)

conn.putrequest('GET','/')

conn.putheader('Content-length', 42)

self.assertTrue(b'Content-length: 42' in conn._buffer)

def test_ipv6host_header(self):

# Default host header on IPv6 transaction should wrapped by [] if

# its actual IPv6 address

expected = bytes(b'GET /foo HTTP/1.1\r\nHost: [2001::]:81\r\n') + \

bytes(b'Accept-Encoding: identity\r\n\r\n')

conn = client.HTTPConnection('[2001::]:81')

sock = FakeSocket('')

conn.sock = sock

conn.request('GET', '/foo')

self.assertTrue(sock.data.startswith(expected))

expected = bytes(b'GET /foo HTTP/1.1\r\nHost: [2001:102A::]\r\n') + \

bytes(b'Accept-Encoding: identity\r\n\r\n')

conn = client.HTTPConnection('[2001:102A::]')

sock = FakeSocket('')

conn.sock = sock

conn.request('GET', '/foo')

def test_status_lines(self):

# Test HTTP status lines

body = "HTTP/1.1 200 Ok\r\n\r\nText"

sock = FakeSocket(body)

resp = client.HTTPResponse(sock)

resp.begin()

self.assertEqual(resp.read(), b'Text')

self.assertTrue(resp.isclosed())

body = "HTTP/1.1 400.100 Not Ok\r\n\r\nText"

sock = FakeSocket(body)

resp = client.HTTPResponse(sock)

self.assertRaises(client.BadStatusLine, resp.begin)

def test_bad_status_repr(self):

exc = client.BadStatusLine('')

if not utils.PY3:

self.assertEqual(repr(exc), '''BadStatusLine("u\'\'",)''')

else:

self.assertEqual(repr(exc), '''BadStatusLine("\'\'",)''')

def test_partial_reads(self):

# if we have a length, the system knows when to close itself

# same behaviour than when we read the whole thing with read()

body = "HTTP/1.1 200 Ok\r\nContent-Length: 4\r\n\r\nText"

sock = FakeSocket(body)

resp = client.HTTPResponse(sock)

resp.begin()

self.assertEqual(resp.read(2), b'Te')

self.assertFalse(resp.isclosed())

self.assertEqual(resp.read(2), b'xt')

self.assertTrue(resp.isclosed())

def test_partial_reads_no_content_length(self):

# when no length is present, the socket should be gracefully closed when

# all data was read

body = "HTTP/1.1 200 Ok\r\n\r\nText"

sock = FakeSocket(body)

resp = client.HTTPResponse(sock)

resp.begin()

self.assertEqual(resp.read(2), b'Te')

self.assertFalse(resp.isclosed())

self.assertEqual(resp.read(2), b'xt')

self.assertEqual(resp.read(1), b'')

self.assertTrue(resp.isclosed())

def test_partial_reads_incomplete_body(self):

# if the server shuts down the connection before the whole

# content-length is delivered, the socket is gracefully closed

body = "HTTP/1.1 200 Ok\r\nContent-Length: 10\r\n\r\nText"

sock = FakeSocket(body)

resp = client.HTTPResponse(sock)

resp.begin()

self.assertEqual(resp.read(2), b'Te')

self.assertFalse(resp.isclosed())

self.assertEqual(resp.read(2), b'xt')

self.assertEqual(resp.read(1), b'')

self.assertTrue(resp.isclosed())

def test_host_port(self):

# Check invalid host_port

# Note that http.client does not accept user:password@ in the host-port.

for hp in ("www.python.org:abc", "user:password@www.python.org"):

self.assertRaises(client.InvalidURL, client.HTTPConnection, hp)

for hp, h, p in (("[fe80::207:e9ff:fe9b]:8000", "fe80::207:e9ff:fe9b",

8000),

("www.python.org:80", "www.python.org", 80),

("www.python.org", "www.python.org", 80),

("www.python.org:", "www.python.org", 80),

("[fe80::207:e9ff:fe9b]", "fe80::207:e9ff:fe9b", 80)):

c = client.HTTPConnection(hp)

self.assertEqual(h, c.host)

self.assertEqual(p, c.port)

def test_response_headers(self):

# test response with multiple message headers with the same field name.

text = ('HTTP/1.1 200 OK\r\n'

'Set-Cookie: Customer="WILE_E_COYOTE";'

' Version="1"; Path="/acme"\r\n'

'Set-Cookie: Part_Number="Rocket_Launcher_0001"; Version="1";'

' Path="/acme"\r\n'

'\r\n'

'No body\r\n')

hdr = ('Customer="WILE_E_COYOTE"; Version="1"; Path="/acme"'

', '

'Part_Number="Rocket_Launcher_0001"; Version="1"; Path="/acme"')

s = FakeSocket(text)

r = client.HTTPResponse(s)

r.begin()

cookies = r.getheader("Set-Cookie")

self.assertEqual(cookies, hdr)

def test_read_head(self):

# Test that the library doesn't attempt to read any data

# from a HEAD request. (Tickles SF bug #622042.)

sock = FakeSocket(

'HTTP/1.1 200 OK\r\n'

'Content-Length: 14432\r\n'

'\r\n',

NoEOFBytesIO)

resp = client.HTTPResponse(sock, method="HEAD")

resp.begin()

if resp.read():

self.fail("Did not expect response from HEAD request")

def test_send_file(self):

expected = (bytes(b'GET /foo HTTP/1.1\r\nHost: example.com\r\n') +

bytes(b'Accept-Encoding: identity\r\nContent-Length:'))

# __file__ will usually be the .pyc, i.e. binary data

with open(__file__, 'rb') as body:

conn = client.HTTPConnection('example.com')

sock = FakeSocket(body)

conn.sock = sock

conn.request('GET', '/foo', body)

self.assertTrue(sock.data.startswith(expected), '%r != %r' %

(sock.data[:len(expected)], expected))

def test_send(self):

expected = bytes(b'this is a test this is only a test')

conn = client.HTTPConnection('example.com')

sock = FakeSocket(None)

conn.sock = sock

conn.send(expected)

self.assertEqual(expected, sock.data)

sock.data = bytes(b'')

if utils.PY3:

mydata = array.array('b', expected)

else:

mydata = array.array(b'b', expected)

conn.send(mydata)

self.assertEqual(expected, sock.data)

sock.data = bytes(b'')

conn.send(io.BytesIO(expected))

self.assertEqual(expected, sock.data)

def test_chunked(self):

chunked_start = (

'HTTP/1.1 200 OK\r\n'

'Transfer-Encoding: chunked\r\n\r\n'

'a\r\n'

'hello worl\r\n'

'1\r\n'

'd\r\n'

)

sock = FakeSocket(chunked_start + '0\r\n')

resp = client.HTTPResponse(sock, method="GET")

resp.begin()

self.assertEqual(resp.read(), b'hello world')

resp.close()

for x in ('', 'foo\r\n'):

sock = FakeSocket(chunked_start + x)

resp = client.HTTPResponse(sock, method="GET")

resp.begin()

try:

resp.read()

except client.IncompleteRead as i:

self.assertEqual(i.partial, b'hello world')

self.assertEqual(repr(i),'IncompleteRead(11 bytes read)')

self.assertEqual(str(i),'IncompleteRead(11 bytes read)')

else:

self.fail('IncompleteRead expected')

finally:

resp.close()

def test_chunked_head(self):

chunked_start = (

'HTTP/1.1 200 OK\r\n'

'Transfer-Encoding: chunked\r\n\r\n'

'a\r\n'

'hello world\r\n'

'1\r\n'

'd\r\n'

)

sock = FakeSocket(chunked_start + '0\r\n')

resp = client.HTTPResponse(sock, method="HEAD")

resp.begin()

self.assertEqual(resp.read(), b'')

self.assertEqual(resp.status, 200)

self.assertEqual(resp.reason, 'OK')

self.assertTrue(resp.isclosed())

def test_negative_content_length(self):

sock = FakeSocket('HTTP/1.1 200 OK\r\n'

'Content-Length: -1\r\n\r\nHello\r\n')

resp = client.HTTPResponse(sock, method="GET")

resp.begin()

self.assertEqual(resp.read(), b'Hello\r\n')

self.assertTrue(resp.isclosed())

def test_incomplete_read(self):

sock = FakeSocket('HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\nHello\r\n')

resp = client.HTTPResponse(sock, method="GET")

resp.begin()

try:

resp.read()

except client.IncompleteRead as i:

self.assertEqual(i.partial, b'Hello\r\n')

self.assertEqual(repr(i),

"IncompleteRead(7 bytes read, 3 more expected)")

self.assertEqual(str(i),

"IncompleteRead(7 bytes read, 3 more expected)")

self.assertTrue(resp.isclosed())

else:

self.fail('IncompleteRead expected')

def test_epipe(self):

sock = EPipeSocket(

"HTTP/1.0 401 Authorization Required\r\n"

"Content-type: text/html\r\n"

"WWW-Authenticate: Basic realm=\"example\"\r\n",

b"Content-Length")

conn = client.HTTPConnection("example.com")

conn.sock = sock

self.assertRaises(socket.error,

lambda: conn.request("PUT", "/url", "body"))

resp = conn.getresponse()

self.assertEqual(401, resp.status)

self.assertEqual("Basic realm=\"example\"",

resp.getheader("www-authenticate"))

def test_filenoattr(self):

# Just test the fileno attribute in the HTTPResponse Object.

body = "HTTP/1.1 200 Ok\r\n\r\nText"

sock = FakeSocket(body)

resp = client.HTTPResponse(sock)

self.assertTrue(hasattr(resp,'fileno'),

'HTTPResponse should expose a fileno attribute')

# Test lines overflowing the max line size (_MAXLINE in http.client)

def test_overflowing_status_line(self):

self.skipTest("disabled for HTTP 0.9 support")

body = "HTTP/1.1 200 Ok" + "k" * 65536 + "\r\n"

resp = client.HTTPResponse(FakeSocket(body))

self.assertRaises((client.LineTooLong, client.BadStatusLine), resp.begin)

def test_overflowing_header_line(self):

body = (

'HTTP/1.1 200 OK\r\n'

'X-Foo: bar' + 'r' * 65536 + '\r\n\r\n'

)

resp = client.HTTPResponse(FakeSocket(body))

self.assertRaises(client.LineTooLong, resp.begin)

def test_overflowing_chunked_line(self):

body = (

'HTTP/1.1 200 OK\r\n'

'Transfer-Encoding: chunked\r\n\r\n'

+ '0' * 65536 + 'a\r\n'

'hello world\r\n'

'0\r\n'

)

resp = client.HTTPResponse(FakeSocket(body))

resp.begin()

self.assertRaises(client.LineTooLong, resp.read)

def test_early_eof(self):

# Test httpresponse with no \r\n termination,

body = "HTTP/1.1 200 Ok"

sock = FakeSocket(body)

resp = client.HTTPResponse(sock)

resp.begin()

self.assertEqual(resp.read(), b'')

def setUp(self):

self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

self.port = support.bind_port(self.serv)

self.source_port = support.find_unused_port()

self.serv.listen(5)

self.conn = None

def tearDown(self):

if self.conn:

self.conn.close()

self.conn = None

self.serv.close()

self.serv = None

def testHTTPConnectionSourceAddress(self):

self.conn = client.HTTPConnection(HOST, self.port,

source_address=('', self.source_port))

self.conn.connect()

self.assertEqual(self.conn.sock.getsockname()[1], self.source_port)

@unittest.skipIf(not hasattr(client, 'HTTPSConnection'),

def testHTTPSConnectionSourceAddress(self):

self.conn = client.HTTPSConnection(HOST, self.port,

source_address=('', self.source_port))

# We don't test anything here other the constructor not barfing as

# this code doesn't deal with setting up an active running SSL server

PORT = None

def setUp(self):

self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

TimeoutTest.PORT = support.bind_port(self.serv)

self.serv.listen(5)

def tearDown(self):

self.serv.close()

self.serv = None

def testTimeoutAttribute(self):

'''This will prove that the timeout gets through

# default -- use global socket timeout

self.assertTrue(socket.getdefaulttimeout() is None)

socket.setdefaulttimeout(30)

try:

httpConn = client.HTTPConnection(HOST, TimeoutTest.PORT)

httpConn.connect()

finally:

socket.setdefaulttimeout(None)

self.assertEqual(httpConn.sock.gettimeout(), 30)

httpConn.close()

# no timeout -- do not use global socket default

self.assertTrue(socket.getdefaulttimeout() is None)

socket.setdefaulttimeout(30)

try:

httpConn = client.HTTPConnection(HOST, TimeoutTest.PORT,

timeout=None)

httpConn.connect()

finally:

socket.setdefaulttimeout(None)

self.assertEqual(httpConn.sock.gettimeout(), None)

httpConn.close()

# a value

httpConn = client.HTTPConnection(HOST, TimeoutTest.PORT, timeout=30)

httpConn.connect()

self.assertEqual(httpConn.sock.gettimeout(), 30)

def test_attributes(self):

# simple test to check it's storing it

if hasattr(client, 'HTTPSConnection'):

h = client.HTTPSConnection(HOST, TimeoutTest.PORT, timeout=30)

self.assertEqual(h.timeout, 30)

@unittest.skipIf(not hasattr(client, 'HTTPSConnection'), 'http.client.HTTPSConnection not available')

def test_host_port(self):

# Check invalid host_port

# Note that httplib does not accept user:password@ in the host-port.

for hp in ("www.python.org:abc", "user:password@www.python.org"):

self.assertRaises(client.InvalidURL, client.HTTPSConnection, hp)

for hp, h, p in (("[fe80::207:e9ff:fe9b]:8000", "fe80::207:e9ff:fe9b",

8000),

("pypi.python.org:443", "pypi.python.org", 443),

("pypi.python.org", "pypi.python.org", 443),

("pypi.python.org:", "pypi.python.org", 443),

("[fe80::207:e9ff:fe9b]", "fe80::207:e9ff:fe9b", 443)):

c = client.HTTPSConnection(hp)

self.assertEqual(h, c.host)