aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--test/test_language.py253
-rw-r--r--test/test_language_http.py260
2 files changed, 260 insertions, 253 deletions
diff --git a/test/test_language.py b/test/test_language.py
index f72e677b..159a6d5d 100644
--- a/test/test_language.py
+++ b/test/test_language.py
@@ -4,8 +4,6 @@ from libpathod import language
from libpathod.language import base, http, websockets, writer, exceptions
import tutils
-language.TESTING = True
-
def parse_request(s):
return language.parse_requests(s)[0]
@@ -275,11 +273,6 @@ class TestMisc:
v3 = v2.freeze({})
assert v2.value.val == v3.value.val
- def test_internal_response(self):
- d = cStringIO.StringIO()
- s = http.make_error_response("foo")
- language.serve(s, d, {})
-
class TestHeaders:
def test_header(self):
@@ -479,132 +472,6 @@ class TestPauses:
assert l.freeze({}).spec() == l.spec()
-class TestRequest:
- def test_nonascii(self):
- tutils.raises("ascii", parse_request, "get:\xf0")
-
- def test_err(self):
- tutils.raises(language.ParseException, parse_request, 'GET')
-
- def test_simple(self):
- r = parse_request('GET:"/foo"')
- assert r.method.string() == "GET"
- assert r.path.string() == "/foo"
- r = parse_request('GET:/foo')
- assert r.path.string() == "/foo"
- r = parse_request('GET:@1k')
- assert len(r.path.string()) == 1024
-
- def test_multiple(self):
- r = language.parse_requests("GET:/ PUT:/")
- assert r[0].method.string() == "GET"
- assert r[1].method.string() == "PUT"
- assert len(r) == 2
-
- l = """
- GET
- "/foo"
- ir,@1
-
- PUT
-
- "/foo
-
-
-
- bar"
-
- ir,@1
- """
- r = language.parse_requests(l)
- assert len(r) == 2
- assert r[0].method.string() == "GET"
- assert r[1].method.string() == "PUT"
-
- l = """
- get:"http://localhost:9999/p/200":ir,@1
- get:"http://localhost:9999/p/200":ir,@2
- """
- r = language.parse_requests(l)
- assert len(r) == 2
- assert r[0].method.string() == "GET"
- assert r[1].method.string() == "GET"
-
- def test_pathodspec(self):
- l = "get:/p:s'200'"
- r = language.parse_requests(l)
- assert len(r) == 1
- assert len(r[0].tokens) == 3
- assert isinstance(r[0].tokens[2], base.PathodSpec)
- assert r[0].values({})
-
- def test_render(self):
- s = cStringIO.StringIO()
- r = parse_request("GET:'/foo'")
- assert language.serve(
- r,
- s,
- language.Settings(request_host = "foo.com")
- )
-
- def test_multiline(self):
- l = """
- GET
- "/foo"
- ir,@1
- """
- r = parse_request(l)
- assert r.method.string() == "GET"
- assert r.path.string() == "/foo"
- assert r.actions
-
- l = """
- GET
-
- "/foo
-
-
-
- bar"
-
- ir,@1
- """
- r = parse_request(l)
- assert r.method.string() == "GET"
- assert r.path.string().endswith("bar")
- assert r.actions
-
- def test_spec(self):
- def rt(s):
- s = parse_request(s).spec()
- assert parse_request(s).spec() == s
- rt("get:/foo")
- rt("get:/foo:da")
-
- def test_freeze(self):
- r = parse_request("GET:/:b@100").freeze(language.Settings())
- assert len(r.spec()) > 100
-
- def test_path_generator(self):
- r = parse_request("GET:@100").freeze(language.Settings())
- assert len(r.spec()) > 100
-
- def test_websocket(self):
- r = parse_request('ws:/path/')
- res = r.resolve(language.Settings())
- assert res.method.string().lower() == "get"
- assert res.tok(base.Path).value.val == "/path/"
- assert res.tok(base.Method).value.val.lower() == "get"
- assert http.get_header("Upgrade", res.headers).value.val == "websocket"
-
- r = parse_request('ws:put:/path/')
- res = r.resolve(language.Settings())
- assert r.method.string().lower() == "put"
- assert res.tok(base.Path).value.val == "/path/"
- assert res.tok(base.Method).value.val.lower() == "put"
- assert http.get_header("Upgrade", res.headers).value.val == "websocket"
-
-
class TestWebsocketFrame:
def test_spec(self):
e = websockets.WebsocketFrame.expr()
@@ -619,7 +486,6 @@ class TestWebsocketFrame:
class TestWriteValues:
-
def test_send_chunk(self):
v = "foobarfoobar"
for bs in range(1, len(v) + 2):
@@ -700,122 +566,3 @@ class TestWriteValues:
r = language.parse_response("400:ia,'xx'")
language.serve(r, s, {})
assert s.getvalue().endswith('xx')
-
-
-class TestResponse:
- def dummy_response(self):
- return language.parse_response("400'msg'")
-
- def test_response(self):
- r = language.parse_response("400:m'msg'")
- assert r.code.string() == "400"
- assert r.reason.string() == "msg"
-
- r = language.parse_response("400:m'msg':b@100b")
- assert r.reason.string() == "msg"
- assert r.body.values({})
- assert str(r)
-
- r = language.parse_response("200")
- assert r.code.string() == "200"
- assert not r.reason
- assert "OK" in [i[:] for i in r.preamble({})]
-
- def test_render(self):
- s = cStringIO.StringIO()
- r = language.parse_response("400:m'msg'")
- assert language.serve(r, s, {})
-
- r = language.parse_response("400:p0,100:dr")
- assert "p0" in r.spec()
- s = r.preview_safe()
- assert "p0" not in s.spec()
-
- def test_raw(self):
- s = cStringIO.StringIO()
- r = language.parse_response("400:b'foo'")
- language.serve(r, s, {})
- v = s.getvalue()
- assert "Content-Length" in v
-
- s = cStringIO.StringIO()
- r = language.parse_response("400:b'foo':r")
- language.serve(r, s, {})
- v = s.getvalue()
- assert "Content-Length" not in v
-
- def test_length(self):
- def testlen(x):
- s = cStringIO.StringIO()
- language.serve(x, s, language.Settings())
- assert x.length(language.Settings()) == len(s.getvalue())
- testlen(language.parse_response("400:m'msg':r"))
- testlen(language.parse_response("400:m'msg':h'foo'='bar':r"))
- testlen(language.parse_response("400:m'msg':h'foo'='bar':b@100b:r"))
-
- def test_maximum_length(self):
- def testlen(x):
- s = cStringIO.StringIO()
- m = x.maximum_length({})
- language.serve(x, s, {})
- assert m >= len(s.getvalue())
-
- r = language.parse_response("400:m'msg':b@100:d0")
- testlen(r)
-
- r = language.parse_response("400:m'msg':b@100:d0:i0,'foo'")
- testlen(r)
-
- r = language.parse_response("400:m'msg':b@100:d0:i0,'foo'")
- testlen(r)
-
- def test_parse_err(self):
- tutils.raises(
- language.ParseException, language.parse_response, "400:msg,b:"
- )
- try:
- language.parse_response("400'msg':b:")
- except language.ParseException, v:
- assert v.marked()
- assert str(v)
-
- def test_nonascii(self):
- tutils.raises("ascii", language.parse_response, "foo:b\xf0")
-
- def test_parse_header(self):
- r = language.parse_response('400:h"foo"="bar"')
- assert http.get_header("foo", r.headers)
-
- def test_parse_pause_before(self):
- r = language.parse_response("400:p0,10")
- assert r.actions[0].spec() == "p0,10"
-
- def test_parse_pause_after(self):
- r = language.parse_response("400:pa,10")
- assert r.actions[0].spec() == "pa,10"
-
- def test_parse_pause_random(self):
- r = language.parse_response("400:pr,10")
- assert r.actions[0].spec() == "pr,10"
-
- def test_parse_stress(self):
- # While larger values are known to work on linux, len() technically
- # returns an int and a python 2.7 int on windows has 32bit precision.
- # Therefore, we should keep the body length < 2147483647 bytes in our
- # tests.
- r = language.parse_response("400:b@1g")
- assert r.length({})
-
- def test_spec(self):
- def rt(s):
- s = language.parse_response(s).spec()
- assert language.parse_response(s).spec() == s
- rt("400:b@100g")
- rt("400")
- rt("400:da")
-
- def test_websockets(self):
- r = language.parse_response("ws")
- tutils.raises("no websocket key", r.resolve, language.Settings())
- res = r.resolve(language.Settings(websocket_key="foo"))
- assert res.code.string() == "101"
diff --git a/test/test_language_http.py b/test/test_language_http.py
new file mode 100644
index 00000000..50473339
--- /dev/null
+++ b/test/test_language_http.py
@@ -0,0 +1,260 @@
+import cStringIO
+
+from libpathod import language
+from libpathod.language import http, base
+import tutils
+
+
+def parse_request(s):
+ return language.parse_requests(s)[0]
+
+
+def test_make_error_response():
+ d = cStringIO.StringIO()
+ s = http.make_error_response("foo")
+ language.serve(s, d, {})
+
+
+class TestRequest:
+ def test_nonascii(self):
+ tutils.raises("ascii", parse_request, "get:\xf0")
+
+ def test_err(self):
+ tutils.raises(language.ParseException, parse_request, 'GET')
+
+ def test_simple(self):
+ r = parse_request('GET:"/foo"')
+ assert r.method.string() == "GET"
+ assert r.path.string() == "/foo"
+ r = parse_request('GET:/foo')
+ assert r.path.string() == "/foo"
+ r = parse_request('GET:@1k')
+ assert len(r.path.string()) == 1024
+
+ def test_multiple(self):
+ r = language.parse_requests("GET:/ PUT:/")
+ assert r[0].method.string() == "GET"
+ assert r[1].method.string() == "PUT"
+ assert len(r) == 2
+
+ l = """
+ GET
+ "/foo"
+ ir,@1
+
+ PUT
+
+ "/foo
+
+
+
+ bar"
+
+ ir,@1
+ """
+ r = language.parse_requests(l)
+ assert len(r) == 2
+ assert r[0].method.string() == "GET"
+ assert r[1].method.string() == "PUT"
+
+ l = """
+ get:"http://localhost:9999/p/200":ir,@1
+ get:"http://localhost:9999/p/200":ir,@2
+ """
+ r = language.parse_requests(l)
+ assert len(r) == 2
+ assert r[0].method.string() == "GET"
+ assert r[1].method.string() == "GET"
+
+ def test_pathodspec(self):
+ l = "get:/p:s'200'"
+ r = language.parse_requests(l)
+ assert len(r) == 1
+ assert len(r[0].tokens) == 3
+ assert isinstance(r[0].tokens[2], base.PathodSpec)
+ assert r[0].values({})
+
+ def test_render(self):
+ s = cStringIO.StringIO()
+ r = parse_request("GET:'/foo'")
+ assert language.serve(
+ r,
+ s,
+ language.Settings(request_host = "foo.com")
+ )
+
+ def test_multiline(self):
+ l = """
+ GET
+ "/foo"
+ ir,@1
+ """
+ r = parse_request(l)
+ assert r.method.string() == "GET"
+ assert r.path.string() == "/foo"
+ assert r.actions
+
+ l = """
+ GET
+
+ "/foo
+
+
+
+ bar"
+
+ ir,@1
+ """
+ r = parse_request(l)
+ assert r.method.string() == "GET"
+ assert r.path.string().endswith("bar")
+ assert r.actions
+
+ def test_spec(self):
+ def rt(s):
+ s = parse_request(s).spec()
+ assert parse_request(s).spec() == s
+ rt("get:/foo")
+ rt("get:/foo:da")
+
+ def test_freeze(self):
+ r = parse_request("GET:/:b@100").freeze(language.Settings())
+ assert len(r.spec()) > 100
+
+ def test_path_generator(self):
+ r = parse_request("GET:@100").freeze(language.Settings())
+ assert len(r.spec()) > 100
+
+ def test_websocket(self):
+ r = parse_request('ws:/path/')
+ res = r.resolve(language.Settings())
+ assert res.method.string().lower() == "get"
+ assert res.tok(base.Path).value.val == "/path/"
+ assert res.tok(base.Method).value.val.lower() == "get"
+ assert http.get_header("Upgrade", res.headers).value.val == "websocket"
+
+ r = parse_request('ws:put:/path/')
+ res = r.resolve(language.Settings())
+ assert r.method.string().lower() == "put"
+ assert res.tok(base.Path).value.val == "/path/"
+ assert res.tok(base.Method).value.val.lower() == "put"
+ assert http.get_header("Upgrade", res.headers).value.val == "websocket"
+
+
+class TestResponse:
+ def dummy_response(self):
+ return language.parse_response("400'msg'")
+
+ def test_response(self):
+ r = language.parse_response("400:m'msg'")
+ assert r.code.string() == "400"
+ assert r.reason.string() == "msg"
+
+ r = language.parse_response("400:m'msg':b@100b")
+ assert r.reason.string() == "msg"
+ assert r.body.values({})
+ assert str(r)
+
+ r = language.parse_response("200")
+ assert r.code.string() == "200"
+ assert not r.reason
+ assert "OK" in [i[:] for i in r.preamble({})]
+
+ def test_render(self):
+ s = cStringIO.StringIO()
+ r = language.parse_response("400:m'msg'")
+ assert language.serve(r, s, {})
+
+ r = language.parse_response("400:p0,100:dr")
+ assert "p0" in r.spec()
+ s = r.preview_safe()
+ assert "p0" not in s.spec()
+
+ def test_raw(self):
+ s = cStringIO.StringIO()
+ r = language.parse_response("400:b'foo'")
+ language.serve(r, s, {})
+ v = s.getvalue()
+ assert "Content-Length" in v
+
+ s = cStringIO.StringIO()
+ r = language.parse_response("400:b'foo':r")
+ language.serve(r, s, {})
+ v = s.getvalue()
+ assert "Content-Length" not in v
+
+ def test_length(self):
+ def testlen(x):
+ s = cStringIO.StringIO()
+ language.serve(x, s, language.Settings())
+ assert x.length(language.Settings()) == len(s.getvalue())
+ testlen(language.parse_response("400:m'msg':r"))
+ testlen(language.parse_response("400:m'msg':h'foo'='bar':r"))
+ testlen(language.parse_response("400:m'msg':h'foo'='bar':b@100b:r"))
+
+ def test_maximum_length(self):
+ def testlen(x):
+ s = cStringIO.StringIO()
+ m = x.maximum_length({})
+ language.serve(x, s, {})
+ assert m >= len(s.getvalue())
+
+ r = language.parse_response("400:m'msg':b@100:d0")
+ testlen(r)
+
+ r = language.parse_response("400:m'msg':b@100:d0:i0,'foo'")
+ testlen(r)
+
+ r = language.parse_response("400:m'msg':b@100:d0:i0,'foo'")
+ testlen(r)
+
+ def test_parse_err(self):
+ tutils.raises(
+ language.ParseException, language.parse_response, "400:msg,b:"
+ )
+ try:
+ language.parse_response("400'msg':b:")
+ except language.ParseException, v:
+ assert v.marked()
+ assert str(v)
+
+ def test_nonascii(self):
+ tutils.raises("ascii", language.parse_response, "foo:b\xf0")
+
+ def test_parse_header(self):
+ r = language.parse_response('400:h"foo"="bar"')
+ assert http.get_header("foo", r.headers)
+
+ def test_parse_pause_before(self):
+ r = language.parse_response("400:p0,10")
+ assert r.actions[0].spec() == "p0,10"
+
+ def test_parse_pause_after(self):
+ r = language.parse_response("400:pa,10")
+ assert r.actions[0].spec() == "pa,10"
+
+ def test_parse_pause_random(self):
+ r = language.parse_response("400:pr,10")
+ assert r.actions[0].spec() == "pr,10"
+
+ def test_parse_stress(self):
+ # While larger values are known to work on linux, len() technically
+ # returns an int and a python 2.7 int on windows has 32bit precision.
+ # Therefore, we should keep the body length < 2147483647 bytes in our
+ # tests.
+ r = language.parse_response("400:b@1g")
+ assert r.length({})
+
+ def test_spec(self):
+ def rt(s):
+ s = language.parse_response(s).spec()
+ assert language.parse_response(s).spec() == s
+ rt("400:b@100g")
+ rt("400")
+ rt("400:da")
+
+ def test_websockets(self):
+ r = language.parse_response("ws")
+ tutils.raises("no websocket key", r.resolve, language.Settings())
+ res = r.resolve(language.Settings(websocket_key="foo"))
+ assert res.code.string() == "101"