| import time |
| |
| import pytest |
| |
| from webob import Request, Response |
| from webob.dec import wsgify |
| from webob.client import SendRequest |
| |
| |
| @wsgify |
| def simple_app(req): |
| data = {'headers': dict(req.headers), |
| 'body': req.text, |
| 'method': req.method, |
| } |
| return Response(json=data) |
| |
| @pytest.mark.usefixtures("serve") |
| def test_client(serve, client_app=None): |
| with serve(simple_app) as server: |
| req = Request.blank(server.url, method='POST', content_type='application/json', |
| json={'test': 1}) |
| resp = req.send(client_app) |
| assert resp.status_code == 200, resp.status |
| assert resp.json['headers']['Content-Type'] == 'application/json' |
| assert resp.json['method'] == 'POST' |
| # Test that these values get filled in: |
| del req.environ['SERVER_NAME'] |
| del req.environ['SERVER_PORT'] |
| resp = req.send(client_app) |
| assert resp.status_code == 200, resp.status |
| req = Request.blank(server.url) |
| del req.environ['SERVER_NAME'] |
| del req.environ['SERVER_PORT'] |
| assert req.send(client_app).status_code == 200 |
| req.headers['Host'] = server.url.lstrip('http://') |
| del req.environ['SERVER_NAME'] |
| del req.environ['SERVER_PORT'] |
| resp = req.send(client_app) |
| assert resp.status_code == 200, resp.status |
| del req.environ['SERVER_NAME'] |
| del req.environ['SERVER_PORT'] |
| del req.headers['Host'] |
| assert req.environ.get('SERVER_NAME') is None |
| assert req.environ.get('SERVER_PORT') is None |
| assert req.environ.get('HTTP_HOST') is None |
| with pytest.raises(ValueError): |
| req.send(client_app) |
| req = Request.blank(server.url) |
| req.environ['CONTENT_LENGTH'] = 'not a number' |
| assert req.send(client_app).status_code == 200 |
| |
| |
| def no_length_app(environ, start_response): |
| start_response('200 OK', [('Content-type', 'text/plain')]) |
| return [b'ok'] |
| |
| |
| @pytest.mark.usefixtures("serve") |
| def test_no_content_length(serve, client_app=None): |
| with serve(no_length_app) as server: |
| req = Request.blank(server.url) |
| resp = req.send(client_app) |
| assert resp.status_code == 200, resp.status |
| |
| |
| @wsgify |
| def cookie_app(req): |
| resp = Response('test') |
| resp.headers.add('Set-Cookie', 'a=b') |
| resp.headers.add('Set-Cookie', 'c=d') |
| resp.headerlist.append(('X-Crazy', 'value\r\n continuation')) |
| return resp |
| |
| |
| @pytest.mark.usefixtures("serve") |
| def test_client_cookies(serve, client_app=None): |
| with serve(cookie_app) as server: |
| req = Request.blank(server.url + '/?test') |
| resp = req.send(client_app) |
| assert resp.headers.getall('Set-Cookie') == ['a=b', 'c=d'] |
| assert resp.headers['X-Crazy'] == 'value, continuation', repr(resp.headers['X-Crazy']) |
| |
| |
| @wsgify |
| def slow_app(req): |
| time.sleep(2) |
| return Response('ok') |
| |
| |
| @pytest.mark.usefixtures("serve") |
| def test_client_slow(serve, client_app=None): |
| if client_app is None: |
| client_app = SendRequest() |
| if not client_app._timeout_supported(client_app.HTTPConnection): |
| # timeout isn't supported |
| return |
| with serve(slow_app) as server: |
| req = Request.blank(server.url) |
| req.environ['webob.client.timeout'] = 0.1 |
| resp = req.send(client_app) |
| assert resp.status_code == 504, resp.status |