from six.moves import cStringIO as StringIO import pytest from pathod import pathod, version from netlib import tcp from netlib.exceptions import HttpException, TlsException import tutils class TestPathod(object): def test_logging(self): s = 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 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): # 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. tutils.raises(Exception, self.pathoc, ["get:/:p1,1"]) 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( 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(tutils.DaemonTests): ssl = True ssloptions = dict( certs=[("*", tutils.test_data.path("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(tutils.DaemonTests): ssl = True ssloptions = dict( cn="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(tutils.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(tutils.DaemonTests): hexdump = True def test_hexdump(self): assert self.get(r"200:b'\xf0'") class TestNocraft(tutils.DaemonTests): nocraft = True def test_nocraft(self): r = self.get(r"200:b'\xf0'") assert r.status_code == 800 assert "Crafting disabled" in r.content class CommonTests(tutils.DaemonTests): def test_binarydata(self): assert self.get(r"200:b'\xf0'") assert self.d.last_log() # FIXME: Other binary data elements @pytest.mark.skip(reason="race condition") 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_info(self): assert tuple(self.d.info()["version"]) == version.IVERSION @pytest.mark.skip(reason="race condition") def test_logs(self): assert self.d.clear_log() assert not self.d.last_log() assert self.get("202:da") assert len(self.d.log()) == 1 assert self.d.clear_log() assert len(self.d.log()) == 0 def test_disconnect(self): rsp = self.get("202:b@100k:d200") assert len(rsp.content) < 200 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(tutils.DaemonTests): ssl = True noweb = True noapi = True nohang = True if tcp.HAS_ALPN: def test_http2(self): r, _ = self.pathoc(["GET:/"], ssl=True, use_http2=True) assert r[0].status_code == 800