1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
|
from libpathod import pathod, version
from netlib import tcp, http
import tutils
class TestPathod:
def test_instantiation(self):
p = pathod.Pathod(
("127.0.0.1", 0),
anchors = [(".*", "200")]
)
assert p.anchors
tutils.raises("invalid regex", pathod.Pathod, ("127.0.0.1", 0), anchors=[("*", "200")])
tutils.raises("invalid page spec", pathod.Pathod, ("127.0.0.1", 0), anchors=[("foo", "bar")])
def test_logging(self):
p = pathod.Pathod(("127.0.0.1", 0))
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 i in range(p.LOGBUF + 1):
p.add_log(dict(s="foo"))
assert len(p.get_log()) <= p.LOGBUF
class CommonTests(tutils.DaemonTests):
def test_sizelimit(self):
r = self.get("200:b@1g")
assert r.status_code == 800
l = self.d.log()[0]
assert "too large" in l["response"]["error"]
def test_preline(self):
v = self.pathoc(r"get:'/p/200':i0,'\r\n'")
assert v[1] == 200
def test_info(self):
assert tuple(self.d.info()["version"]) == version.IVERSION
def test_logs(self):
l = len(self.d.log())
rsp = self.get("202")
assert len(self.d.log()) == l+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<file")
assert rsp.status_code == 200
assert rsp.content.strip() == "testfile"
def test_anchor(self):
rsp = self.getpath("anchor/foo")
assert rsp.status_code == 202
def test_invalid_first_line(self):
c = tcp.TCPClient("localhost", self.d.port)
c.connect()
if self.SSL:
c.convert_to_ssl()
c.wfile.write("foo\n\n\n")
c.wfile.flush()
l = self.d.log()[0]
assert l["type"] == "error"
assert "foo" in l["msg"]
def test_invalid_body(self):
tutils.raises(http.HttpError, self.pathoc, "get:/:h'content-length'='foo'")
l = self.d.log()[0]
assert l["type"] == "error"
assert "Invalid" in l["msg"]
def test_access_denied(self):
rsp = self.get("=nonexistent")
assert rsp.status_code == 800
class TestDaemon(CommonTests):
SSL = False
class TestDaemonSSL(CommonTests):
SSL = True
def test_ssl_conn_failure(self):
c = tcp.TCPClient("localhost", self.d.port)
c.rbufsize = 0
c.wbufsize = 0
c.connect()
try:
while 1:
c.wfile.write("\r\n\r\n\r\n")
except:
pass
l = self.d.log()[0]
assert l["type"] == "error"
assert "SSL" in l["msg"]
|
