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
127
128
|
import tempfile
import os
import re
import shutil
import cStringIO
from contextlib import contextmanager
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 = cStringIO.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 = cStringIO.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 = utils.Data(__name__)
def render(r, settings=language.Settings()):
r = r.resolve(settings)
s = cStringIO.StringIO()
assert language.serve(r, s, settings)
return s.getvalue()
|
