diff options
Diffstat (limited to 'test/test_usr1.py')
-rw-r--r-- | test/test_usr1.py | 110 |
1 files changed, 55 insertions, 55 deletions
diff --git a/test/test_usr1.py b/test/test_usr1.py index 4bff0242..ce756fc0 100644 --- a/test/test_usr1.py +++ b/test/test_usr1.py @@ -1,87 +1,87 @@ import os import signal -from unit.applications.lang.python import TestApplicationPython +from unit.applications.lang.python import ApplicationPython from unit.log import Log from unit.utils import waitforfiles +prerequisites = {'modules': {'python': 'any'}} -class TestUSR1(TestApplicationPython): - prerequisites = {'modules': {'python': 'any'}} +client = ApplicationPython() - def test_usr1_access_log(self, temp_dir, unit_pid): - self.load('empty') - log = 'access.log' - log_new = 'new.log' - log_path = f'{temp_dir}/{log}' +def test_usr1_access_log(search_in_file, temp_dir, unit_pid, wait_for_record): + client.load('empty') - assert 'success' in self.conf( - f'"{log_path}"', 'access_log' - ), 'access log configure' + log = 'access.log' + log_new = 'new.log' + log_path = f'{temp_dir}/{log}' - assert waitforfiles(log_path), 'open' + assert 'success' in client.conf( + f'"{log_path}"', 'access_log' + ), 'access log configure' - os.rename(log_path, f'{temp_dir}/{log_new}') + assert waitforfiles(log_path), 'open' - assert self.get()['status'] == 200 + os.rename(log_path, f'{temp_dir}/{log_new}') - assert ( - self.wait_for_record(r'"GET / HTTP/1.1" 200 0 "-" "-"', log_new) - is not None - ), 'rename new' - assert not os.path.isfile(log_path), 'rename old' + assert client.get()['status'] == 200 - os.kill(unit_pid, signal.SIGUSR1) + assert ( + wait_for_record(r'"GET / HTTP/1.1" 200 0 "-" "-"', log_new) is not None + ), 'rename new' + assert not os.path.isfile(log_path), 'rename old' - assert waitforfiles(log_path), 'reopen' + os.kill(unit_pid, signal.SIGUSR1) - assert self.get(url='/usr1')['status'] == 200 + assert waitforfiles(log_path), 'reopen' - assert ( - self.wait_for_record(r'"GET /usr1 HTTP/1.1" 200 0 "-" "-"', log) - is not None - ), 'reopen 2' - assert self.search_in_log(r'/usr1', log_new) is None, 'rename new 2' + assert client.get(url='/usr1')['status'] == 200 - def test_usr1_unit_log(self, temp_dir, unit_pid): - self.load('log_body') + assert ( + wait_for_record(r'"GET /usr1 HTTP/1.1" 200 0 "-" "-"', log) is not None + ), 'reopen 2' + assert search_in_file(r'/usr1', log_new) is None, 'rename new 2' - log_new = 'new.log' - log_path = f'{temp_dir}/unit.log' - log_path_new = f'{temp_dir}/{log_new}' - os.rename(log_path, log_path_new) +def test_usr1_unit_log(search_in_file, temp_dir, unit_pid, wait_for_record): + client.load('log_body') - Log.swap(log_new) + log_new = 'new.log' + log_path = f'{temp_dir}/unit.log' + log_path_new = f'{temp_dir}/{log_new}' + + os.rename(log_path, log_path_new) - try: - body = 'body_for_a_log_new\n' - assert self.post(body=body)['status'] == 200 + Log.swap(log_new) - assert self.wait_for_record(body, log_new) is not None, 'rename new' - assert not os.path.isfile(log_path), 'rename old' + try: + body = 'body_for_a_log_new\n' + assert client.post(body=body)['status'] == 200 - os.kill(unit_pid, signal.SIGUSR1) + assert wait_for_record(body, log_new) is not None, 'rename new' + assert not os.path.isfile(log_path), 'rename old' - assert waitforfiles(log_path), 'reopen' + os.kill(unit_pid, signal.SIGUSR1) - body = 'body_for_a_log_unit\n' - assert self.post(body=body)['status'] == 200 + assert waitforfiles(log_path), 'reopen' - assert self.wait_for_record(body) is not None, 'rename new' - assert self.search_in_log(body, log_new) is None, 'rename new 2' + body = 'body_for_a_log_unit\n' + assert client.post(body=body)['status'] == 200 - finally: - # merge two log files into unit.log to check alerts + assert wait_for_record(body) is not None, 'rename new' + assert search_in_file(body, log_new) is None, 'rename new 2' - with open(log_path, 'r', errors='ignore') as unit_log: - log = unit_log.read() + finally: + # merge two log files into unit.log to check alerts - with open(log_path, 'w') as unit_log, open( - log_path_new, 'r', errors='ignore' - ) as unit_log_new: - unit_log.write(unit_log_new.read()) - unit_log.write(log) + with open(log_path, 'r', errors='ignore') as unit_log: + log = unit_log.read() - Log.swap(log_new) + with open(log_path, 'w') as unit_log, open( + log_path_new, 'r', errors='ignore' + ) as unit_log_new: + unit_log.write(unit_log_new.read()) + unit_log.write(log) + + Log.swap(log_new) |