diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/netlib/test_multidict.py | 4 | ||||
| -rw-r--r-- | test/pathod/test_app.py | 85 | ||||
| -rw-r--r-- | test/pathod/test_language_actions.py | 20 | ||||
| -rw-r--r-- | test/pathod/test_language_base.py | 8 | ||||
| -rw-r--r-- | test/pathod/test_language_writer.py | 62 | ||||
| -rw-r--r-- | test/pathod/test_pathoc.py | 128 | ||||
| -rw-r--r-- | test/pathod/test_pathod.py | 22 | ||||
| -rw-r--r-- | test/pathod/tutils.py | 14 |
8 files changed, 99 insertions, 244 deletions
diff --git a/test/netlib/test_multidict.py b/test/netlib/test_multidict.py index 7319f1c5..a35d5cc5 100644 --- a/test/netlib/test_multidict.py +++ b/test/netlib/test_multidict.py @@ -120,7 +120,7 @@ class TestMultiDict(object): ("a", "b"), ("x", "x"), ("c", "d"), - ("X", "x"), + ("X", "X"), ("e", "f"), )) md.set_all("x", ["1", "2", "3"]) @@ -128,7 +128,7 @@ class TestMultiDict(object): ("a", "b"), ("x", "1"), ("c", "d"), - ("x", "2"), + ("X", "2"), ("e", "f"), ("x", "3"), ) diff --git a/test/pathod/test_app.py b/test/pathod/test_app.py deleted file mode 100644 index fbaa773c..00000000 --- a/test/pathod/test_app.py +++ /dev/null @@ -1,85 +0,0 @@ -import tutils - - -class TestApp(tutils.DaemonTests): - SSL = False - - def test_index(self): - r = self.getpath("/") - assert r.status_code == 200 - assert r.content - - def test_about(self): - r = self.getpath("/about") - assert r.status_code == 200 - - def test_download(self): - r = self.getpath("/download") - assert r.status_code == 200 - - def test_docs(self): - assert self.getpath("/docs/pathod").status_code == 200 - assert self.getpath("/docs/pathoc").status_code == 200 - assert self.getpath("/docs/language").status_code == 200 - assert self.getpath("/docs/pathod").status_code == 200 - assert self.getpath("/docs/test").status_code == 200 - - def test_log(self): - assert self.getpath("/log").status_code == 200 - assert self.get("200:da").status_code == 200 - id = self.d.expect_log(1)[0]["id"] - assert self.getpath("/log").status_code == 200 - assert self.getpath("/log/%s" % id).status_code == 200 - assert self.getpath("/log/9999999").status_code == 404 - - def test_log_binary(self): - assert self.get("200:h@10b=@10b:da") - - def test_response_preview(self): - r = self.getpath("/response_preview", params=dict(spec="200")) - assert r.status_code == 200 - assert 'Response' in r.content - - r = self.getpath("/response_preview", params=dict(spec="foo")) - assert r.status_code == 200 - assert 'Error' in r.content - - r = self.getpath("/response_preview", params=dict(spec="200:b@100m")) - assert r.status_code == 200 - assert "too large" in r.content - - r = self.getpath("/response_preview", params=dict(spec="200:b@5k")) - assert r.status_code == 200 - assert 'Response' in r.content - - r = self.getpath( - "/response_preview", - params=dict( - spec="200:b<nonexistent")) - assert r.status_code == 200 - assert 'File access denied' in r.content - - r = self.getpath("/response_preview", params=dict(spec="200:b<file")) - assert r.status_code == 200 - assert 'testfile' in r.content - - def test_request_preview(self): - r = self.getpath("/request_preview", params=dict(spec="get:/")) - assert r.status_code == 200 - assert 'Request' in r.content - - r = self.getpath("/request_preview", params=dict(spec="foo")) - assert r.status_code == 200 - assert 'Error' in r.content - - r = self.getpath("/request_preview", params=dict(spec="get:/:b@100m")) - assert r.status_code == 200 - assert "too large" in r.content - - r = self.getpath("/request_preview", params=dict(spec="get:/:b@5k")) - assert r.status_code == 200 - assert 'Request' in r.content - - r = self.getpath("/request_preview", params=dict(spec="")) - assert r.status_code == 200 - assert 'empty spec' in r.content diff --git a/test/pathod/test_language_actions.py b/test/pathod/test_language_actions.py index 81d2155d..f12d8105 100644 --- a/test/pathod/test_language_actions.py +++ b/test/pathod/test_language_actions.py @@ -1,11 +1,11 @@ -from six.moves import cStringIO as StringIO +from six import BytesIO from pathod.language import actions from pathod import language def parse_request(s): - return language.parse_pathoc(s).next() + return next(language.parse_pathoc(s)) def test_unique_name(): @@ -16,9 +16,9 @@ def test_unique_name(): class TestDisconnects: def test_parse_pathod(self): - a = language.parse_pathod("400:d0").next().actions[0] + a = next(language.parse_pathod("400:d0")).actions[0] assert a.spec() == "d0" - a = language.parse_pathod("400:dr").next().actions[0] + a = next(language.parse_pathod("400:dr")).actions[0] assert a.spec() == "dr" def test_at(self): @@ -42,18 +42,18 @@ class TestDisconnects: class TestInject: def test_parse_pathod(self): - a = language.parse_pathod("400:ir,@100").next().actions[0] + a = next(language.parse_pathod("400:ir,@100")).actions[0] assert a.offset == "r" assert a.value.datatype == "bytes" assert a.value.usize == 100 - a = language.parse_pathod("400:ia,@100").next().actions[0] + a = next(language.parse_pathod("400:ia,@100")).actions[0] assert a.offset == "a" def test_at(self): e = actions.InjectAt.expr() v = e.parseString("i0,'foo'")[0] - assert v.value.val == "foo" + assert v.value.val == b"foo" assert v.offset == 0 assert isinstance(v, actions.InjectAt) @@ -61,8 +61,8 @@ class TestInject: assert v.offset == "r" def test_serve(self): - s = StringIO() - r = language.parse_pathod("400:i0,'foo'").next() + s = BytesIO() + r = next(language.parse_pathod("400:i0,'foo'")) assert language.serve(r, s, {}) def test_spec(self): @@ -96,7 +96,7 @@ class TestPauses: assert v.offset == "a" def test_request(self): - r = language.parse_pathod('400:p10,10').next() + r = next(language.parse_pathod('400:p10,10')) assert r.actions[0].spec() == "p10,10" def test_spec(self): diff --git a/test/pathod/test_language_base.py b/test/pathod/test_language_base.py index 47e51bb0..075dc2b8 100644 --- a/test/pathod/test_language_base.py +++ b/test/pathod/test_language_base.py @@ -38,7 +38,7 @@ class TestTokValueNakedLiteral: class TestTokValueLiteral: - def test_espr(self): + def test_expr(self): v = base.TokValueLiteral("foo") assert v.expr() assert v.val == b"foo" @@ -132,7 +132,7 @@ class TestTokValueFile: with tutils.tmpdir() as t: p = os.path.join(t, "path") with open(p, "wb") as f: - f.write("x" * 10000) + f.write(b"x" * 10000) assert v.get_generator(language.Settings(staticdir=t)) @@ -207,13 +207,13 @@ class TestMisc: p = os.path.join(t, "path") s = base.Settings(staticdir=t) with open(p, "wb") as f: - f.write("a" * 20) + f.write(b"a" * 20) v = e.parseString("m<path")[0] tutils.raises("invalid value length", v.values, s) p = os.path.join(t, "path") with open(p, "wb") as f: - f.write("a" * 4) + f.write(b"a" * 4) v = e.parseString("m<path")[0] assert v.values(s) diff --git a/test/pathod/test_language_writer.py b/test/pathod/test_language_writer.py index 0a85524f..c02f66f3 100644 --- a/test/pathod/test_language_writer.py +++ b/test/pathod/test_language_writer.py @@ -1,53 +1,53 @@ -from six.moves import cStringIO as StringIO +from six import BytesIO from pathod import language from pathod.language import writer def test_send_chunk(): - v = "foobarfoobar" + v = b"foobarfoobar" for bs in range(1, len(v) + 2): - s = StringIO() + s = BytesIO() writer.send_chunk(s, v, bs, 0, len(v)) assert s.getvalue() == v for start in range(len(v)): for end in range(len(v)): - s = StringIO() + s = BytesIO() writer.send_chunk(s, v, bs, start, end) assert s.getvalue() == v[start:end] def test_write_values_inject(): - tst = "foo" + tst = b"foo" - s = StringIO() - writer.write_values(s, [tst], [(0, "inject", "aaa")], blocksize=5) - assert s.getvalue() == "aaafoo" + s = BytesIO() + writer.write_values(s, [tst], [(0, "inject", b"aaa")], blocksize=5) + assert s.getvalue() == b"aaafoo" - s = StringIO() - writer.write_values(s, [tst], [(1, "inject", "aaa")], blocksize=5) - assert s.getvalue() == "faaaoo" + s = BytesIO() + writer.write_values(s, [tst], [(1, "inject", b"aaa")], blocksize=5) + assert s.getvalue() == b"faaaoo" - s = StringIO() - writer.write_values(s, [tst], [(1, "inject", "aaa")], blocksize=5) - assert s.getvalue() == "faaaoo" + s = BytesIO() + writer.write_values(s, [tst], [(1, "inject", b"aaa")], blocksize=5) + assert s.getvalue() == b"faaaoo" def test_write_values_disconnects(): - s = StringIO() - tst = "foo" * 100 + s = BytesIO() + tst = b"foo" * 100 writer.write_values(s, [tst], [(0, "disconnect")], blocksize=5) assert not s.getvalue() def test_write_values(): - tst = "foobarvoing" - s = StringIO() + tst = b"foobarvoing" + s = BytesIO() writer.write_values(s, [tst], []) assert s.getvalue() == tst for bs in range(1, len(tst) + 2): for off in range(len(tst)): - s = StringIO() + s = BytesIO() writer.write_values( s, [tst], [(off, "disconnect")], blocksize=bs ) @@ -55,36 +55,36 @@ def test_write_values(): def test_write_values_pauses(): - tst = "".join(str(i) for i in range(10)) + tst = "".join(str(i) for i in range(10)).encode() for i in range(2, 10): - s = StringIO() + s = BytesIO() writer.write_values( s, [tst], [(2, "pause", 0), (1, "pause", 0)], blocksize=i ) assert s.getvalue() == tst for i in range(2, 10): - s = StringIO() + s = BytesIO() writer.write_values(s, [tst], [(1, "pause", 0)], blocksize=i) assert s.getvalue() == tst - tst = ["".join(str(i) for i in range(10))] * 5 + tst = [tst] * 5 for i in range(2, 10): - s = StringIO() + s = BytesIO() writer.write_values(s, tst[:], [(1, "pause", 0)], blocksize=i) - assert s.getvalue() == "".join(tst) + assert s.getvalue() == b"".join(tst) def test_write_values_after(): - s = StringIO() - r = language.parse_pathod("400:da").next() + s = BytesIO() + r = next(language.parse_pathod("400:da")) language.serve(r, s, {}) - s = StringIO() - r = language.parse_pathod("400:pa,0").next() + s = BytesIO() + r = next(language.parse_pathod("400:pa,0")) language.serve(r, s, {}) - s = StringIO() - r = language.parse_pathod("400:ia,'xx'").next() + s = BytesIO() + r = next(language.parse_pathod("400:ia,'xx'")) language.serve(r, s, {}) assert s.getvalue().endswith('xx') diff --git a/test/pathod/test_pathoc.py b/test/pathod/test_pathoc.py index 6e36c4bf..8b69a2a6 100644 --- a/test/pathod/test_pathoc.py +++ b/test/pathod/test_pathoc.py @@ -1,7 +1,4 @@ -import json from six.moves import cStringIO as StringIO -import re -import pytest from mock import Mock from netlib import http @@ -9,10 +6,9 @@ from netlib import tcp from netlib.exceptions import NetlibException from netlib.http import http1, http2 -from pathod import pathoc, test, version, pathod, language +from pathod import pathoc, language from netlib.tutils import raises import tutils -from test.mitmproxy.tutils import skip_windows def test_response(): @@ -20,37 +16,7 @@ def test_response(): assert repr(r) -class _TestDaemon: - ssloptions = pathod.SSLOptions() - - @classmethod - def setup_class(cls): - cls.d = test.Daemon( - ssl=cls.ssl, - ssloptions=cls.ssloptions, - staticdir=tutils.test_data.path("data"), - anchors=[ - (re.compile("/anchor/.*"), "202") - ] - ) - - @classmethod - def teardown_class(cls): - cls.d.shutdown() - - def setUp(self): - self.d.clear_log() - - def test_info(self): - c = pathoc.Pathoc( - ("127.0.0.1", self.d.port), - ssl=self.ssl, - fp=None - ) - c.connect() - resp = c.request("get:/api/info") - assert tuple(json.loads(resp.content)["version"]) == version.IVERSION - +class PathocTestDaemon(tutils.DaemonTests): def tval( self, requests, @@ -77,23 +43,23 @@ class _TestDaemon: showsummary=showsummary, fp=s ) - c.connect(showssl=showssl, fp=s) - if timeout: - c.settimeout(timeout) - for i in requests: - r = language.parse_pathoc(i).next() - if explain: - r = r.freeze(language.Settings()) - try: - c.request(r) - except NetlibException: - pass + with c.connect(showssl=showssl, fp=s): + if timeout: + c.settimeout(timeout) + for i in requests: + r = language.parse_pathoc(i).next() + if explain: + r = r.freeze(language.Settings()) + try: + c.request(r) + except NetlibException: + pass return s.getvalue() -class TestDaemonSSL(_TestDaemon): +class TestDaemonSSL(PathocTestDaemon): ssl = True - ssloptions = pathod.SSLOptions( + ssloptions = dict( request_client_cert=True, sans=["test1.com", "test2.com"], alpn_select=b'h2', @@ -106,11 +72,10 @@ class TestDaemonSSL(_TestDaemon): sni="foobar.com", fp=None ) - c.connect() - c.request("get:/p/200") - r = c.request("get:/api/log") - d = json.loads(r.content) - assert d["log"][0]["request"]["sni"] == "foobar.com" + with c.connect(): + c.request("get:/p/200") + log = self.d.log() + assert log[0]["request"]["sni"] == "foobar.com" def test_showssl(self): assert "certificate chain" in self.tval(["get:/p/200"], showssl=True) @@ -122,11 +87,11 @@ class TestDaemonSSL(_TestDaemon): clientcert=tutils.test_data.path("data/clientcert/client.pem"), fp=None ) - c.connect() - c.request("get:/p/200") - r = c.request("get:/api/log") - d = json.loads(r.content) - assert d["log"][0]["request"]["clientcert"]["keyinfo"] + with c.connect(): + c.request("get:/p/200") + + log = self.d.log() + assert log[0]["request"]["clientcert"]["keyinfo"] def test_http2_without_ssl(self): fp = StringIO() @@ -139,7 +104,7 @@ class TestDaemonSSL(_TestDaemon): tutils.raises(NotImplementedError, c.connect) -class TestDaemon(_TestDaemon): +class TestDaemon(PathocTestDaemon): ssl = False def test_ssl_error(self): @@ -163,7 +128,7 @@ class TestDaemon(_TestDaemon): 201]) assert "202" in self.tval(["get:'/p/202:b@1'"], ignorecodes=[200, 201]) - def test_timeout(self): + def _test_timeout(self): assert "Timeout" in self.tval(["get:'/p/200:p0,100'"], timeout=0.01) assert "HTTP" in self.tval( ["get:'/p/200:p5,100'"], @@ -178,8 +143,8 @@ class TestDaemon(_TestDaemon): ) def test_showresp(self): - reqs = ["get:/api/info:p0,0", "get:/api/info:p0,0"] - assert self.tval(reqs).count("200") == 2 + reqs = ["get:/p/200:da", "get:/p/200:da"] + assert self.tval(reqs).count("200 OK") == 2 assert self.tval(reqs, showresp=True).count("HTTP/1.1 200 OK") == 2 assert self.tval( reqs, showresp=True, hexdump=True @@ -187,7 +152,7 @@ class TestDaemon(_TestDaemon): def test_showresp_httperr(self): v = self.tval(["get:'/p/200:d20'"], showresp=True, showsummary=True) - assert "Invalid headers" in v + assert "Invalid header" in v assert "HTTP/" in v def test_explain(self): @@ -195,8 +160,8 @@ class TestDaemon(_TestDaemon): assert "b@100" not in self.tval(reqs, explain=True) def test_showreq(self): - reqs = ["get:/api/info:p0,0", "get:/api/info:p0,0"] - assert self.tval(reqs, showreq=True).count("GET /api") == 2 + reqs = ["get:/p/200:da", "get:/p/200:da"] + assert self.tval(reqs, showreq=True).count("GET /p/200") == 2 assert self.tval( reqs, showreq=True, hexdump=True ).count("0000000000") == 2 @@ -206,23 +171,20 @@ class TestDaemon(_TestDaemon): def test_websocket_shutdown(self): c = pathoc.Pathoc(("127.0.0.1", self.d.port), fp=None) - c.connect() - c.request("ws:/") - c.stop() + with c.connect(): + c.request("ws:/") - @skip_windows - @pytest.mark.skip(reason="race condition") def test_wait_finish(self): c = pathoc.Pathoc( ("127.0.0.1", self.d.port), fp=None, ws_read_limit=1 ) - c.connect() - c.request("ws:/") - c.request("wf:f'wf:x100'") - [i for i in c.wait(timeout=0, finish=False)] - [i for i in c.wait(timeout=0)] + with c.connect(): + c.request("ws:/") + c.request("wf:f'wf:x100'") + [i for i in c.wait(timeout=0, finish=False)] + [i for i in c.wait(timeout=0)] def test_connect_fail(self): to = ("foobar", 80) @@ -264,8 +226,9 @@ class TestDaemon(_TestDaemon): c.socks_connect(("example.com", 0xDEAD)) -class TestDaemonHTTP2(_TestDaemon): +class TestDaemonHTTP2(PathocTestDaemon): ssl = True + explain = False if tcp.HAS_ALPN: @@ -295,10 +258,9 @@ class TestDaemonHTTP2(_TestDaemon): tmp_convert_to_ssl = c.convert_to_ssl c.convert_to_ssl = Mock() c.convert_to_ssl.side_effect = tmp_convert_to_ssl - c.connect() - - _, kwargs = c.convert_to_ssl.call_args - assert set(kwargs['alpn_protos']) == set([b'http/1.1', b'h2']) + with c.connect(): + _, kwargs = c.convert_to_ssl.call_args + assert set(kwargs['alpn_protos']) == set([b'http/1.1', b'h2']) def test_request(self): c = pathoc.Pathoc( @@ -307,6 +269,6 @@ class TestDaemonHTTP2(_TestDaemon): ssl=True, use_http2=True, ) - c.connect() - resp = c.request("get:/p/200") + with c.connect(): + resp = c.request("get:/p/200") assert resp.status_code == 200 diff --git a/test/pathod/test_pathod.py b/test/pathod/test_pathod.py index ec9c169f..0bbad6c2 100644 --- a/test/pathod/test_pathod.py +++ b/test/pathod/test_pathod.py @@ -23,18 +23,10 @@ class TestPathod(object): assert len(p.get_log()) <= p.LOGBUF -class TestNoWeb(tutils.DaemonTests): - noweb = True - - def test_noweb(self): - assert self.get("200:da").status_code == 200 - assert self.getpath("/").status_code == 800 - - class TestTimeout(tutils.DaemonTests): timeout = 0.01 - def test_noweb(self): + def test_timeout(self): # FIXME: Add float values to spec language, reduce test timeout to # increase test performance # This is a bodge - we have some platform difference that causes @@ -43,16 +35,6 @@ class TestTimeout(tutils.DaemonTests): assert self.d.last_log()["type"] == "timeout" -class TestNoApi(tutils.DaemonTests): - noapi = True - - def test_noapi(self): - assert self.getpath("/log").status_code == 404 - r = self.getpath("/") - assert r.status_code == 200 - assert "Log" not in r.content - - class TestNotAfterConnect(tutils.DaemonTests): ssl = False ssloptions = dict( @@ -271,8 +253,6 @@ class TestDaemonSSL(CommonTests): class TestHTTP2(tutils.DaemonTests): ssl = True - noweb = True - noapi = True nohang = True if tcp.HAS_ALPN: diff --git a/test/pathod/tutils.py b/test/pathod/tutils.py index b9f38d86..56cd2002 100644 --- a/test/pathod/tutils.py +++ b/test/pathod/tutils.py @@ -24,14 +24,13 @@ def treader(bytes): class DaemonTests(object): - noweb = False - noapi = False nohang = False ssl = False timeout = None hexdump = False ssloptions = None nocraft = False + explain = True @classmethod def setup_class(cls): @@ -47,15 +46,13 @@ class DaemonTests(object): ssl=cls.ssl, ssloptions=so, sizelimit=1 * 1024 * 1024, - noweb=cls.noweb, - noapi=cls.noapi, nohang=cls.nohang, timeout=cls.timeout, hexdump=cls.hexdump, nocraft=cls.nocraft, logreq=True, logresp=True, - explain=True + explain=cls.explain ) @classmethod @@ -65,8 +62,7 @@ class DaemonTests(object): def teardown(self): self.d.wait_for_silence() - if not (self.noweb or self.noapi): - self.d.clear_log() + self.d.clear_log() def _getpath(self, path, params=None): scheme = "https" if self.ssl else "http" @@ -102,7 +98,9 @@ class DaemonTests(object): fp=logfp, ) with c.connect(): - resp = c.request("get:/p/%s" % urllib.quote(spec).encode("string_escape")) + resp = c.request( + "get:/p/%s" % urllib.quote(spec).encode("string_escape") + ) return resp def pathoc( |