xref: /unit/test/test_usr1.py (revision 1635:97afbb6c5a15)
1import os
2from subprocess import call
3
4from conftest import waitforfiles
5from unit.applications.lang.python import TestApplicationPython
6
7
8class TestUSR1(TestApplicationPython):
9    prerequisites = {'modules': {'python': 'any'}}
10
11    def test_usr1_access_log(self):
12        self.load('empty')
13
14        log = 'access.log'
15        log_new = 'new.log'
16        log_path = self.temp_dir + '/' + log
17
18        assert 'success' in self.conf(
19            '"' + log_path + '"', 'access_log'
20        ), 'access log configure'
21
22        assert waitforfiles(log_path), 'open'
23
24        os.rename(log_path, self.temp_dir + '/' + log_new)
25
26        assert self.get()['status'] == 200
27
28        assert (
29            self.wait_for_record(r'"GET / HTTP/1.1" 200 0 "-" "-"', log_new)
30            is not None
31        ), 'rename new'
32        assert not os.path.isfile(log_path), 'rename old'
33
34        with open(self.temp_dir + '/unit.pid', 'r') as f:
35            pid = f.read().rstrip()
36
37        call(['kill', '-s', 'USR1', pid])
38
39        assert waitforfiles(log_path), 'reopen'
40
41        assert self.get(url='/usr1')['status'] == 200
42
43        self.stop()
44
45        assert (
46            self.wait_for_record(r'"GET /usr1 HTTP/1.1" 200 0 "-" "-"', log)
47            is not None
48        ), 'reopen 2'
49        assert self.search_in_log(r'/usr1', log_new) is None, 'rename new 2'
50
51    def test_usr1_unit_log(self):
52        self.load('log_body')
53
54        log_new = 'new.log'
55        log_path = self.temp_dir + '/unit.log'
56        log_path_new = self.temp_dir + '/' + log_new
57
58        os.rename(log_path, log_path_new)
59
60        body = 'body_for_a_log_new'
61        assert self.post(body=body)['status'] == 200
62
63        assert self.wait_for_record(body, log_new) is not None, 'rename new'
64        assert not os.path.isfile(log_path), 'rename old'
65
66        with open(self.temp_dir + '/unit.pid', 'r') as f:
67            pid = f.read().rstrip()
68
69        call(['kill', '-s', 'USR1', pid])
70
71        assert waitforfiles(log_path), 'reopen'
72
73        body = 'body_for_a_log_unit'
74        assert self.post(body=body)['status'] == 200
75
76        self.stop()
77
78        assert self.wait_for_record(body) is not None, 'rename new'
79        assert self.search_in_log(body, log_new) is None, 'rename new 2'
80
81        # merge two log files into unit.log to check alerts
82
83        with open(log_path, 'w') as unit_log, open(
84            log_path_new, 'r'
85        ) as unit_log_new:
86            unit_log.write(unit_log_new.read())
87