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()