xref: /unit/test/test_node_application.py (revision 2073:bc6ad31ce286) 
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                'Content-Type': 'text/html',
128            },
129            start=True,
130            body=body,
131            read_timeout=1,
132        )
133
134        assert resp['body'] == '0123456789' * 500, 'keep-alive 1'
135
136        body = '0123456789'
137        resp = self.post(
138            headers={
139                'Host': 'localhost',
140                'Connection': 'close',
141                'Content-Type': 'text/html',
142            },
143            sock=sock,
144            body=body,
145        )
146
147        assert resp['body'] == body, 'keep-alive 2'
148
149    def test_node_application_write_buffer(self):
150        self.load('write_buffer')
151
152        assert self.get()['body'] == 'buffer', 'write buffer'
153
154    def test_node_application_write_callback(self, temp_dir):
155        self.load('write_callback')
156
157        assert self.get()['body'] == 'helloworld', 'write callback order'
158        assert waitforfiles(temp_dir + '/node/callback'), 'write callback'
159
160    def test_node_application_write_before_write_head(self):
161        self.load('write_before_write_head')
162
163        assert self.get()['status'] == 200, 'write before writeHead'
164
165    def test_node_application_double_end(self):
166        self.load('double_end')
167
168        assert self.get()['status'] == 200, 'double end'
169        assert self.get()['status'] == 200, 'double end 2'
170
171    def test_node_application_write_return(self):
172        self.load('write_return')
173
174        assert self.get()['body'] == 'bodytrue', 'write return'
175
176    def test_node_application_remove_header(self):
177        self.load('remove_header')
178
179        resp = self.get(
180            headers={
181                'Host': 'localhost',
182                'X-Remove': 'X-Header',
183                'Connection': 'close',
184            }
185        )
186        assert resp['headers']['Was-Header'] == 'true', 'was header'
187        assert resp['headers']['Has-Header'] == 'false', 'has header'
188        assert not ('X-Header' in resp['headers']), 'remove header'
189
190    def test_node_application_remove_header_nonexisting(self):
191        self.load('remove_header')
192
193        assert (
194            self.get(
195                headers={
196                    'Host': 'localhost',
197                    'X-Remove': 'blah',
198                    'Connection': 'close',
199                }
200            )['headers']['Has-Header']
201            == 'true'
202        ), 'remove header nonexisting'
203
204    def test_node_application_update_header(self):
205        self.load('update_header')
206
207        assert self.get()['headers']['X-Header'] == 'new', 'update header'
208
209    def test_node_application_set_header_array(self):
210        self.load('set_header_array')
211
212        assert self.get()['headers']['Set-Cookie'] == [
213            'tc=one,two,three',
214            'tc=four,five,six',
215        ], 'set header array'
216
217    @pytest.mark.skip('not yet')
218    def test_node_application_status_message(self):
219        self.load('status_message')
220
221        assert re.search(r'200 blah', self.get(raw_resp=True)), 'status message'
222
223    def test_node_application_get_header_type(self):
224        self.load('get_header_type')
225
226        assert self.get()['headers']['X-Type'] == 'number', 'get header type'
227
228    def test_node_application_header_name_case(self):
229        self.load('header_name_case')
230
231        headers = self.get()['headers']
232
233        assert headers['X-HEADER'] == '3', 'header value'
234        assert 'X-Header' not in headers, 'insensitive'
235        assert 'X-header' not in headers, 'insensitive 2'
236
237    def test_node_application_promise_handler_write_after_end(self):
238        self.load('promise_handler')
239
240        assert (
241            self.post(
242                headers={
243                    'Host': 'localhost',
244                    'Content-Type': 'text/html',
245                    'X-Write-Call': '1',
246                    'Connection': 'close',
247                },
248                body='callback',
249            )['status']
250            == 200
251        ), 'promise handler request write after end'
252
253    def test_node_application_promise_end(self, temp_dir):
254        self.load('promise_end')
255
256        assert (
257            self.post(
258                headers={
259                    'Host': 'localhost',
260                    'Content-Type': 'text/html',
261                    'Connection': 'close',
262                },
263                body='end',
264            )['status']
265            == 200
266        ), 'promise end request'
267        assert waitforfiles(temp_dir + '/node/callback'), 'promise end'
268
269    @pytest.mark.skip('not yet')
270    def test_node_application_header_name_valid(self):
271        self.load('header_name_valid')
272
273        assert 'status' not in self.get(), 'header name valid'
274
275    def test_node_application_header_value_object(self):
276        self.load('header_value_object')
277
278        assert 'X-Header' in self.get()['headers'], 'header value object'
279
280    def test_node_application_get_header_names(self):
281        self.load('get_header_names')
282
283        assert self.get()['headers']['X-Names'] == [
284            'date',
285            'x-header',
286        ], 'get header names'
287
288    def test_node_application_has_header(self):
289        self.load('has_header')
290
291        assert (
292            self.get(
293                headers={
294                    'Host': 'localhost',
295                    'X-Header': 'length',
296                    'Connection': 'close',
297                }
298            )['headers']['X-Has-Header']
299            == 'false'
300        ), 'has header length'
301
302        assert (
303            self.get(
304                headers={
305                    'Host': 'localhost',
306                    'X-Header': 'Date',
307                    'Connection': 'close',
308                }
309            )['headers']['X-Has-Header']
310            == 'false'
311        ), 'has header date'
312
313    def test_node_application_write_multiple(self):
314        self.load('write_multiple')
315
316        assert self.get()['body'] == 'writewrite2end', 'write multiple'
317