def test_shared_data_middleware(tmpdir):
    def null_application(environ, start_response):
        start_response('404 NOT FOUND', [('Content-Type', 'text/plain')])
        yield b'NOT FOUND'

    test_dir = str(tmpdir)
    with open(path.join(test_dir, to_native(u'äöü', 'utf-8')), 'w') as test_file:
        test_file.write(u'FOUND')

    app = wsgi.SharedDataMiddleware(null_application, {
        '/':        path.join(path.dirname(__file__), 'res'),
        '/sources': path.join(path.dirname(__file__), 'res'),
        '/pkg':     ('werkzeug.debug', 'shared'),
        '/foo':     test_dir
    })

    for p in '/test.txt', '/sources/test.txt', '/foo/äöü':
        app_iter, status, headers = run_wsgi_app(app, create_environ(p))
        assert status == '200 OK'
        with closing(app_iter) as app_iter:
            data = b''.join(app_iter).strip()
        assert data == b'FOUND'

    app_iter, status, headers = run_wsgi_app(
        app, create_environ('/pkg/debugger.js'))
    with closing(app_iter) as app_iter:
        contents = b''.join(app_iter)
    assert b'$(function() {' in contents

    app_iter, status, headers = run_wsgi_app(
        app, create_environ('/missing'))
    assert status == '404 NOT FOUND'
    assert b''.join(app_iter).strip() == b'NOT FOUND'

    def test_shared_data_middleware(self):
        def null_application(environ, start_response):
            start_response("404 NOT FOUND", [("Content-Type", "text/plain")])
            yield b"NOT FOUND"

        app = wsgi.SharedDataMiddleware(
            null_application,
            {
                "/": path.join(path.dirname(__file__), "res"),
                "/sources": path.join(path.dirname(__file__), "res"),
                "/pkg": ("werkzeug.debug", "shared"),
            },
        )

        for p in "/test.txt", "/sources/test.txt":
            app_iter, status, headers = run_wsgi_app(app, create_environ(p))
            self.assert_equal(status, "200 OK")
            with closing(app_iter) as app_iter:
                data = b"".join(app_iter).strip()
            self.assert_equal(data, b"FOUND")

        app_iter, status, headers = run_wsgi_app(app, create_environ("/pkg/debugger.js"))
        with closing(app_iter) as app_iter:
            contents = b"".join(app_iter)
        self.assert_in(b"$(function() {", contents)

        app_iter, status, headers = run_wsgi_app(app, create_environ("/missing"))
        self.assert_equal(status, "404 NOT FOUND")
        self.assert_equal(b"".join(app_iter).strip(), b"NOT FOUND")

def test_shared_data_middleware(tmpdir):
    def null_application(environ, start_response):
        start_response("404 NOT FOUND", [("Content-Type", "text/plain")])
        yield b"NOT FOUND"

    test_dir = str(tmpdir)
    with open(path.join(test_dir, to_native(u"äöü", "utf-8")), "w") as test_file:
        test_file.write(u"FOUND")

    app = wsgi.SharedDataMiddleware(
        null_application,
        {
            "/": path.join(path.dirname(__file__), "res"),
            "/sources": path.join(path.dirname(__file__), "res"),
            "/pkg": ("werkzeug.debug", "shared"),
            "/foo": test_dir,
        },
    )

    for p in "/test.txt", "/sources/test.txt", "/foo/äöü":
        app_iter, status, headers = run_wsgi_app(app, create_environ(p))
        assert status == "200 OK"
        with closing(app_iter) as app_iter:
            data = b"".join(app_iter).strip()
        assert data == b"FOUND"

    app_iter, status, headers = run_wsgi_app(app, create_environ("/pkg/debugger.js"))
    with closing(app_iter) as app_iter:
        contents = b"".join(app_iter)
    assert b"$(function() {" in contents

    app_iter, status, headers = run_wsgi_app(app, create_environ("/missing"))
    assert status == "404 NOT FOUND"
    assert b"".join(app_iter).strip() == b"NOT FOUND"

def test_dispatchermiddleware():
    def null_application(environ, start_response):
        start_response('404 NOT FOUND', [('Content-Type', 'text/plain')])
        yield b'NOT FOUND'

    def dummy_application(environ, start_response):
        start_response('200 OK', [('Content-Type', 'text/plain')])
        yield to_bytes(environ['SCRIPT_NAME'])

    app = wsgi.DispatcherMiddleware(null_application, {
        '/test1': dummy_application,
        '/test2/very': dummy_application,
    })
    tests = {
        '/test1': ('/test1', '/test1/asfd', '/test1/very'),
        '/test2/very': ('/test2/very', '/test2/very/long/path/after/script/name')
    }
    for name, urls in tests.items():
        for p in urls:
            environ = create_environ(p)
            app_iter, status, headers = run_wsgi_app(app, environ)
            assert status == '200 OK'
            assert b''.join(app_iter).strip() == to_bytes(name)

    app_iter, status, headers = run_wsgi_app(
        app, create_environ('/missing'))
    assert status == '404 NOT FOUND'
    assert b''.join(app_iter).strip() == b'NOT FOUND'

    def test_create_environ(self):
        env = create_environ('/foo?bar=baz', 'http://example.org/')
        expected = {
            'wsgi.multiprocess':    False,
            'wsgi.version':         (1, 0),
            'wsgi.run_once':        False,
            'wsgi.errors':          sys.stderr,
            'wsgi.multithread':     False,
            'wsgi.url_scheme':      'http',
            'SCRIPT_NAME':          '',
            'CONTENT_TYPE':         '',
            'CONTENT_LENGTH':       '0',
            'SERVER_NAME':          'example.org',
            'REQUEST_METHOD':       'GET',
            'HTTP_HOST':            'example.org',
            'PATH_INFO':            '/foo',
            'SERVER_PORT':          '80',
            'SERVER_PROTOCOL':      'HTTP/1.1',
            'QUERY_STRING':         'bar=baz'
        }
        for key, value in expected.iteritems():
            assert env[key] == value
        assert env['wsgi.input'].read(0) == ''

        assert create_environ('/foo', 'http://example.com/')['SCRIPT_NAME'] == ''

    def test_shared_data_middleware(self):
        def null_application(environ, start_response):
            start_response('404 NOT FOUND', [('Content-Type', 'text/plain')])
            yield b'NOT FOUND'
        app = wsgi.SharedDataMiddleware(null_application, {
            '/':        path.join(path.dirname(__file__), 'res'),
            '/sources': path.join(path.dirname(__file__), 'res'),
            '/pkg':     ('werkzeug.debug', 'shared')
        })

        for p in '/test.txt', '/sources/test.txt':
            app_iter, status, headers = run_wsgi_app(app, create_environ(p))
            self.assert_equal(status, '200 OK')
            with closing(app_iter) as app_iter:
                data = b''.join(app_iter).strip()
            self.assert_equal(data, b'FOUND')

        app_iter, status, headers = run_wsgi_app(
            app, create_environ('/pkg/debugger.js'))
        with closing(app_iter) as app_iter:
            contents = b''.join(app_iter)
        self.assert_in(b'$(function() {', contents)

        app_iter, status, headers = run_wsgi_app(
            app, create_environ('/missing'))
        self.assert_equal(status, '404 NOT FOUND')
        self.assert_equal(b''.join(app_iter).strip(), b'NOT FOUND')

def test_204_and_1XX_response_has_no_content_length():
    response = wrappers.Response(status=204)
    assert response.content_length is None

    headers = response.get_wsgi_headers(create_environ())
    assert 'Content-Length' not in headers

    response = wrappers.Response(status=100)
    assert response.content_length is None

    headers = response.get_wsgi_headers(create_environ())
    assert 'Content-Length' not in headers

def test_follow_external_redirect_on_same_subdomain():
    env = create_environ('/', base_url='http://example.com')
    c = Client(external_subdomain_redirect_demo_app, allow_subdomain_redirects=True)
    c.get(environ_overrides=env, follow_redirects=True)

    # check that this does not work for real external domains
    env = create_environ('/', base_url='http://localhost')
    assert_raises(RuntimeError, lambda: c.get(environ_overrides=env, follow_redirects=True))

    # check that subdomain redirects fail if no `allow_subdomain_redirects` is applied
    c = Client(external_subdomain_redirect_demo_app)
    assert_raises(RuntimeError, lambda: c.get(environ_overrides=env, follow_redirects=True))

    def test_server_name_interpolation(self):
        server_name = "example.invalid"
        map = r.Map([r.Rule("/", endpoint="index"), r.Rule("/", endpoint="alt", subdomain="alt")])

        env = create_environ("/", "http://%s/" % server_name)
        adapter = map.bind_to_environ(env, server_name=server_name)
        assert adapter.match() == ("index", {})

        env = create_environ("/", "http://alt.%s/" % server_name)
        adapter = map.bind_to_environ(env, server_name=server_name)
        assert adapter.match() == ("alt", {})

        env = create_environ("/", "http://%s/" % server_name)
        adapter = map.bind_to_environ(env, server_name="foo")
        assert adapter.subdomain == "<invalid>"

def test_server_name_casing():
    m = r.Map([r.Rule("/", endpoint="index", subdomain="foo")])

    env = create_environ()
    env["SERVER_NAME"] = env["HTTP_HOST"] = "FOO.EXAMPLE.COM"
    a = m.bind_to_environ(env, server_name="example.com")
    assert a.match("/") == ("index", {})

    env = create_environ()
    env["SERVER_NAME"] = "127.0.0.1"
    env["SERVER_PORT"] = "5000"
    del env["HTTP_HOST"]
    a = m.bind_to_environ(env, server_name="example.com")
    with pytest.raises(r.NotFound):
        a.match()

def test_range_request_without_complete_length():
    env = create_environ()
    response = wrappers.Response('Hello World')
    env['HTTP_RANGE'] = 'bytes=-'
    response.make_conditional(env, accept_ranges=True, complete_length=None)
    assert response.status_code == 200
    assert response.data == b'Hello World'

def test_wsgi_app_exception(app, monkeypatch):
    match = pretend.stub(match=pretend.call_recorder(lambda: ("warehouse.fake.view", {})))
    urls = pretend.stub(bind_to_environ=pretend.call_recorder(lambda e: match))
    response = pretend.call_recorder(lambda e, s: None)

    class FakeException(HTTPException):

        # @pretend.call_recorder
        def __call__(self, *args, **kwargs):
            return response

    @pretend.call_recorder
    def fake_view(*args, **kwargs):
        raise FakeException("An error has occurred")

    fake_module = pretend.stub(view=fake_view)
    import_module = pretend.call_recorder(lambda mod: fake_module)

    monkeypatch.setattr(importlib, "import_module", import_module)

    environ = create_environ()
    start_response = pretend.stub()

    app.urls = urls

    app.wsgi_app(environ, start_response)

    assert match.match.calls == [pretend.call()]
    assert urls.bind_to_environ.calls == [pretend.call(environ)]
    assert import_module.calls == [pretend.call("warehouse.fake")]
    assert fake_view.calls == [pretend.call(app, mock.ANY)]

def test_external_building_with_port_bind_to_environ_wrong_servername():
    """Test external URL building with port number (map.bind_to_environ) with wrong server name raises ValueError"""
    map = Map([
        Rule('/', endpoint='index'),
    ])
    environ = create_environ('/', 'http://example.org:5000/')
    assert_raises(ValueError, lambda: map.bind_to_environ(environ, server_name="example.org"))

 def test_generate_response_with_only_accept_header(self):
     env = create_environ(headers={'Accept': 'application/json'})
     req = Request(env)
     resp = utils.generate_response(request=req)
     self.assertEqual(resp.status_code, 200)
     self.assertEqual(resp.status, '200 OK')
     self.assertEqual(resp.content_type, 'application/json')

 def test_generate_response_with_valid_header_and_invalid_extension(self):
     env = create_environ(headers={'Accept': 'application/json'})
     req = Request(env)
     resp = utils.generate_response(request=req, data=None, \
         format='foo')
     self.assertEqual(resp.status_code, 406)
     self.assertEqual(resp.status, '406 NOT ACCEPTABLE')

def test_etag_response_mixin():
    """ETag response-wrapper mixin"""
    response = Response('Hello World')
    assert response.get_etag() == (None, None)
    response.add_etag()
    assert response.get_etag() == ('b10a8db164e0754105b7a99be72e3fe5', False)
    assert not response.cache_control
    response.cache_control.must_revalidate = True
    response.cache_control.max_age = 60
    response.headers['Content-Length'] = len(response.data)
    assert response.headers['Cache-Control'] == 'must-revalidate, max-age=60'

    assert 'date' not in response.headers
    env = create_environ()
    env.update({
        'REQUEST_METHOD':       'GET',
        'HTTP_IF_NONE_MATCH':   response.get_etag()[0]
    })
    response.make_conditional(env)
    assert 'date' in response.headers

    # after the thing is invoked by the server as wsgi application
    # (we're emulating this here), there must not be any entity
    # headers left and the status code would have to be 304
    resp = Response.from_app(response, env)
    assert resp.status_code == 304
    assert not 'content-length' in resp.headers

    # make sure date is not overriden
    response = Response('Hello World')
    response.date = 1337
    d = response.date
    response.make_conditional(env)
    assert response.date == d

def test_package(serial, md5_hash, monkeypatch):
    safe_join = pretend.call_recorder(
        lambda *a, **k: "/tmp/packages/any/t/test-1.0.tar.gz"
    )
    _fp = pretend.stub(__enter__=lambda: None, __exit__=lambda *a: None)
    _open = pretend.call_recorder(lambda *a, **k: _fp)
    wrap_file = lambda *a, **k: None
    mtime = pretend.call_recorder(lambda f: 123457)
    getsize = pretend.call_recorder(lambda f: 54321)

    monkeypatch.setattr(simple, "safe_join", safe_join)
    monkeypatch.setattr(simple, "open", _open, raising=False)
    monkeypatch.setattr(simple, "wrap_file", wrap_file)
    monkeypatch.setattr(os.path, "getmtime", mtime)
    monkeypatch.setattr(os.path, "getsize", getsize)

    gpff = pretend.call_recorder(lambda p: "test")
    get_md5 = pretend.call_recorder(
        lambda p: md5_hash
    )
    get_last_serial = pretend.call_recorder(lambda p: serial)

    app = pretend.stub(
        config=pretend.stub(
            cache=pretend.stub(browser=False, varnish=False),
            paths=pretend.stub(packages="/tmp"),
        ),
        models=pretend.stub(
            packaging=pretend.stub(
                get_project_for_filename=gpff,
                get_filename_md5=get_md5,
                get_last_serial=get_last_serial,
            ),
        ),
    )
    request = pretend.stub(environ=create_environ())

    resp = simple.package(app, request, path="packages/any/t/test-1.0.tar.gz")

    if serial:
        assert resp.headers["X-PyPI-Last-Serial"] == str(serial)
    else:
        assert "X-PyPI-Last-Serial" not in resp.headers

    assert resp.headers["Surrogate-Key"] == "package package~test"
    assert resp.headers["Content-Length"] == "54321"

    assert safe_join.calls == [
        pretend.call("/tmp", "packages/any/t/test-1.0.tar.gz"),
    ]
    assert _open.calls == [
        pretend.call("/tmp/packages/any/t/test-1.0.tar.gz", "rb"),
    ]
    assert mtime.calls == [pretend.call("/tmp/packages/any/t/test-1.0.tar.gz")]
    assert getsize.calls == [
        pretend.call("/tmp/packages/any/t/test-1.0.tar.gz"),
    ]
    assert gpff.calls == [pretend.call("test-1.0.tar.gz")]
    assert get_md5.calls == [pretend.call("test-1.0.tar.gz")]
    assert get_last_serial.calls == [pretend.call("test")]

 def test_path_info_from_request_uri_fix(self):
     app = fixers.PathInfoFromRequestUriFix(path_check_app)
     for key in 'REQUEST_URI', 'REQUEST_URL', 'UNENCODED_URL':
         env = dict(create_environ(), SCRIPT_NAME='/test', PATH_INFO='/?????')
         env[key] = '/test/foo%25bar?drop=this'
         response = Response.from_app(app, env)
         assert response.get_data() == b'PATH_INFO: /foo%bar\nSCRIPT_NAME: /test'

    def test_proxy_fix(self):
        @Request.application
        def app(request):
            return Response('%s|%s' % (
                request.remote_addr,
                # do not use request.host as this fixes too :)
                request.environ['HTTP_HOST']
            ))
        app = fixers.ProxyFix(app, num_proxies=2)
        environ = dict(
            create_environ(),
            HTTP_X_FORWARDED_PROTO="https",
            HTTP_X_FORWARDED_HOST='example.com',
            HTTP_X_FORWARDED_FOR='1.2.3.4, 5.6.7.8',
            REMOTE_ADDR='127.0.0.1',
            HTTP_HOST='fake'
        )

        response = Response.from_app(app, environ)

        assert response.get_data() == b'1.2.3.4|example.com'

        # And we must check that if it is a redirection it is
        # correctly done:

        redirect_app = redirect('/foo/bar.hml')
        response = Response.from_app(redirect_app, environ)

        wsgi_headers = response.get_wsgi_headers(environ)
        assert wsgi_headers['Location'] == 'https://example.com/foo/bar.hml'

 def test_environ_nonascii_pathinfo(self):
     environ = create_environ("/лошадь")
     m = r.Map([r.Rule("/", endpoint="index"), r.Rule(u"/лошадь", endpoint="horse")])
     a = m.bind_to_environ(environ)
     self.assert_strict_equal(a.match("/"), ("index", {}))
     self.assert_strict_equal(a.match("/лошадь"), ("horse", {}))
     self.assert_raises(r.NotFound, a.match, "/барсук")