xref: /unit/test/test_node_application.py (revision 1971:3410f9d2a662) 
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(
222            r'200 blah', self.get(raw_resp=True)
223        ), 'status message'
224
225    def test_node_application_get_header_type(self):
226        self.load('get_header_type')
227
228        assert self.get()['headers']['X-Type'] == 'number', 'get header type'
229
230    def test_node_application_header_name_case(self):
231        self.load('header_name_case')
232
233        headers = self.get()['headers']
234
235        assert headers['X-HEADER'] == '3', 'header value'
236        assert 'X-Header' not in headers, 'insensitive'
237        assert 'X-header' not in headers, 'insensitive 2'
238
239    def test_node_application_promise_handler_write_after_end(self):
240        self.load('promise_handler')
241
242        assert (
243            self.post(
244                headers={
245                    'Host': 'localhost',
246                    'Content-Type': 'text/html',
247                    'X-Write-Call': '1',
248                    'Connection': 'close',
249                },
250                body='callback',
251            )['status']
252            == 200
253        ), 'promise handler request write after end'
254
255    def test_node_application_promise_end(self, temp_dir):
256        self.load('promise_end')
257
258        assert (
259            self.post(
260                headers={
261                    'Host': 'localhost',
262                    'Content-Type': 'text/html',
263                    'Connection': 'close',
264                },
265                body='end',
266            )['status']
267            == 200
268        ), 'promise end request'
269        assert waitforfiles(temp_dir + '/node/callback'), 'promise end'
270
271    @pytest.mark.skip('not yet')
272    def test_node_application_header_name_valid(self):
273        self.load('header_name_valid')
274
275        assert 'status' not in self.get(), 'header name valid'
276
277    def test_node_application_header_value_object(self):
278        self.load('header_value_object')
279
280        assert 'X-Header' in self.get()['headers'], 'header value object'
281
282    def test_node_application_get_header_names(self):
283        self.load('get_header_names')
284
285        assert self.get()['headers']['X-Names'] == [
286            'date',
287            'x-header',
288        ], 'get header names'
289
290    def test_node_application_has_header(self):
291        self.load('has_header')
292
293        assert (
294            self.get(
295                headers={
296                    'Host': 'localhost',
297                    'X-Header': 'length',
298                    'Connection': 'close',
299                }
300            )['headers']['X-Has-Header']
301            == 'false'
302        ), 'has header length'
303
304        assert (
305            self.get(
306                headers={
307                    'Host': 'localhost',
308                    'X-Header': 'Date',
309                    'Connection': 'close',
310                }
311            )['headers']['X-Has-Header']
312            == 'false'
313        ), 'has header date'
314
315    def test_node_application_write_multiple(self):
316        self.load('write_multiple')
317
318        assert self.get()['body'] == 'writewrite2end', 'write multiple'
319