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