def __init__(self, headers=[], url=None):

"""

import email

# The email.message_from_string is available on both Py2.7 and Py3.3

self._headers = email.message_from_string("\n".join(headers))

self._url = url

"""Perform a single request / response cycle, returning Cookie: header."""

req = urllib_request.Request(url)

cookiejar.add_cookie_header(req)

cookie_hdr = req.get_header("Cookie", "")

headers = []

for hdr in set_cookie_hdrs:

headers.append("%s: %s" % (hdr_name, hdr))

res = FakeResponse(headers, url)

cookiejar.extract_cookies(res, req)

def test___all__(self):

# Verify which names are exposed

for module in 'request', 'response', 'parse', 'error', 'robotparser':

context = {}

exec('from future.backports.urllib.%s import *' % module, context)

del context['__builtins__']

if module == 'request' and os.name == 'nt':

u, p = context.pop('url2pathname'), context.pop('pathname2url')

self.assertEqual(u.__module__, 'nturl2path')

self.assertEqual(p.__module__, 'nturl2path')

for k, v in context.items():

self.assertEqual(v.__module__, 'future.backports.urllib.%s' % module,

"%r is exposed in 'future.backports.urllib.%s' but defined in %r" %

(k, module, v.__module__))

def test_trivial(self):

# A couple trivial tests

self.assertRaises(ValueError, urllib_request.urlopen, 'bogus url')

# XXX Name hacking to get this to work on Windows.

fname = os.path.abspath(urllib_request.__file__).replace('\\', '/')

if os.name == 'nt':

file_url = "file:///%s" % fname

else:

file_url = "file://%s" % fname

f = urllib_request.urlopen(file_url)

f.read()

f.close()

def test_parse_http_list(self):

tests = [

('a,b,c', ['a', 'b', 'c']),

('path"o,l"og"i"cal, example', ['path"o,l"og"i"cal', 'example']),

('a, b, "c", "d", "e,f", g, h',

['a', 'b', '"c"', '"d"', '"e,f"', 'g', 'h']),

('a="b\\"c", d="e\\,f", g="h\\\\i"',

['a="b"c"', 'd="e,f"', 'g="h\\i"'])]

for string, list in tests:

self.assertEqual(urllib_request.parse_http_list(string), list)

def test_URLError_reasonstr(self):

err = urllib_error.URLError('reason')

def test_request_headers_dict(self):

"""

url = "http://example.com"

self.assertEqual(Request(url,

headers={"Spam-eggs": "blah"}

).headers["Spam-eggs"], "blah")

self.assertEqual(Request(url,

headers={"spam-EggS": "blah"}

).headers["Spam-eggs"], "blah")

def test_request_headers_methods(self):

"""

url = "http://example.com"

req = Request(url, headers={"Spam-eggs": "blah"})

self.assertTrue(req.has_header("Spam-eggs"))

self.assertEqual(req.header_items(), [('Spam-eggs', 'blah')])

req.add_header("Foo-Bar", "baz")

self.assertEqual(sorted(req.header_items()),

[('Foo-bar', 'baz'), ('Spam-eggs', 'blah')])

self.assertFalse(req.has_header("Not-there"))

self.assertIsNone(req.get_header("Not-there"))

self.assertEqual(req.get_header("Not-there", "default"), "default")

def test_password_manager(self):

mgr = urllib_request.HTTPPasswordMgr()

add = mgr.add_password

find_user_pass = mgr.find_user_password

add("Some Realm", "http://example.com/", "joe", "password")

add("Some Realm", "http://example.com/ni", "ni", "ni")

add("c", "http://example.com/foo", "foo", "ni")

add("c", "http://example.com/bar", "bar", "nini")

add("b", "http://example.com/", "first", "blah")

add("b", "http://example.com/", "second", "spam")

add("a", "http://example.com", "1", "a")

add("Some Realm", "http://c.example.com:3128", "3", "c")

add("Some Realm", "d.example.com", "4", "d")

add("Some Realm", "e.example.com:3128", "5", "e")

self.assertEqual(find_user_pass("Some Realm", "example.com"),

('joe', 'password'))

#self.assertEqual(find_user_pass("Some Realm", "http://example.com/ni"),

# ('ni', 'ni'))

self.assertEqual(find_user_pass("Some Realm", "http://example.com"),

('joe', 'password'))

self.assertEqual(find_user_pass("Some Realm", "http://example.com/"),

('joe', 'password'))

self.assertEqual(

find_user_pass("Some Realm", "http://example.com/spam"),

('joe', 'password'))

self.assertEqual(

find_user_pass("Some Realm", "http://example.com/spam/spam"),

('joe', 'password'))

self.assertEqual(find_user_pass("c", "http://example.com/foo"),

('foo', 'ni'))

self.assertEqual(find_user_pass("c", "http://example.com/bar"),

('bar', 'nini'))

self.assertEqual(find_user_pass("b", "http://example.com/"),

('second', 'spam'))

# No special relationship between a.example.com and example.com:

self.assertEqual(find_user_pass("a", "http://example.com/"),

('1', 'a'))

self.assertEqual(find_user_pass("a", "http://a.example.com/"),

(None, None))

# Ports:

self.assertEqual(find_user_pass("Some Realm", "c.example.com"),

(None, None))

self.assertEqual(find_user_pass("Some Realm", "c.example.com:3128"),

('3', 'c'))

self.assertEqual(

find_user_pass("Some Realm", "http://c.example.com:3128"),

('3', 'c'))

self.assertEqual(find_user_pass("Some Realm", "d.example.com"),

('4', 'd'))

self.assertEqual(find_user_pass("Some Realm", "e.example.com:3128"),

('5', 'e'))

def test_password_manager_default_port(self):

"""

mgr = urllib_request.HTTPPasswordMgr()

add = mgr.add_password

find_user_pass = mgr.find_user_password

add("f", "http://g.example.com:80", "10", "j")

add("g", "http://h.example.com", "11", "k")

add("h", "i.example.com:80", "12", "l")

add("i", "j.example.com", "13", "m")

self.assertEqual(find_user_pass("f", "g.example.com:100"),

(None, None))

self.assertEqual(find_user_pass("f", "g.example.com:80"),

('10', 'j'))

self.assertEqual(find_user_pass("f", "g.example.com"),

(None, None))

self.assertEqual(find_user_pass("f", "http://g.example.com:100"),

(None, None))

self.assertEqual(find_user_pass("f", "http://g.example.com:80"),

('10', 'j'))

self.assertEqual(find_user_pass("f", "http://g.example.com"),

('10', 'j'))

self.assertEqual(find_user_pass("g", "h.example.com"), ('11', 'k'))

self.assertEqual(find_user_pass("g", "h.example.com:80"), ('11', 'k'))

self.assertEqual(find_user_pass("g", "http://h.example.com:80"),

('11', 'k'))

self.assertEqual(find_user_pass("h", "i.example.com"), (None, None))

self.assertEqual(find_user_pass("h", "i.example.com:80"), ('12', 'l'))

self.assertEqual(find_user_pass("h", "http://i.example.com:80"),

('12', 'l'))

self.assertEqual(find_user_pass("i", "j.example.com"), ('13', 'm'))

self.assertEqual(find_user_pass("i", "j.example.com:80"),

(None, None))

self.assertEqual(find_user_pass("i", "http://j.example.com"),

('13', 'm'))

self.assertEqual(find_user_pass("i", "http://j.example.com:80"),

addheaders = []

def open(self, req, data=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT):

self.req, self.data, self.timeout = req, data, timeout

def error(self, proto, *args):

def __init__(self, code, msg, headers, data, url=None):

io.StringIO.__init__(self, data)

self.code, self.msg, self.headers, self.url = code, msg, headers, url

def info(self):

return self.headers

def geturl(self):

def add_cookie_header(self, request):

self.ach_req = request

def extract_cookies(self, response, request):

def __init__(self, meth_name, action, handle):

self.meth_name = meth_name

self.handle = handle

self.action = action

def __call__(self, *args):

def __init__(self, fp, msg, status, reason):

self.fp = fp

self.msg = msg

self.status = status

self.reason = reason

self.code = 200

def read(self):

return ''

def info(self):

return {}

def geturl(self):

def __init__(self):

self.level = 0

self.req_headers = []

self.data = None

self.raise_on_endheaders = False

self.sock = None

self._tunnel_headers = {}

def __call__(self, host, timeout=socket._GLOBAL_DEFAULT_TIMEOUT):

self.host = host

self.timeout = timeout

return self

def set_debuglevel(self, level):

self.level = level

def set_tunnel(self, host, port=None, headers=None):

self._tunnel_host = host

self._tunnel_port = port

if headers:

self._tunnel_headers = headers

else:

self._tunnel_headers.clear()

def request(self, method, url, body=None, headers=None):

self.method = method

self.selector = url

if headers is not None:

self.req_headers += headers.items()

self.req_headers.sort()

if body:

self.data = body

if self.raise_on_endheaders:

import socket

raise socket.error()

def getresponse(self):

return MockHTTPResponse(MockFile(), {}, 200, "OK")

def close(self):

# useful for testing handler machinery

# see add_ordered_mock_handlers() docstring

handler_order = 500

def __init__(self, methods):

self._define_methods(methods)

def _define_methods(self, methods):

for spec in methods:

if len(spec) == 2: name, action = spec

else: name, action = spec, None

meth = FakeMethod(name, action, self.handle)

setattr(self.__class__, name, meth)

def handle(self, fn_name, action, *args, **kwds):

self.parent.calls.append((self, fn_name, args, kwds))

if action is None:

return None

elif action == "return self":

return self

elif action == "return response":

res = MockResponse(200, "OK", {}, "")

return res

elif action == "return request":

return Request("http://blah/")

elif action.startswith("error"):

code = action[action.rfind(" ")+1:]

try:

code = int(code)

except ValueError:

pass

res = MockResponse(200, "OK", {}, "")

return self.parent.error("http", args[0], res, code, "", {})

elif action == "raise":

raise urllib_error.URLError("blah")

assert False

def close(self): pass

def add_parent(self, parent):

self.parent = parent

self.parent.calls = []

def __lt__(self, other):

if not hasattr(other, "handler_order"):

# No handler_order, leave in original order. Yuck.

return True

"""Create MockHandlers and add them to an OpenerDirector.

handlers = []

count = 0

for meths in meth_spec:

class MockHandlerSubclass(MockHandler): pass

h = MockHandlerSubclass(meths)

h.handler_order += count

h.add_parent(opener)

count = count + 1

handlers.append(h)

opener.add_handler(h)

opener = OpenerDirector()

for h in handler_instances:

opener.add_handler(h)

# useful for testing redirections and auth

# sends supplied headers and code as first response

# sends 200 OK as second response

def __init__(self, code, headers):

self.code = code

self.headers = headers

self.reset()

def reset(self):

self._count = 0

self.requests = []

def http_open(self, req):

import future.backports.email as email

import copy

self.requests.append(copy.deepcopy(req))

if self._count == 0:

self._count = self._count + 1

name = http.client.responses[self.code]

msg = email.message_from_string(self.headers)

return self.parent.error(

"http", req, MockFile(), self.code, name, msg)

else:

self.req = req

msg = email.message_from_string("\r\n\r\n")

# Useful for testing the Proxy-Authorization request by verifying the

# properties of httpcon

def __init__(self):

urllib_request.AbstractHTTPHandler.__init__(self)

self.httpconn = MockHTTPClass()

def https_open(self, req):

def add_password(self, realm, uri, user, password):

self.realm = realm

self.url = uri

self.user = user

self.password = password

def find_user_password(self, realm, authuri):

self.target_realm = realm

self.target_url = authuri

def test_add_non_handler(self):

class NonHandler(object):

pass

self.assertRaises(TypeError,

OpenerDirector().add_handler, NonHandler())

def test_badly_named_methods(self):

# test work-around for three methods that accidentally follow the

# naming conventions for handler methods

# (*_open() / *_request() / *_response())

# These used to call the accidentally-named methods, causing a

# TypeError in real code; here, returning self from these mock

# methods would either cause no exception, or AttributeError.

from future.backports.urllib.error import URLError

o = OpenerDirector()

meth_spec = [

[("do_open", "return self"), ("proxy_open", "return self")],

[("redirect_request", "return self")],

]

add_ordered_mock_handlers(o, meth_spec)

o.add_handler(urllib_request.UnknownHandler())

for scheme in "do", "proxy", "redirect":

self.assertRaises(URLError, o.open, scheme+"://example.com/")

def test_handled(self):

# handler returning non-None means no more handlers will be called

o = OpenerDirector()

meth_spec = [

["http_open", "ftp_open", "http_error_302"],

["ftp_open"],

[("http_open", "return self")],

[("http_open", "return self")],

]

handlers = add_ordered_mock_handlers(o, meth_spec)

req = Request("http://example.com/")

r = o.open(req)

# Second .http_open() gets called, third doesn't, since second returned

# non-None. Handlers without .http_open() never get any methods called

# on them.

# In fact, second mock handler defining .http_open() returns self

# (instead of response), which becomes the OpenerDirector's return

# value.

self.assertEqual(r, handlers[2])

calls = [(handlers[0], "http_open"), (handlers[2], "http_open")]

for expected, got in zip(calls, o.calls):

handler, name, args, kwds = got

self.assertEqual((handler, name), expected)

self.assertEqual(args, (req,))

def test_handler_order(self):

o = OpenerDirector()

handlers = []

for meths, handler_order in [

([("http_open", "return self")], 500),

(["http_open"], 0),

]:

class MockHandlerSubclass(MockHandler): pass

h = MockHandlerSubclass(meths)

h.handler_order = handler_order

handlers.append(h)

o.add_handler(h)

o.open("http://example.com/")

# handlers called in reverse order, thanks to their sort order

self.assertEqual(o.calls[0][0], handlers[1])

self.assertEqual(o.calls[1][0], handlers[0])

def test_raise(self):

# raising URLError stops processing of request

o = OpenerDirector()

meth_spec = [

[("http_open", "raise")],

[("http_open", "return self")],

]

handlers = add_ordered_mock_handlers(o, meth_spec)

req = Request("http://example.com/")

self.assertRaises(urllib_error.URLError, o.open, req)

self.assertEqual(o.calls, [(handlers[0], "http_open", (req,), {})])

def test_http_error(self):

# XXX http_error_default

# http errors are a special case

o = OpenerDirector()

meth_spec = [

[("http_open", "error 302")],

[("http_error_400", "raise"), "http_open"],

[("http_error_302", "return response"), "http_error_303",

"http_error"],

[("http_error_302")],

]

handlers = add_ordered_mock_handlers(o, meth_spec)

class Unknown(object):

def __eq__(self, other): return True

req = Request("http://example.com/")

o.open(req)

assert len(o.calls) == 2

calls = [(handlers[0], "http_open", (req,)),

(handlers[2], "http_error_302",

(req, Unknown(), 302, "", {}))]

for expected, got in zip(calls, o.calls):

handler, method_name, args = expected

self.assertEqual((handler, method_name), got[:2])

self.assertEqual(args, got[2])

def test_processors(self):

# *_request / *_response methods get called appropriately

o = OpenerDirector()

meth_spec = [

[("http_request", "return request"),

("http_response", "return response")],

[("http_request", "return request"),

("http_response", "return response")],

]

handlers = add_ordered_mock_handlers(o, meth_spec)

req = Request("http://example.com/")

o.open(req)

# processor methods are called on *all* handlers that define them,

# not just the first handler that handles the request

calls = [

(handlers[0], "http_request"), (handlers[1], "http_request"),

(handlers[0], "http_response"), (handlers[1], "http_response")]

for i, (handler, name, args, kwds) in enumerate(o.calls):

if i < 2:

# *_request

self.assertEqual((handler, name), calls[i])

self.assertEqual(len(args), 1)

self.assertIsInstance(args[0], Request)

else:

# *_response

self.assertEqual((handler, name), calls[i])

self.assertEqual(len(args), 2)

self.assertIsInstance(args[0], Request)

# response from opener.open is None, because there's no

# handler that defines http_open to handle it

self.assertTrue(args[1] is None or

isinstance(args[1], MockResponse))

def test_method_deprecations(self):

req = Request("http://www.example.com")

with self.assertWarns(DeprecationWarning):

req.add_data("data")

with self.assertWarns(DeprecationWarning):

req.get_data()

with self.assertWarns(DeprecationWarning):

req.has_data()

with self.assertWarns(DeprecationWarning):

req.get_host()

with self.assertWarns(DeprecationWarning):

req.get_selector()

with self.assertWarns(DeprecationWarning):

req.is_unverifiable()

with self.assertWarns(DeprecationWarning):

req.get_origin_req_host()

with self.assertWarns(DeprecationWarning):

try:

path.encode("utf-8")

except UnicodeEncodeError:

raise unittest.SkipTest("path is not encodable to utf8")

urlpath = urllib_request.pathname2url(path)

if os.name == "nt" and urlpath.startswith("///"):

urlpath = urlpath[2:]

# XXX don't ask me about the mac...

def test_ftp(self):

class MockFTPWrapper(object):

def __init__(self, data): self.data = data

def retrfile(self, filename, filetype):

self.filename, self.filetype = filename, filetype

return io.StringIO(self.data), len(self.data)

def close(self): pass

class NullFTPHandler(urllib_request.FTPHandler):

def __init__(self, data): self.data = data

def connect_ftp(self, user, passwd, host, port, dirs,

timeout=socket._GLOBAL_DEFAULT_TIMEOUT):

self.user, self.passwd = user, passwd

self.host, self.port = host, port

self.dirs = dirs

self.ftpwrapper = MockFTPWrapper(self.data)

return self.ftpwrapper

import ftplib

data = "rheum rhaponicum"

h = NullFTPHandler(data)

h.parent = MockOpener()

for url, host, port, user, passwd, type_, dirs, filename, mimetype in [

("ftp://localhost/foo/bar/baz.html",

"localhost", ftplib.FTP_PORT, "", "", "I",

["foo", "bar"], "baz.html", "text/html"),

("ftp://parrot@localhost/foo/bar/baz.html",

"localhost", ftplib.FTP_PORT, "parrot", "", "I",

["foo", "bar"], "baz.html", "text/html"),

("ftp://%25parrot@localhost/foo/bar/baz.html",

"localhost", ftplib.FTP_PORT, "%parrot", "", "I",

["foo", "bar"], "baz.html", "text/html"),

("ftp://%2542parrot@localhost/foo/bar/baz.html",

"localhost", ftplib.FTP_PORT, "%42parrot", "", "I",

["foo", "bar"], "baz.html", "text/html"),

("ftp://localhost:80/foo/bar/",

"localhost", 80, "", "", "D",

["foo", "bar"], "", None),

("ftp://localhost/baz.gif;type=a",

"localhost", ftplib.FTP_PORT, "", "", "A",

[], "baz.gif", None), # XXX really this should guess image/gif

]:

req = Request(url)

req.timeout = None

r = h.ftp_open(req)

# ftp authentication not yet implemented by FTPHandler

self.assertEqual(h.user, user)

self.assertEqual(h.passwd, passwd)

self.assertEqual(h.host, socket.gethostbyname(host))

self.assertEqual(h.port, port)

self.assertEqual(h.dirs, dirs)

self.assertEqual(h.ftpwrapper.filename, filename)

self.assertEqual(h.ftpwrapper.filetype, type_)

headers = r.info()

self.assertEqual(headers.get("Content-type"), mimetype)

self.assertEqual(int(headers["Content-length"]), len(data))

def test_file(self):

import future.backports.email.utils as email_utils

import socket

h = urllib_request.FileHandler()

o = h.parent = MockOpener()

TESTFN = support.TESTFN

urlpath = sanepathname2url(os.path.abspath(TESTFN))

towrite = b"hello, world\n"

urls = [

"file://localhost%s" % urlpath,

"file://%s" % urlpath,

"file://%s%s" % (socket.gethostbyname('localhost'), urlpath),

]

try:

localaddr = socket.gethostbyname(socket.gethostname())

except socket.gaierror:

localaddr = ''

if localaddr:

urls.append("file://%s%s" % (localaddr, urlpath))

for url in urls:

f = open(TESTFN, "wb")

try:

try:

f.write(towrite)

finally:

f.close()

r = h.file_open(Request(url))

try:

data = r.read()

headers = r.info()

respurl = r.geturl()

finally:

r.close()

stats = os.stat(TESTFN)

modified = email_utils.formatdate(stats.st_mtime, usegmt=True)

finally:

os.remove(TESTFN)

self.assertEqual(data, towrite)

self.assertEqual(headers["Content-type"], "text/plain")

self.assertEqual(headers["Content-length"], "13")

self.assertEqual(headers["Last-modified"], modified)

self.assertEqual(respurl, url)

for url in [

"file://localhost:80%s" % urlpath,

"file:///file_does_not_exist.txt",

"file://%s:80%s/%s" % (socket.gethostbyname('localhost'),

os.getcwd(), TESTFN),

"file://somerandomhost.ontheinternet.com%s/%s" %

(os.getcwd(), TESTFN),

]:

try:

f = open(TESTFN, "wb")

try:

f.write(towrite)

finally:

f.close()

self.assertRaises(urllib_error.URLError,

h.file_open, Request(url))

finally:

os.remove(TESTFN)

h = urllib_request.FileHandler()

o = h.parent = MockOpener()

# XXXX why does // mean ftp (and /// mean not ftp!), and where

# is file: scheme specified? I think this is really a bug, and

# what was intended was to distinguish between URLs like:

# file:/blah.txt (a file)

# file://localhost/blah.txt (a file)

# file:///blah.txt (a file)

# file://ftp.example.com/blah.txt (an ftp URL)

for url, ftp in [

("file://ftp.example.com//foo.txt", False),

("file://ftp.example.com///foo.txt", False),

("file://ftp.example.com/foo.txt", False),

("file://somehost//foo/something.txt", False),

("file://localhost//foo/something.txt", False),

]:

req = Request(url)

try:

h.file_open(req)

# XXXX remove OSError when bug fixed

except (urllib_error.URLError, OSError):

self.assertFalse(ftp)

else:

self.assertIs(o.req, req)

self.assertEqual(req.type, "ftp")

self.assertEqual(req.type == "ftp", ftp)

@skip26

def test_http(self):

h = urllib_request.AbstractHTTPHandler()

o = h.parent = MockOpener()

url = "http://example.com/"

for method, data in [("GET", None), ("POST", b"blah")]:

req = Request(url, data, {"Foo": "bar"})

req.timeout = None

req.add_unredirected_header("Spam", "eggs")

http = MockHTTPClass()

r = h.do_open(http, req)

# result attributes

r.read; r.readline # wrapped MockFile methods

r.info; r.geturl # addinfourl methods

r.code, r.msg == 200, "OK" # added from MockHTTPClass.getreply()

hdrs = r.info()

hdrs.get; hdrs.__contains__ # r.info() gives dict from .getreply()

self.assertEqual(r.geturl(), url)

self.assertEqual(http.host, "example.com")

self.assertEqual(http.level, 0)

self.assertEqual(http.method, method)

self.assertEqual(http.selector, "/")

self.assertEqual(http.req_headers,

[("Connection", "close"),

("Foo", "bar"), ("Spam", "eggs")])

self.assertEqual(http.data, data)

# check socket.error converted to URLError

http.raise_on_endheaders = True

self.assertRaises(urllib_error.URLError, h.do_open, http, req)

# Check for TypeError on POST data which is str.

req = Request("http://example.com/","badpost")

self.assertRaises(TypeError, h.do_request_, req)

# check adding of standard headers

o.addheaders = [("Spam", "eggs")]

for data in b"", None: # POST, GET

req = Request("http://example.com/", data)

r = MockResponse(200, "OK", {}, "")

newreq = h.do_request_(req)

if data is None: # GET

self.assertNotIn("Content-length", req.unredirected_hdrs)

self.assertNotIn("Content-type", req.unredirected_hdrs)

else: # POST

self.assertEqual(req.unredirected_hdrs["Content-length"], "0")

self.assertEqual(req.unredirected_hdrs["Content-type"],

"application/x-www-form-urlencoded")

# XXX the details of Host could be better tested

self.assertEqual(req.unredirected_hdrs["Host"], "example.com")

self.assertEqual(req.unredirected_hdrs["Spam"], "eggs")

# don't clobber existing headers

req.add_unredirected_header("Content-length", "foo")

req.add_unredirected_header("Content-type", "bar")

req.add_unredirected_header("Host", "baz")

req.add_unredirected_header("Spam", "foo")

newreq = h.do_request_(req)

self.assertEqual(req.unredirected_hdrs["Content-length"], "foo")

self.assertEqual(req.unredirected_hdrs["Content-type"], "bar")

self.assertEqual(req.unredirected_hdrs["Host"], "baz")

self.assertEqual(req.unredirected_hdrs["Spam"], "foo")

# Check iterable body support

def iterable_body():

yield b"one"

yield b"two"

yield b"three"

for headers in {}, {"Content-Length": 11}:

req = Request("http://example.com/", iterable_body(), headers)

if not headers:

# Having an iterable body without a Content-Length should

# raise an exception

self.assertRaises(ValueError, h.do_request_, req)

else:

newreq = h.do_request_(req)

# A file object.

# Test only Content-Length attribute of request.

file_obj = io.BytesIO()

file_obj.write(b"Something\nSomething\nSomething\n")

for headers in {}, {"Content-Length": 30}:

req = Request("http://example.com/", file_obj, headers)

if not headers:

# Having an iterable body without a Content-Length should

# raise an exception

self.assertRaises(ValueError, h.do_request_, req)

else:

newreq = h.do_request_(req)

self.assertEqual(int(newreq.get_header('Content-length')),30)

file_obj.close()

# array.array Iterable - Content Length is calculated

iterable_array = array.array(text_to_native_str("I"),

[1,2,3,4])

for headers in {}, {"Content-Length": 16}:

req = Request("http://example.com/", iterable_array, headers)

newreq = h.do_request_(req)

self.assertEqual(int(newreq.get_header('Content-length')),16)

@skip26

def test_http_doubleslash(self):

# Checks the presence of any unnecessary double slash in url does not

# break anything. Previously, a double slash directly after the host

# could cause incorrect parsing.

h = urllib_request.AbstractHTTPHandler()

h.parent = MockOpener()

data = b""

ds_urls = [

"http://example.com/foo/bar/baz.html",

"http://example.com//foo/bar/baz.html",

"http://example.com/foo//bar/baz.html",

"http://example.com/foo/bar//baz.html"

]

for ds_url in ds_urls:

ds_req = Request(ds_url, data)

# Check whether host is determined correctly if there is no proxy

np_ds_req = h.do_request_(ds_req)

self.assertEqual(np_ds_req.unredirected_hdrs["Host"],"example.com")

# Check whether host is determined correctly if there is a proxy

ds_req.set_proxy("someproxy:3128",None)

p_ds_req = h.do_request_(ds_req)

self.assertEqual(p_ds_req.unredirected_hdrs["Host"],"example.com")

def test_fixpath_in_weirdurls(self):

# Issue4493: urllib2 to supply '/' when to urls where path does not

# start with'/'

h = urllib_request.AbstractHTTPHandler()

h.parent = MockOpener()

weird_url = 'http://www.python.org?getspam'

req = Request(weird_url)

newreq = h.do_request_(req)

self.assertEqual(newreq.host,'www.python.org')

self.assertEqual(newreq.selector,'/?getspam')

url_without_path = 'http://www.python.org'

req = Request(url_without_path)

newreq = h.do_request_(req)

self.assertEqual(newreq.host,'www.python.org')

self.assertEqual(newreq.selector,'')

def test_errors(self):

h = urllib_request.HTTPErrorProcessor()

o = h.parent = MockOpener()

url = "http://example.com/"

req = Request(url)

# all 2xx are passed through

r = MockResponse(200, "OK", {}, "", url)

newr = h.http_response(req, r)

self.assertIs(r, newr)

self.assertFalse(hasattr(o, "proto")) # o.error not called

r = MockResponse(202, "Accepted", {}, "", url)

newr = h.http_response(req, r)

self.assertIs(r, newr)

self.assertFalse(hasattr(o, "proto")) # o.error not called

r = MockResponse(206, "Partial content", {}, "", url)

newr = h.http_response(req, r)

self.assertIs(r, newr)

self.assertFalse(hasattr(o, "proto")) # o.error not called

# anything else calls o.error (and MockOpener returns None, here)