xref: /unit/test/test_node_application.py (revision 2330:4b1f175f9c88)
1import re
2
3import pytest
4from unit.applications.lang.node import TestApplicationNode
5from unit.utils import waitforfiles
6
7
8class TestNodeApplication(TestApplicationNode):
9    prerequisites = {'modules': {'node': 'all'}}
10
11    def assert_basic_application(self):
12        resp = self.get()
13        assert resp['headers']['Content-Type'] == 'text/plain', 'basic header'
14        assert resp['body'] == 'Hello World\n', 'basic body'
15
16    def test_node_application_basic(self):
17        self.load('basic')
18
19        self.assert_basic_application()
20
21    def test_node_application_loader_unit_http(self):
22        self.load('loader/unit_http')
23
24        self.assert_basic_application()
25
26    def test_node_application_loader_transitive_dependency(self):
27        self.load('loader/transitive_dependency')
28
29        self.assert_basic_application()
30
31    def test_node_application_seq(self):
32        self.load('basic')
33
34        assert self.get()['status'] == 200, 'seq'
35        assert self.get()['status'] == 200, 'seq 2'
36
37    def test_node_application_variables(self):
38        self.load('variables')
39
40        body = 'Test body string.'
41
42        resp = self.post(
43            headers={
44                'Host': 'localhost',
45                'Content-Type': 'text/html',
46                'Custom-Header': 'blah',
47                'Connection': 'close',
48            },
49            body=body,
50        )
51
52        assert resp['status'] == 200, 'status'
53        headers = resp['headers']
54        header_server = headers.pop('Server')
55        assert re.search(r'Unit/[\d\.]+', header_server), 'server header'
56
57        date = headers.pop('Date')
58        assert date[-4:] == ' GMT', 'date header timezone'
59        assert (
60            abs(self.date_to_sec_epoch(date) - self.sec_epoch()) < 5
61        ), 'date header'
62
63        raw_headers = headers.pop('Request-Raw-Headers')
64        assert re.search(
65            r'^(?:Host|localhost|Content-Type|'
66            r'text\/html|Custom-Header|blah|Content-Length|17|Connection|'
67            r'close|,)+$',
68            raw_headers,
69        ), 'raw headers'
70
71        assert headers == {
72            'Connection': 'close',
73            'Content-Length': str(len(body)),
74            'Content-Type': 'text/html',
75            'Request-Method': 'POST',
76            'Request-Uri': '/',
77            'Http-Host': 'localhost',
78            'Server-Protocol': 'HTTP/1.1',
79            'Custom-Header': 'blah',
80        }, 'headers'
81        assert resp['body'] == body, 'body'
82
83    def test_node_application_get_variables(self):
84        self.load('get_variables')
85
86        resp = self.get(url='/?var1=val1&var2=&var3')
87        assert resp['headers']['X-Var-1'] == 'val1', 'GET variables'
88        assert resp['headers']['X-Var-2'] == '', 'GET variables 2'
89        assert resp['headers']['X-Var-3'] == '', 'GET variables 3'
90
91    def test_node_application_post_variables(self):
92        self.load('post_variables')
93
94        resp = self.post(
95            headers={
96                'Content-Type': 'application/x-www-form-urlencoded',
97                'Host': 'localhost',
98                'Connection': 'close',
99            },
100            body='var1=val1&var2=&var3',
101        )
102
103        assert resp['headers']['X-Var-1'] == 'val1', 'POST variables'
104        assert resp['headers']['X-Var-2'] == '', 'POST variables 2'
105        assert resp['headers']['X-Var-3'] == '', 'POST variables 3'
106
107    def test_node_application_404(self):
108        self.load('404')
109
110        resp = self.get()
111
112        assert resp['status'] == 404, '404 status'
113        assert re.search(
114            r'<title>404 Not Found</title>', resp['body']
115        ), '404 body'
116
117    def test_node_keepalive_body(self):
118        self.load('mirror')
119
120        assert self.get()['status'] == 200, 'init'
121
122        body = '0123456789' * 500
123        (resp, sock) = self.post(
124            headers={
125                'Host': 'localhost',
126                'Connection': 'keep-alive',
127            },
128            start=True,
129            body=body,
130            read_timeout=1,
131        )
132
133        assert resp['body'] == '0123456789' * 500, 'keep-alive 1'
134
135        body = '0123456789'
136        resp = self.post(sock=sock, body=body)
137
138        assert resp['body'] == body, 'keep-alive 2'
139
140    def test_node_application_write_buffer(self):
141        self.load('write_buffer')
142
143        assert self.get()['body'] == 'buffer', 'write buffer'
144
145    def test_node_application_write_callback(self, temp_dir):
146        self.load('write_callback')
147
148        assert self.get()['body'] == 'helloworld', 'write callback order'
149        assert waitforfiles(f'{temp_dir}/node/callback'), 'write callback'
150
151    def test_node_application_write_before_write_head(self):
152        self.load('write_before_write_head')
153
154        assert self.get()['status'] == 200, 'write before writeHead'
155
156    def test_node_application_double_end(self):
157        self.load('double_end')
158
159        assert self.get()['status'] == 200, 'double end'
160        assert self.get()['status'] == 200, 'double end 2'
161
162    def test_node_application_write_return(self):
163        self.load('write_return')
164
165        assert self.get()['body'] == 'bodytrue', 'write return'
166
167    def test_node_application_remove_header(self):
168        self.load('remove_header')
169
170        resp = self.get(
171            headers={
172                'Host': 'localhost',
173                'X-Remove': 'X-Header',
174                'Connection': 'close',
175            }
176        )
177        assert resp['headers']['Was-Header'] == 'true', 'was header'
178        assert resp['headers']['Has-Header'] == 'false', 'has header'
179        assert not ('X-Header' in resp['headers']), 'remove header'
180
181    def test_node_application_remove_header_nonexisting(self):
182        self.load('remove_header')
183
184        assert (
185            self.get(
186                headers={
187                    'Host': 'localhost',
188                    'X-Remove': 'blah',
189                    'Connection': 'close',
190                }
191            )['headers']['Has-Header']
192            == 'true'
193        ), 'remove header nonexisting'
194
195    def test_node_application_update_header(self):
196        self.load('update_header')
197
198        assert self.get()['headers']['X-Header'] == 'new', 'update header'
199
200    def test_node_application_set_header_array(self):
201        self.load('set_header_array')
202
203        assert self.get()['headers']['Set-Cookie'] == [
204            'tc=one,two,three',
205            'tc=four,five,six',
206        ], 'set header array'
207
208    @pytest.mark.skip('not yet')
209    def test_node_application_status_message(self):
210        self.load('status_message')
211
212        assert re.search(r'200 blah', self.get(raw_resp=True)), 'status message'
213
214    def test_node_application_get_header_type(self):
215        self.load('get_header_type')
216
217        assert self.get()['headers']['X-Type'] == 'number', 'get header type'
218
219    def test_node_application_header_name_case(self):
220        self.load('header_name_case')
221
222        headers = self.get()['headers']
223
224        assert headers['X-HEADER'] == '3', 'header value'
225        assert 'X-Header' not in headers, 'insensitive'
226        assert 'X-header' not in headers, 'insensitive 2'
227
228    def test_node_application_promise_handler_write_after_end(self):
229        self.load('promise_handler')
230
231        assert (
232            self.post(
233                headers={
234                    'Host': 'localhost',
235                    'Content-Type': 'text/html',
236                    'X-Write-Call': '1',
237                    'Connection': 'close',
238                },
239                body='callback',
240            )['status']
241            == 200
242        ), 'promise handler request write after end'
243
244    def test_node_application_promise_end(self, temp_dir):
245        self.load('promise_end')
246
247        assert (
248            self.post(
249                headers={
250                    'Host': 'localhost',
251                    'Content-Type': 'text/html',
252                    'Connection': 'close',
253                },
254                body='end',
255            )['status']
256            == 200
257        ), 'promise end request'
258        assert waitforfiles(f'{temp_dir}/node/callback'), 'promise end'
259
260    @pytest.mark.skip('not yet')
261    def test_node_application_header_name_valid(self):
262        self.load('header_name_valid')
263
264        assert 'status' not in self.get(), 'header name valid'
265
266    def test_node_application_header_value_object(self):
267        self.load('header_value_object')
268
269        assert 'X-Header' in self.get()['headers'], 'header value object'
270
271    def test_node_application_get_header_names(self):
272        self.load('get_header_names')
273
274        assert self.get()['headers']['X-Names'] == [
275            'date',
276            'x-header',
277        ], 'get header names'
278
279    def test_node_application_has_header(self):
280        self.load('has_header')
281
282        assert (
283            self.get(
284                headers={
285                    'Host': 'localhost',
286                    'X-Header': 'length',
287                    'Connection': 'close',
288                }
289            )['headers']['X-Has-Header']
290            == 'false'
291        ), 'has header length'
292
293        assert (
294            self.get(
295                headers={
296                    'Host': 'localhost',
297                    'X-Header': 'Date',
298                    'Connection': 'close',
299                }
300            )['headers']['X-Has-Header']
301            == 'false'
302        ), 'has header date'
303
304    def test_node_application_write_multiple(self):
305        self.load('write_multiple')
306
307        assert self.get()['body'] == 'writewrite2end', 'write multiple'
308