import io import pytest from pathod import pathod from mitmproxy.net import tcp from mitmproxy import exceptions from mitmproxy.test import tutils from . import tservers class TestPathod: def test_logging(self): s = io.StringIO() p = pathod.Pathod(("127.0.0.1", 0), logfp=s) assert len(p.get_log()) == 0 id = p.add_log(dict(s="foo")) assert p.log_by_id(id) assert len(p.get_log()) == 1 p.clear_log() assert len(p.get_log()) == 0 for _ in range(p.LOGBUF + 1): p.add_log(dict(s="foo")) assert len(p.get_log()) <= p.LOGBUF class TestTimeout(tservers.DaemonTests): timeout = 0.01 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 # different exceptions to be raised here. with pytest.raises(Exception): self.pathoc(["get:/:p1,1"]) assert self.d.last_log()["type"] == "timeout" class TestNotAfterConnect(tservers.DaemonTests): ssl = False ssloptions = dict( not_after_connect=True ) def test_connect(self): r, _ = self.pathoc( [r"get:'http://foo.com/p/202':da"], connect_to=("localhost", self.d.port) ) assert r[0].status_code == 202 class TestCustomCert(tservers.DaemonTests): ssl = True ssloptions = dict( certs=[("*", tutils.test_data.path("pathod/data/testkey.pem"))], ) def test_connect(self): r, _ = self.pathoc([r"get:/p/202"]) r = r[0] assert r.status_code == 202 assert r.sslinfo assert "test.com" in str(r.sslinfo.certchain[0].get_subject()) class TestSSLCN(tservers.DaemonTests): ssl = True ssloptions = dict( cn=b"foo.com" ) def test_connect(self): r, _ = self.pathoc([r"get:/p/202"]) r = r[0] assert r.status_code == 202 assert r.sslinfo assert r.sslinfo.certchain[0].get_subject().CN == "foo.com" class TestNohang(tservers.DaemonTests): nohang = True def test_nohang(self): r = self.get("200:p0,0") assert r.status_code == 800 l = self.d.last_log() assert "Pauses have been disabled" in l["response"]["msg"] class TestHexdump(tservers.DaemonTests): hexdump = True def test_hexdump(self): assert self.get(r"200:b'\xf0'") class TestNocraft(tservers.DaemonTests): nocraft = True def test_nocraft(self): r = self.get(r"200:b'\xf0'") assert r.status_code == 800 assert b"Crafting disabled" in r.content class CommonTests(tservers.DaemonTests): def test_binarydata(self): assert self.get(r"200:b'\xf0'") assert self.d.last_log() # FIXME: Other binary data elements def test_sizelimit(self): r = self.get("200:b@1g") assert r.status_code == 800 l = self.d.last_log() assert "too large" in l["response"]["msg"] def test_preline(self): r, _ = self.pathoc([r"get:'/p/200':i0,'\r\n'"]) assert r[0].status_code == 200 def test_logs(self): self.d.clear_log() assert self.get("202:da") assert self.d.expect_log(1) self.d.clear_log() assert len(self.d.log()) == 0 def test_disconnect(self): with pytest.raises(Exception, match="Unexpected EOF"): self.get("202:b@100k:d200") def test_parserr(self): rsp = self.get("400:msg,b:") assert rsp.status_code == 800 def test_static(self): rsp = self.get("200:b 0 class TestHTTP2(tservers.DaemonTests): ssl = True nohang = True def test_http2(self): r, _ = self.pathoc(["GET:/"], ssl=True, use_http2=True) assert r[0].status_code == 800