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