import os import re import time from subprocess import call import unittest import unit class TestUnitAccessLog(unit.TestUnitApplicationPython): def setUpClass(): unit.TestUnit().check_modules('python') def load(self, script): super().load(script) self.conf('"' + self.testdir + '/access.log"', '/access_log') def search_in_log(self, pattern, name='access.log'): with open(self.testdir + '/' + name, 'r') as f: return re.search(pattern, f.read()) def test_access_log_keepalive(self): self.load('mirror') (resp, sock) = self.post(headers={ 'Connection': 'keep-alive', 'Content-Type': 'text/html', 'Host': 'localhost' }, start=True, body='01234') time.sleep(0.2) self.assertIsNotNone( self.search_in_log(r'"POST / HTTP/1.1" 200 5'), 'keepalive 1') resp = self.post(headers={ 'Connection': 'close', 'Content-Type': 'text/html', 'Host': 'localhost' }, sock=sock, body='0123456789') time.sleep(0.2) self.stop() self.assertIsNotNone( self.search_in_log(r'"POST / HTTP/1.1" 200 10'), 'keepalive 2') def test_access_log_pipeline(self): self.load('empty') self.http(b"""GET / HTTP/1.1 Host: localhost Referer: Referer-1 GET / HTTP/1.1 Host: localhost Referer: Referer-2 GET / HTTP/1.1 Host: localhost Referer: Referer-3 Connection: close """, raw_resp=True, raw=True) time.sleep(0.2) self.stop() self.assertIsNotNone( self.search_in_log(r'"GET / HTTP/1.1" 200 0 "Referer-1" "-"'), 'pipeline 1') self.assertIsNotNone( self.search_in_log(r'"GET / HTTP/1.1" 200 0 "Referer-2" "-"'), 'pipeline 2') self.assertIsNotNone( self.search_in_log(r'"GET / HTTP/1.1" 200 0 "Referer-3" "-"'), 'pipeline 3') def test_access_log_ipv6(self): self.load('empty') self.conf({ "[::1]:7080": { "application": "empty" } }, '/listeners') self.get(sock_type='ipv6') time.sleep(0.2) self.stop() self.assertIsNotNone( self.search_in_log( r'::1 - - \[.+\] "GET / HTTP/1.1" 200 0 "-" "-"'), 'ipv6') def test_access_log_unix(self): self.load('empty') addr = self.testdir + '/sock' self.conf({ "unix:" + addr: { "application": "empty" } }, '/listeners') self.get(sock_type='unix', addr=addr) time.sleep(0.2) self.stop() self.assertIsNotNone(self.search_in_log( r'unix: - - \[.+\] "GET / HTTP/1.1" 200 0 "-" "-"'), 'unix') def test_access_log_referer(self): self.load('empty') self.get(headers={ 'Host': 'localhost', 'Referer': 'referer-value', 'Connection': 'close' }) time.sleep(0.2) self.stop() self.assertIsNotNone( self.search_in_log(r'"GET / HTTP/1.1" 200 0 "referer-value" "-"'), 'referer') def test_access_log_user_agent(self): self.load('empty') self.get(headers={ 'Host': 'localhost', 'User-Agent': 'user-agent-value', 'Connection': 'close' }) time.sleep(0.2) self.stop() self.assertIsNotNone( self.search_in_log( r'"GET / HTTP/1.1" 200 0 "-" "user-agent-value"'), 'user agent') def test_access_log_http10(self): self.load('empty') self.get(http_10=True) time.sleep(0.2) self.stop() self.assertIsNotNone( self.search_in_log( r'"GET / HTTP/1.0" 200 0 "-" "-"'), 'http 1.0') def test_access_log_partial(self): self.load('empty') self.http(b"""GE""", raw_resp=True, raw=True) time.sleep(0.2) self.stop() self.assertIsNotNone( self.search_in_log(r'"GE" 400 0 "-" "-"'), 'partial') def test_access_log_partial_2(self): self.load('empty') self.http(b"""GET /\n""", raw_resp=True, raw=True) time.sleep(0.2) self.stop() self.assertIsNotNone( self.search_in_log(r'"GET /" 400 \d+ "-" "-"'), 'partial 2') def test_access_log_partial_3(self): self.load('empty') self.http(b"""GET / HTTP/1.1""", raw_resp=True, raw=True) time.sleep(0.2) self.stop() self.assertIsNotNone( self.search_in_log(r'"GET /" 400 0 "-" "-"'), 'partial 3') def test_access_log_partial_4(self): self.load('empty') resp = self.http(b"""GET / HTTP/1.1\n""", raw_resp=True, raw=True) time.sleep(0.2) self.stop() self.assertIsNotNone( self.search_in_log(r'"GET / HTTP/1.1" 400 0 "-" "-"'), 'partial 4') def test_access_log_partial_5(self): self.load('empty') self.http(b"""GET / HTTP/1.1\n\n""", raw_resp=True, raw=True) self.stop() self.assertIsNotNone( self.search_in_log(r'"GET / HTTP/1.1" 200 0 "-" "-"'), 'partial 5') def test_access_log_get_parameters(self): self.load('empty') self.get(url='/?blah&var=val') time.sleep(0.2) self.stop() self.assertIsNotNone( self.search_in_log( r'"GET /\?blah&var=val HTTP/1.1" 200 0 "-" "-"'), 'get parameters') def test_access_log_delete(self): self.load('empty') self.conf_delete('/access_log') self.get(url='/delete') time.sleep(0.2) self.stop() self.assertIsNone(self.search_in_log(r'/delete'), 'delete') def test_access_log_change(self): self.load('empty') self.get() self.conf('"' + self.testdir + '/new.log"', '/access_log') self.get() time.sleep(0.2) self.stop() self.assertIsNotNone( self.search_in_log(r'"GET / HTTP/1.1" 200 0 "-" "-"', 'new.log'), 'change') def test_access_log_reopen(self): self.load('empty') log_path = self.testdir + '/access.log' self.assertTrue(self.waitforfiles(log_path), 'open') log_path_new = self.testdir + '/new.log' os.rename(log_path, log_path_new) self.get() time.sleep(0.2) self.assertIsNotNone( self.search_in_log(r'"GET / HTTP/1.1" 200 0 "-" "-"', 'new.log'), 'rename new') self.assertFalse(os.path.isfile(log_path), 'rename old') with open(self.testdir + '/unit.pid', 'r') as f: pid = f.read().rstrip() call(['kill', '-s', 'USR1', pid]) self.assertTrue(self.waitforfiles(log_path), 'reopen') self.get(url='/usr1') time.sleep(0.2) self.assertIsNone( self.search_in_log(r'/usr1', 'new.log'), 'rename new 2') self.assertIsNotNone( self.search_in_log(r'"GET /usr1 HTTP/1.1" 200 0 "-" "-"'), 'reopen 2') if __name__ == '__main__': unittest.main()