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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
import tempfile
import re
import shutil
from six.moves import cStringIO as StringIO
import netlib
from pathod import utils, test, pathoc, pathod, language
from netlib import tcp
import requests
def treader(bytes):
"""
Construct a tcp.Read object from bytes.
"""
fp = StringIO(bytes)
return tcp.Reader(fp)
class DaemonTests(object):
noweb = False
noapi = False
nohang = False
ssl = False
timeout = None
hexdump = False
ssloptions = None
nocraft = False
@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,
noweb=cls.noweb,
noapi=cls.noapi,
nohang=cls.nohang,
timeout=cls.timeout,
hexdump=cls.hexdump,
nocraft=cls.nocraft,
logreq=True,
logresp=True,
explain=True
)
@classmethod
def teardown_class(cls):
cls.d.shutdown()
shutil.rmtree(cls.confdir)
def teardown(self):
if not (self.noweb or self.noapi):
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 get(self, spec):
resp = requests.get(self.d.p(spec), verify=False)
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 = StringIO()
c = pathoc.Pathoc(
("localhost", self.d.port),
ssl=ssl,
ws_read_limit=ws_read_limit,
timeout=timeout,
fp=logfp,
use_http2=use_http2,
)
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 = netlib.tutils.tmpdir
raises = netlib.tutils.raises
test_data = netlib.utils.Data(__name__)
def render(r, settings=language.Settings()):
r = r.resolve(settings)
s = StringIO()
assert language.serve(r, s, settings)
return s.getvalue()
|