aboutsummaryrefslogtreecommitdiffstats
path: root/test/pathod/tutils.py
blob: 3a94b6eb0da9911653e1cea83c369fb3214aff45 (plain)
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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
import tempfile
import re
import shutil
import requests
from six.moves import cStringIO as StringIO
from six.moves import urllib
from six import BytesIO

from netlib import tcp
from netlib import utils
from netlib import tutils

from pathod import language
from pathod import pathoc
from pathod import pathod
from pathod import test


def treader(bytes):
    """
        Construct a tcp.Read object from bytes.
    """
    fp = BytesIO(bytes)
    return tcp.Reader(fp)


class DaemonTests(object):
    nohang = False
    ssl = False
    timeout = None
    hexdump = False
    ssloptions = None
    nocraft = False
    explain = True

    @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,
            nohang=cls.nohang,
            timeout=cls.timeout,
            hexdump=cls.hexdump,
            nocraft=cls.nocraft,
            logreq=True,
            logresp=True,
            explain=cls.explain
        )

    @classmethod
    def teardown_class(cls):
        cls.d.shutdown()
        shutil.rmtree(cls.confdir)

    def teardown(self):
        self.d.wait_for_silence()
        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 getpath(self, path, params=None):
        logfp = StringIO()
        c = pathoc.Pathoc(
            ("localhost", self.d.port),
            ssl=self.ssl,
            fp=logfp,
        )
        with c.connect():
            if params:
                path = path + "?" + urllib.parse.urlencode(params)
            resp = c.request("get:%s" % path)
            return resp

    def get(self, spec):
        logfp = StringIO()
        c = pathoc.Pathoc(
            ("localhost", self.d.port),
            ssl=self.ssl,
            fp=logfp,
        )
        with c.connect():
            resp = c.request(
                "get:/p/%s" % urllib.parse.quote(spec)
            )
            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,
        )
        with 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 = tutils.tmpdir

raises = tutils.raises

test_data = utils.Data(__name__)


def render(r, settings=language.Settings()):
    r = r.resolve(settings)
    s = BytesIO()
    assert language.serve(r, s, settings)
    return s.getvalue()