aboutsummaryrefslogtreecommitdiffstats
path: root/test/pathod/tutils.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/pathod/tutils.py')
-rw-r--r--test/pathod/tutils.py153
1 files changed, 0 insertions, 153 deletions
diff --git a/test/pathod/tutils.py b/test/pathod/tutils.py
deleted file mode 100644
index ccc3df43..00000000
--- a/test/pathod/tutils.py
+++ /dev/null
@@ -1,153 +0,0 @@
-import tempfile
-import re
-import shutil
-import requests
-import io
-import urllib
-
-from mitmproxy.utils import data
-
-from mitmproxy.net import tcp
-from mitmproxy.test import tutils
-
-from pathod import language
-from pathod import pathoc
-from pathod import pathod
-from pathod import test
-
-
-def treader(bytes):
- """
- Construct a tcp.Read object from bytes.
- """
- fp = io.BytesIO(bytes)
- return tcp.Reader(fp)
-
-
-class DaemonTests:
- nohang = False
- ssl = False
- timeout = None
- hexdump = False
- ssloptions = None
- nocraft = False
- explain = True
-
- @classmethod
- def setup_class(cls):
- opts = cls.ssloptions or {}
- cls.confdir = tempfile.mkdtemp()
- opts["confdir"] = cls.confdir
- so = pathod.SSLOptions(**opts)
- cls.d = test.Daemon(
- staticdir=test_data.path("data"),
- anchors=[
- (re.compile("/anchor/.*"), "202:da")
- ],
- ssl=cls.ssl,
- ssloptions=so,
- sizelimit=1 * 1024 * 1024,
- nohang=cls.nohang,
- timeout=cls.timeout,
- hexdump=cls.hexdump,
- nocraft=cls.nocraft,
- logreq=True,
- logresp=True,
- explain=cls.explain
- )
-
- @classmethod
- def teardown_class(cls):
- cls.d.shutdown()
- shutil.rmtree(cls.confdir)
-
- def teardown(self):
- self.d.wait_for_silence()
- self.d.clear_log()
-
- def _getpath(self, path, params=None):
- scheme = "https" if self.ssl else "http"
- resp = requests.get(
- "%s://localhost:%s/%s" % (
- scheme,
- self.d.port,
- path
- ),
- verify=False,
- params=params
- )
- return resp
-
- def getpath(self, path, params=None):
- logfp = io.StringIO()
- c = pathoc.Pathoc(
- ("localhost", self.d.port),
- ssl=self.ssl,
- fp=logfp,
- )
- with c.connect():
- if params:
- path = path + "?" + urllib.parse.urlencode(params)
- resp = c.request("get:%s" % path)
- return resp
-
- def get(self, spec):
- logfp = io.StringIO()
- c = pathoc.Pathoc(
- ("localhost", self.d.port),
- ssl=self.ssl,
- fp=logfp,
- )
- with c.connect():
- resp = c.request(
- "get:/p/%s" % urllib.parse.quote(spec)
- )
- return resp
-
- def pathoc(
- self,
- specs,
- timeout=None,
- connect_to=None,
- ssl=None,
- ws_read_limit=None,
- use_http2=False,
- ):
- """
- Returns a (messages, text log) tuple.
- """
- if ssl is None:
- ssl = self.ssl
- logfp = io.StringIO()
- c = pathoc.Pathoc(
- ("localhost", self.d.port),
- ssl=ssl,
- ws_read_limit=ws_read_limit,
- timeout=timeout,
- fp=logfp,
- use_http2=use_http2,
- )
- with c.connect(connect_to):
- ret = []
- for i in specs:
- resp = c.request(i)
- if resp:
- ret.append(resp)
- for frm in c.wait():
- ret.append(frm)
- c.stop()
- return ret, logfp.getvalue()
-
-
-tmpdir = tutils.tmpdir
-
-raises = tutils.raises
-
-test_data = data.Data(__name__)
-
-
-def render(r, settings=language.Settings()):
- r = r.resolve(settings)
- s = io.BytesIO()
- assert language.serve(r, s, settings)
- return s.getvalue()