def test_content_length():
    r0 = Response('x'*10, content_length=10)

    req_head = Request.blank('/', method='HEAD')
    r1 = req_head.get_response(r0)
    eq_(r1.status_code, 200)
    eq_(r1.body, b'')
    eq_(r1.content_length, 10)

    req_get = Request.blank('/')
    r2 = req_get.get_response(r0)
    eq_(r2.status_code, 200)
    eq_(r2.body, b'x'*10)
    eq_(r2.content_length, 10)

    r3 = Response(app_iter=[b'x']*10)
    eq_(r3.content_length, None)
    eq_(r3.body, b'x'*10)
    eq_(r3.content_length, 10)

    r4 = Response(app_iter=[b'x']*10,
                  content_length=20) # wrong content_length
    eq_(r4.content_length, 20)
    assert_raises(AssertionError, lambda: r4.body)

    req_range = Request.blank('/', range=(0,5))
    r0.conditional_response = True
    r5 = req_range.get_response(r0)
    eq_(r5.status_code, 206)
    eq_(r5.body, b'xxxxx')
    eq_(r5.content_length, 5)

def create_request_from_session(url, session, timeout=DEFAULT_TIMEOUT,
                                verify=True):
    try:
        # Use session to follow redirects:
        with closing(session.head(url, allow_redirects=True,
                                  timeout=timeout,
                                  verify=verify)) as head:
            req = Request.blank(head.url)
            req.environ['webob.client.timeout'] = timeout

            # Get cookies from head:
            cookies_dict = head.cookies.get_dict()

            # Set request cookies to the head cookies:
            req.headers['Cookie'] = ','.join(name + '=' +
                                             cookies_dict[name]
                                             for name in cookies_dict)
            # Set the headers to the session headers:
            for item in head.request.headers:
                req.headers[item] = head.request.headers[item]
            return req
    except (MissingSchema, InvalidSchema):
        # Missing schema can occur in tests when the url
        # is not pointing to any resource. Simply pass.
        req = Request.blank(url)
        req.environ['webob.client.timeout'] = timeout
        return req
    except Timeout:
        raise HTTPError('Timeout')

def test_content_length():
    r0 = Response('x' * 10, content_length=10)

    req_head = Request.blank('/', method='HEAD')
    r1 = req_head.get_response(r0)
    assert r1.status_code == 200
    assert r1.body == b''
    assert r1.content_length == 10

    req_get = Request.blank('/')
    r2 = req_get.get_response(r0)
    assert r2.status_code == 200
    assert r2.body == b'x' * 10
    assert r2.content_length == 10

    r3 = Response(app_iter=[b'x'] * 10)
    assert r3.content_length is None
    assert r3.body == b'x' * 10
    assert r3.content_length == 10

    r4 = Response(app_iter=[b'x'] * 10,
                  content_length=20) # wrong content_length
    assert r4.content_length == 20
    with pytest.raises(AssertionError):
        r4.body

    req_range = Request.blank('/', range=(0, 5))
    r0.conditional_response = True
    r5 = req_range.get_response(r0)
    assert r5.status_code == 206
    assert r5.body == b'xxxxx'
    assert r5.content_length == 5

    def tapp(env, sr):
        req = Request(env)
        req = req.decode()

        v = req.POST[req.query_string]

        if hasattr(v, 'filename'):
            r = Response(text_('%s\n%r' % (v.filename, v.value)))
        else:
            r = Response(v)
        return r(env, sr)

 def tapp(env, sr):
     req = Request(env)
     #import pprint; pprint.pprint(req.environ)
     #print(req.body)
     req = req.decode()
     #import pprint; pprint.pprint(req.environ)
     #print(req.body)
     v = req.POST[req.query_string]
     if hasattr(v, 'filename'):
         r = Response(text_('%s\n%r' % (v.filename, v.value)))
     else:
         r = Response(v)
     return r(env, sr)

def test_conditional_response_if_none_match_weak():
    req = Request.blank('/', headers={'if-none-match': '"bar"'})
    req_weak = Request.blank('/', headers={'if-none-match': 'W/"bar"'})
    resp = Response(app_iter=['foo\n'], conditional_response=True, etag='bar')
    resp_weak = Response(app_iter=['foo\n'], conditional_response=True, headers={'etag': 'W/"bar"'})
    for rq in [req, req_weak]:
        for rp in [resp, resp_weak]:
            rq.get_response(rp).status_code == 304

    r2 = Response(app_iter=['foo\n'], conditional_response=True, headers={'etag': '"foo"'})
    r2_weak = Response(app_iter=['foo\n'], conditional_response=True, headers={'etag': 'W/"foo"'})
    req_weak.get_response(r2).status_code == 200
    req.get_response(r2_weak) == 200

def test_HEAD_conditional_response_range_empty_response():
    req = Request.blank('/',
        method = 'HEAD',
        range=(4,5),
    )
    res = Response('Are we not men?', conditional_response=True)
    assert req.get_response(res).body == b''

def test_environ_getter_default_fset():
    from webob.descriptors import environ_getter

    req = Request.blank("/")
    desc = environ_getter("akey", default="the_default")
    desc.fset(req, "bar")
    eq_(req.environ["akey"], "bar")

def test_upath_property_fset():
    from webob.descriptors import upath_property

    req = Request.blank("/")
    desc = upath_property("akey")
    desc.fset(req, "avalue")
    eq_(desc.fget(req), "avalue")

def test_accept_property_fdel():
    desc = accept_property("Accept-Charset", "14.2")
    req = Request.blank("/", environ={"envkey": "envval"})
    desc.fset(req, "val")
    assert desc.fget(req).header_value == "val"
    desc.fdel(req)
    eq_(type(desc.fget(req)), NilAccept)

def test_operational_error():
    app = ErrorMiddleware(myapp)
    req = Request.blank('/')
    resp = req.get_response(app, catch_exc_info=True)
    assert resp.status_code == 503
    assert 'Retry-After' in resp.headers
    assert 'accessing the database' in resp.body

def test_disconnect_detection_hinted_readline():
    data = 'abc'*(1<<20)
    req = Request.blank('/', POST=data)
    req.is_body_seekable = False
    line = req.body_file.readline(1<<16)
    assert line
    assert bytes_(data).startswith(line)

def test_environ_getter_nodefault_fget():
    from webob.descriptors import environ_getter

    req = Request.blank("/")
    desc = environ_getter("akey")
    desc.fset(req, "bar")
    assert req.environ["akey"] == "bar"

    def test_middleware_direct_call(self):
        @wsgify.middleware
        def mw(req, app):
            return "foo"

        app = mw(Response())
        self.assertEqual(app(Request.blank("/")), "foo")

def test_environ_getter_nodefault_keyerror():
    from webob.descriptors import environ_getter

    req = Request.blank("/")
    desc = environ_getter("akey")
    with pytest.raises(KeyError):
        desc.fget(req)

def test_multi_layer_dataset(multi_layer_app, temp_file):
    req = Request.blank("/")
    resp = req.get_response(multi_layer_app)
    assert resp.status == "200 OK"

    for chunk in resp.app_iter:
        temp_file.write(chunk)
    temp_file.flush()

    z = ZipFile(temp_file.name, "r", ZIP_DEFLATED)
    assert z

    # Should be 2 files for each layer
    assert len(z.namelist()) == 2 * 4
    assert "my_grid_0.asc" in z.namelist()

    # find the first asc file
    asc_filename = filter(lambda x: x.endswith(".asc"), z.namelist())[0]

    with z.open(asc_filename, "r") as f:
        data = f.read()

    assert (
        data
        == """ncols        3
nrows        2
xllcorner    -122.500000000000
yllcorner    53.000000000000
dx           -0.500000000000
dy           1.000000000000
NODATA_value  -9999
 0 1 2
 3 4 5
"""
    )

def test_no_duplicate_headers(real_data_test):
    req = Request.blank(
        "/pr+tasmax+tasmin_day_BCCA+ANUSPLIN300+ACCESS1-0_historical+rcp45_r1i1p1_19500101-21001231.h5.aig?tasmax&"
    )
    resp = req.get_response(real_data_test)
    assert resp.status == "200 OK"
    assert len(resp.headers.keys()) == len(set(resp.headers.keys()))

def test_aaigrid_response(pcic_data_portal, url):
    base = '/data/downscaled_gcms_archive/pr+tasmax+tasmin_day_'
    req = Request.blank(url.format(base))
    resp = req.get_response(pcic_data_portal)

    assert resp.status == '200 OK'
    assert resp.content_type == 'application/zip'

def test_conditional_response_if_modified_since_true():
    from datetime import datetime, timedelta
    req = Request.blank('/', if_modified_since=datetime(2011, 3, 17, 13, 0, 0))
    resp = Response(app_iter=['foo\n'], conditional_response=True,
                    last_modified=req.if_modified_since + timedelta(seconds=1))
    resp = req.get_response(resp)
    assert resp.status_code == 200

def test_climatology_bounds(pcic_data_portal):
    url = '/data/bc_prism/tmin_monClim_PRISM_historical_run1_197101-200012'\
          '.nc.nc?climatology_bounds&'
    req = Request.blank(url)
    resp = req.get_response(pcic_data_portal)

    assert resp.status == '200 OK'
    assert resp.content_type == 'application/x-netcdf'

    f = NamedTemporaryFile(suffix='.nc', delete=False)
    for block in resp.app_iter:
        f.write(block)
    f.close()

    nc = netCDF4.Dataset(f.name)

    assert 'climatology_bounds' in nc.variables

    assert_almost_equal(nc.variables['climatology_bounds'][:],
                        np.array([[0.,  10988.],
                                  [31.,  11017.],
                                  [59.,  11048.],
                                  [90.,  11078.],
                                  [120.,  11109.],
                                  [151.,  11139.],
                                  [181.,  11170.],
                                  [212.,  11201.],
                                  [243.,  11231.],
                                  [273.,  11262.],
                                  [304.,  11292.],
                                  [334.,  11323.],
                                  [0.,  11323.]], dtype=np.float32))
    nc.close()
    os.remove(f.name)