xref: /unit/test/test_php_application.py (revision 1730:44912af5b3e6)
1import os
2import re
3import shutil
4import time
5
6import pytest
7
8from conftest import unit_stop
9from unit.applications.lang.php import TestApplicationPHP
10from unit.option import option
11
12class TestPHPApplication(TestApplicationPHP):
13    prerequisites = {'modules': {'php': 'all'}}
14
15    def before_disable_functions(self):
16        body = self.get()['body']
17
18        assert re.search(r'time: \d+', body), 'disable_functions before time'
19        assert re.search(r'exec: \/\w+', body), 'disable_functions before exec'
20
21    def set_opcache(self, app, val):
22        assert 'success' in self.conf(
23            {"admin": {"opcache.enable": val, "opcache.enable_cli": val,},},
24            'applications/' + app + '/options',
25        )
26
27        opcache = self.get()['headers']['X-OPcache']
28
29        if not opcache or opcache == '-1':
30            pytest.skip('opcache is not supported')
31
32        assert opcache == val, 'opcache value'
33
34    def test_php_application_variables(self):
35        self.load('variables')
36
37        body = 'Test body string.'
38
39        resp = self.post(
40            headers={
41                'Host': 'localhost',
42                'Content-Type': 'text/html',
43                'Custom-Header': 'blah',
44                'Connection': 'close',
45            },
46            body=body,
47            url='/index.php/blah?var=val',
48        )
49
50        assert resp['status'] == 200, 'status'
51        headers = resp['headers']
52        header_server = headers.pop('Server')
53        assert re.search(r'Unit/[\d\.]+', header_server), 'server header'
54        assert (
55            headers.pop('Server-Software') == header_server
56        ), 'server software 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        if 'X-Powered-By' in headers:
65            headers.pop('X-Powered-By')
66
67        headers.pop('Content-type')
68        assert headers == {
69            'Connection': 'close',
70            'Content-Length': str(len(body)),
71            'Request-Method': 'POST',
72            'Path-Info': '/blah',
73            'Request-Uri': '/index.php/blah?var=val',
74            'Http-Host': 'localhost',
75            'Server-Protocol': 'HTTP/1.1',
76            'Custom-Header': 'blah',
77        }, 'headers'
78        assert resp['body'] == body, 'body'
79
80    def test_php_application_query_string(self):
81        self.load('query_string')
82
83        resp = self.get(url='/?var1=val1&var2=val2')
84
85        assert (
86            resp['headers']['Query-String'] == 'var1=val1&var2=val2'
87        ), 'query string'
88
89    def test_php_application_query_string_empty(self):
90        self.load('query_string')
91
92        resp = self.get(url='/?')
93
94        assert resp['status'] == 200, 'query string empty status'
95        assert resp['headers']['Query-String'] == '', 'query string empty'
96
97    def test_php_application_fastcgi_finish_request(self, temp_dir):
98        self.load('fastcgi_finish_request')
99
100        assert self.get()['body'] == '0123'
101
102        unit_stop()
103
104        with open(temp_dir + '/unit.log', 'r', errors='ignore') as f:
105            errs = re.findall(r'Error in fastcgi_finish_request', f.read())
106
107            assert len(errs) == 0, 'no error'
108
109    def test_php_application_fastcgi_finish_request_2(self, temp_dir):
110        self.load('fastcgi_finish_request')
111
112        resp = self.get(url='/?skip')
113        assert resp['status'] == 200
114        assert resp['body'] == ''
115
116        unit_stop()
117
118        with open(temp_dir + '/unit.log', 'r', errors='ignore') as f:
119            errs = re.findall(r'Error in fastcgi_finish_request', f.read())
120
121            assert len(errs) == 0, 'no error'
122
123    def test_php_application_query_string_absent(self):
124        self.load('query_string')
125
126        resp = self.get()
127
128        assert resp['status'] == 200, 'query string absent status'
129        assert resp['headers']['Query-String'] == '', 'query string absent'
130
131    def test_php_application_phpinfo(self):
132        self.load('phpinfo')
133
134        resp = self.get()
135
136        assert resp['status'] == 200, 'status'
137        assert resp['body'] != '', 'body not empty'
138
139    def test_php_application_header_status(self):
140        self.load('header')
141
142        assert (
143            self.get(
144                headers={
145                    'Host': 'localhost',
146                    'Connection': 'close',
147                    'X-Header': 'HTTP/1.1 404 Not Found',
148                }
149            )['status']
150            == 404
151        ), 'status'
152
153        assert (
154            self.get(
155                headers={
156                    'Host': 'localhost',
157                    'Connection': 'close',
158                    'X-Header': 'http/1.1 404 Not Found',
159                }
160            )['status']
161            == 404
162        ), 'status case insensitive'
163
164        assert (
165            self.get(
166                headers={
167                    'Host': 'localhost',
168                    'Connection': 'close',
169                    'X-Header': 'HTTP/ 404 Not Found',
170                }
171            )['status']
172            == 404
173        ), 'status version empty'
174
175    def test_php_application_404(self):
176        self.load('404')
177
178        resp = self.get()
179
180        assert resp['status'] == 404, '404 status'
181        assert re.search(
182            r'<title>404 Not Found</title>', resp['body']
183        ), '404 body'
184
185    def test_php_application_keepalive_body(self):
186        self.load('mirror')
187
188        assert self.get()['status'] == 200, 'init'
189
190        body = '0123456789' * 500
191        (resp, sock) = self.post(
192            headers={
193                'Host': 'localhost',
194                'Connection': 'keep-alive',
195                'Content-Type': 'text/html',
196            },
197            start=True,
198            body=body,
199            read_timeout=1,
200        )
201
202        assert resp['body'] == body, 'keep-alive 1'
203
204        body = '0123456789'
205        resp = self.post(
206            headers={
207                'Host': 'localhost',
208                'Connection': 'close',
209                'Content-Type': 'text/html',
210            },
211            sock=sock,
212            body=body,
213        )
214
215        assert resp['body'] == body, 'keep-alive 2'
216
217    def test_php_application_conditional(self):
218        self.load('conditional')
219
220        assert re.search(r'True', self.get()['body']), 'conditional true'
221        assert re.search(r'False', self.post()['body']), 'conditional false'
222
223    def test_php_application_get_variables(self):
224        self.load('get_variables')
225
226        resp = self.get(url='/?var1=val1&var2=&var3')
227        assert resp['headers']['X-Var-1'] == 'val1', 'GET variables'
228        assert resp['headers']['X-Var-2'] == '', 'GET variables 2'
229        assert resp['headers']['X-Var-3'] == '', 'GET variables 3'
230        assert resp['headers']['X-Var-4'] == 'not set', 'GET variables 4'
231
232    def test_php_application_post_variables(self):
233        self.load('post_variables')
234
235        resp = self.post(
236            headers={
237                'Content-Type': 'application/x-www-form-urlencoded',
238                'Host': 'localhost',
239                'Connection': 'close',
240            },
241            body='var1=val1&var2=',
242        )
243        assert resp['headers']['X-Var-1'] == 'val1', 'POST variables'
244        assert resp['headers']['X-Var-2'] == '', 'POST variables 2'
245        assert resp['headers']['X-Var-3'] == 'not set', 'POST variables 3'
246
247    def test_php_application_cookies(self):
248        self.load('cookies')
249
250        resp = self.get(
251            headers={
252                'Cookie': 'var=val; var2=val2',
253                'Host': 'localhost',
254                'Connection': 'close',
255            }
256        )
257
258        assert resp['headers']['X-Cookie-1'] == 'val', 'cookie'
259        assert resp['headers']['X-Cookie-2'] == 'val2', 'cookie'
260
261    def test_php_application_ini_precision(self):
262        self.load('ini_precision')
263
264        assert self.get()['headers']['X-Precision'] != '4', 'ini value default'
265
266        self.conf(
267            {"file": "ini/php.ini"}, 'applications/ini_precision/options'
268        )
269
270        assert (
271            self.get()['headers']['X-File']
272            == option.test_dir + '/php/ini_precision/ini/php.ini'
273        ), 'ini file'
274        assert self.get()['headers']['X-Precision'] == '4', 'ini value'
275
276    @pytest.mark.skip('not yet')
277    def test_php_application_ini_admin_user(self):
278        self.load('ini_precision')
279
280        assert 'error' in self.conf(
281            {"user": {"precision": "4"}, "admin": {"precision": "5"}},
282            'applications/ini_precision/options',
283        ), 'ini admin user'
284
285    def test_php_application_ini_admin(self):
286        self.load('ini_precision')
287
288        self.conf(
289            {"file": "php.ini", "admin": {"precision": "5"}},
290            'applications/ini_precision/options',
291        )
292
293        assert self.get()['headers']['X-Precision'] == '5', 'ini value admin'
294
295    def test_php_application_ini_user(self):
296        self.load('ini_precision')
297
298        self.conf(
299            {"file": "php.ini", "user": {"precision": "5"}},
300            'applications/ini_precision/options',
301        )
302
303        assert self.get()['headers']['X-Precision'] == '5', 'ini value user'
304
305    def test_php_application_ini_user_2(self):
306        self.load('ini_precision')
307
308        self.conf(
309            {"file": "ini/php.ini"}, 'applications/ini_precision/options'
310        )
311
312        assert self.get()['headers']['X-Precision'] == '4', 'ini user file'
313
314        self.conf(
315            {"precision": "5"}, 'applications/ini_precision/options/user'
316        )
317
318        assert self.get()['headers']['X-Precision'] == '5', 'ini value user'
319
320    def test_php_application_ini_set_admin(self):
321        self.load('ini_precision')
322
323        self.conf(
324            {"admin": {"precision": "5"}}, 'applications/ini_precision/options'
325        )
326
327        assert (
328            self.get(url='/?precision=6')['headers']['X-Precision'] == '5'
329        ), 'ini set admin'
330
331    def test_php_application_ini_set_user(self):
332        self.load('ini_precision')
333
334        self.conf(
335            {"user": {"precision": "5"}}, 'applications/ini_precision/options'
336        )
337
338        assert (
339            self.get(url='/?precision=6')['headers']['X-Precision'] == '6'
340        ), 'ini set user'
341
342    def test_php_application_ini_repeat(self):
343        self.load('ini_precision')
344
345        self.conf(
346            {"user": {"precision": "5"}}, 'applications/ini_precision/options'
347        )
348
349        assert self.get()['headers']['X-Precision'] == '5', 'ini value'
350
351        assert self.get()['headers']['X-Precision'] == '5', 'ini value repeat'
352
353    def test_php_application_disable_functions_exec(self):
354        self.load('time_exec')
355
356        self.before_disable_functions()
357
358        self.conf(
359            {"admin": {"disable_functions": "exec"}},
360            'applications/time_exec/options',
361        )
362
363        body = self.get()['body']
364
365        assert re.search(r'time: \d+', body), 'disable_functions time'
366        assert not re.search(r'exec: \/\w+', body), 'disable_functions exec'
367
368    def test_php_application_disable_functions_comma(self):
369        self.load('time_exec')
370
371        self.before_disable_functions()
372
373        self.conf(
374            {"admin": {"disable_functions": "exec,time"}},
375            'applications/time_exec/options',
376        )
377
378        body = self.get()['body']
379
380        assert not re.search(
381            r'time: \d+', body
382        ), 'disable_functions comma time'
383        assert not re.search(
384            r'exec: \/\w+', body
385        ), 'disable_functions comma exec'
386
387    def test_php_application_disable_functions_space(self):
388        self.load('time_exec')
389
390        self.before_disable_functions()
391
392        self.conf(
393            {"admin": {"disable_functions": "exec time"}},
394            'applications/time_exec/options',
395        )
396
397        body = self.get()['body']
398
399        assert not re.search(
400            r'time: \d+', body
401        ), 'disable_functions space time'
402        assert not re.search(
403            r'exec: \/\w+', body
404        ), 'disable_functions space exec'
405
406    def test_php_application_disable_functions_user(self):
407        self.load('time_exec')
408
409        self.before_disable_functions()
410
411        self.conf(
412            {"user": {"disable_functions": "exec"}},
413            'applications/time_exec/options',
414        )
415
416        body = self.get()['body']
417
418        assert re.search(r'time: \d+', body), 'disable_functions user time'
419        assert not re.search(
420            r'exec: \/\w+', body
421        ), 'disable_functions user exec'
422
423    def test_php_application_disable_functions_nonexistent(self):
424        self.load('time_exec')
425
426        self.before_disable_functions()
427
428        self.conf(
429            {"admin": {"disable_functions": "blah"}},
430            'applications/time_exec/options',
431        )
432
433        body = self.get()['body']
434
435        assert re.search(
436            r'time: \d+', body
437        ), 'disable_functions nonexistent time'
438        assert re.search(
439            r'exec: \/\w+', body
440        ), 'disable_functions nonexistent exec'
441
442    def test_php_application_disable_classes(self):
443        self.load('date_time')
444
445        assert re.search(
446            r'012345', self.get()['body']
447        ), 'disable_classes before'
448
449        self.conf(
450            {"admin": {"disable_classes": "DateTime"}},
451            'applications/date_time/options',
452        )
453
454        assert not re.search(
455            r'012345', self.get()['body']
456        ), 'disable_classes before'
457
458    def test_php_application_disable_classes_user(self):
459        self.load('date_time')
460
461        assert re.search(
462            r'012345', self.get()['body']
463        ), 'disable_classes before'
464
465        self.conf(
466            {"user": {"disable_classes": "DateTime"}},
467            'applications/date_time/options',
468        )
469
470        assert not re.search(
471            r'012345', self.get()['body']
472        ), 'disable_classes before'
473
474    def test_php_application_error_log(self, temp_dir):
475        self.load('error_log')
476
477        assert self.get()['status'] == 200, 'status'
478
479        time.sleep(1)
480
481        assert self.get()['status'] == 200, 'status 2'
482
483        unit_stop()
484
485        pattern = r'\d{4}\/\d\d\/\d\d\s\d\d:.+\[notice\].+Error in application'
486
487        assert self.wait_for_record(pattern) is not None, 'errors print'
488
489        with open(temp_dir + '/unit.log', 'r', errors='ignore') as f:
490            errs = re.findall(pattern, f.read())
491
492            assert len(errs) == 2, 'error_log count'
493
494            date = errs[0].split('[')[0]
495            date2 = errs[1].split('[')[0]
496            assert date != date2, 'date diff'
497
498    def test_php_application_script(self):
499        assert 'success' in self.conf(
500            {
501                "listeners": {"*:7080": {"pass": "applications/script"}},
502                "applications": {
503                    "script": {
504                        "type": "php",
505                        "processes": {"spare": 0},
506                        "root": option.test_dir + "/php/script",
507                        "script": "phpinfo.php",
508                    }
509                },
510            }
511        ), 'configure script'
512
513        resp = self.get()
514
515        assert resp['status'] == 200, 'status'
516        assert resp['body'] != '', 'body not empty'
517
518    def test_php_application_index_default(self):
519        assert 'success' in self.conf(
520            {
521                "listeners": {"*:7080": {"pass": "applications/phpinfo"}},
522                "applications": {
523                    "phpinfo": {
524                        "type": "php",
525                        "processes": {"spare": 0},
526                        "root": option.test_dir + "/php/phpinfo",
527                    }
528                },
529            }
530        ), 'configure index default'
531
532        resp = self.get()
533
534        assert resp['status'] == 200, 'status'
535        assert resp['body'] != '', 'body not empty'
536
537    def test_php_application_extension_check(self, temp_dir):
538        self.load('phpinfo')
539
540        assert self.get(url='/index.wrong')['status'] != 200, 'status'
541
542        new_root = temp_dir + "/php"
543        os.mkdir(new_root)
544        shutil.copy(option.test_dir + '/php/phpinfo/index.wrong', new_root)
545
546        assert 'success' in self.conf(
547            {
548                "listeners": {"*:7080": {"pass": "applications/phpinfo"}},
549                "applications": {
550                    "phpinfo": {
551                        "type": "php",
552                        "processes": {"spare": 0},
553                        "root": new_root,
554                        "working_directory": new_root,
555                    }
556                },
557            }
558        ), 'configure new root'
559
560        resp = self.get()
561        assert str(resp['status']) + resp['body'] != '200', 'status new root'
562
563    def run_php_application_cwd_root_tests(self):
564        assert 'success' in self.conf_delete(
565            'applications/cwd/working_directory'
566        )
567
568        script_cwd = option.test_dir + '/php/cwd'
569
570        resp = self.get()
571        assert resp['status'] == 200, 'status ok'
572        assert resp['body'] == script_cwd, 'default cwd'
573
574        assert 'success' in self.conf(
575            '"' + option.test_dir + '"', 'applications/cwd/working_directory',
576        )
577
578        resp = self.get()
579        assert resp['status'] == 200, 'status ok'
580        assert resp['body'] == script_cwd, 'wdir cwd'
581
582        resp = self.get(url='/?chdir=/')
583        assert resp['status'] == 200, 'status ok'
584        assert resp['body'] == '/', 'cwd after chdir'
585
586        # cwd must be restored
587
588        resp = self.get()
589        assert resp['status'] == 200, 'status ok'
590        assert resp['body'] == script_cwd, 'cwd restored'
591
592        resp = self.get(url='/subdir/')
593        assert resp['body'] == script_cwd + '/subdir', 'cwd subdir'
594
595    def test_php_application_cwd_root(self):
596        self.load('cwd')
597        self.run_php_application_cwd_root_tests()
598
599    def test_php_application_cwd_opcache_disabled(self):
600        self.load('cwd')
601        self.set_opcache('cwd', '0')
602        self.run_php_application_cwd_root_tests()
603
604    def test_php_application_cwd_opcache_enabled(self):
605        self.load('cwd')
606        self.set_opcache('cwd', '1')
607        self.run_php_application_cwd_root_tests()
608
609    def run_php_application_cwd_script_tests(self):
610        self.load('cwd')
611
612        script_cwd = option.test_dir + '/php/cwd'
613
614        assert 'success' in self.conf_delete(
615            'applications/cwd/working_directory'
616        )
617
618        assert 'success' in self.conf('"index.php"', 'applications/cwd/script')
619
620        assert self.get()['body'] == script_cwd, 'default cwd'
621
622        assert self.get(url='/?chdir=/')['body'] == '/', 'cwd after chdir'
623
624        # cwd must be restored
625        assert self.get()['body'] == script_cwd, 'cwd restored'
626
627    def test_php_application_cwd_script(self):
628        self.load('cwd')
629        self.run_php_application_cwd_script_tests()
630
631    def test_php_application_cwd_script_opcache_disabled(self):
632        self.load('cwd')
633        self.set_opcache('cwd', '0')
634        self.run_php_application_cwd_script_tests()
635
636    def test_php_application_cwd_script_opcache_enabled(self):
637        self.load('cwd')
638        self.set_opcache('cwd', '1')
639        self.run_php_application_cwd_script_tests()
640
641    def test_php_application_path_relative(self):
642        self.load('open')
643
644        assert self.get()['body'] == 'test', 'relative path'
645
646        assert (
647            self.get(url='/?chdir=/')['body'] != 'test'
648        ), 'relative path w/ chdir'
649
650        assert self.get()['body'] == 'test', 'relative path 2'
651