1import time 2 3import pytest 4from unit.applications.lang.python import TestApplicationPython 5from unit.option import option 6 7 8class TestAccessLog(TestApplicationPython): 9 prerequisites = {'modules': {'python': 'any'}} 10 11 def load(self, script): 12 super().load(script) 13 14 assert 'success' in self.conf( 15 '"' + option.temp_dir + '/access.log"', 'access_log' 16 ), 'access_log configure' 17 18 def wait_for_record(self, pattern, name='access.log'): 19 return super().wait_for_record(pattern, name) 20 21 def test_access_log_keepalive(self): 22 self.load('mirror') 23 24 assert self.get()['status'] == 200, 'init' 25 26 (resp, sock) = self.post( 27 headers={ 28 'Host': 'localhost', 29 'Connection': 'keep-alive', 30 'Content-Type': 'text/html', 31 }, 32 start=True, 33 body='01234', 34 read_timeout=1, 35 ) 36 37 assert ( 38 self.wait_for_record(r'"POST / HTTP/1.1" 200 5') is not None 39 ), 'keepalive 1' 40 41 resp = self.post( 42 headers={ 43 'Host': 'localhost', 44 'Connection': 'close', 45 'Content-Type': 'text/html', 46 }, 47 sock=sock, 48 body='0123456789', 49 ) 50 51 assert ( 52 self.wait_for_record(r'"POST / HTTP/1.1" 200 10') is not None 53 ), 'keepalive 2' 54 55 def test_access_log_pipeline(self): 56 self.load('empty') 57 58 self.http( 59 b"""GET / HTTP/1.1 60Host: localhost 61Referer: Referer-1 62 63GET / HTTP/1.1 64Host: localhost 65Referer: Referer-2 66 67GET / HTTP/1.1 68Host: localhost 69Referer: Referer-3 70Connection: close 71 72""", 73 raw_resp=True, 74 raw=True, 75 ) 76 77 assert ( 78 self.wait_for_record(r'"GET / HTTP/1.1" 200 0 "Referer-1" "-"') 79 is not None 80 ), 'pipeline 1' 81 assert ( 82 self.wait_for_record(r'"GET / HTTP/1.1" 200 0 "Referer-2" "-"') 83 is not None 84 ), 'pipeline 2' 85 assert ( 86 self.wait_for_record(r'"GET / HTTP/1.1" 200 0 "Referer-3" "-"') 87 is not None 88 ), 'pipeline 3' 89 90 def test_access_log_ipv6(self): 91 self.load('empty') 92 93 assert 'success' in self.conf( 94 {"[::1]:7080": {"pass": "applications/empty"}}, 'listeners' 95 ) 96 97 self.get(sock_type='ipv6') 98 99 assert ( 100 self.wait_for_record( 101 r'::1 - - \[.+\] "GET / HTTP/1.1" 200 0 "-" "-"' 102 ) 103 is not None 104 ), 'ipv6' 105 106 def test_access_log_unix(self): 107 self.load('empty') 108 109 addr = option.temp_dir + '/sock' 110 111 assert 'success' in self.conf( 112 {"unix:" + addr: {"pass": "applications/empty"}}, 'listeners' 113 ) 114 115 self.get(sock_type='unix', addr=addr) 116 117 assert ( 118 self.wait_for_record( 119 r'unix: - - \[.+\] "GET / HTTP/1.1" 200 0 "-" "-"' 120 ) 121 is not None 122 ), 'unix' 123 124 def test_access_log_referer(self): 125 self.load('empty') 126 127 self.get( 128 headers={ 129 'Host': 'localhost', 130 'Referer': 'referer-value', 131 'Connection': 'close', 132 } 133 ) 134 135 assert ( 136 self.wait_for_record(r'"GET / HTTP/1.1" 200 0 "referer-value" "-"') 137 is not None 138 ), 'referer' 139 140 def test_access_log_user_agent(self): 141 self.load('empty') 142 143 self.get( 144 headers={ 145 'Host': 'localhost', 146 'User-Agent': 'user-agent-value', 147 'Connection': 'close', 148 } 149 ) 150 151 assert ( 152 self.wait_for_record( 153 r'"GET / HTTP/1.1" 200 0 "-" "user-agent-value"' 154 ) 155 is not None 156 ), 'user agent' 157 158 def test_access_log_http10(self): 159 self.load('empty') 160 161 self.get(http_10=True) 162 163 assert ( 164 self.wait_for_record(r'"GET / HTTP/1.0" 200 0 "-" "-"') is not None 165 ), 'http 1.0' 166 167 def test_access_log_partial(self): 168 self.load('empty') 169 170 assert self.post()['status'] == 200, 'init' 171 172 resp = self.http(b"""GE""", raw=True, read_timeout=1) 173 174 time.sleep(1) 175 176 assert ( 177 self.wait_for_record(r'"GE" 400 0 "-" "-"') is not None 178 ), 'partial' 179 180 def test_access_log_partial_2(self): 181 self.load('empty') 182 183 assert self.post()['status'] == 200, 'init' 184 185 self.http(b"""GET /\n""", raw=True) 186 187 assert ( 188 self.wait_for_record(r'"GET /" 400 \d+ "-" "-"') is not None 189 ), 'partial 2' 190 191 def test_access_log_partial_3(self): 192 self.load('empty') 193 194 assert self.post()['status'] == 200, 'init' 195 196 resp = self.http(b"""GET / HTTP/1.1""", raw=True, read_timeout=1) 197 198 time.sleep(1) 199 200 assert ( 201 self.wait_for_record(r'"GET /" 400 0 "-" "-"') is not None 202 ), 'partial 3' 203 204 def test_access_log_partial_4(self): 205 self.load('empty') 206 207 assert self.post()['status'] == 200, 'init' 208 209 resp = self.http(b"""GET / HTTP/1.1\n""", raw=True, read_timeout=1) 210 211 time.sleep(1) 212 213 assert ( 214 self.wait_for_record(r'"GET / HTTP/1.1" 400 0 "-" "-"') is not None 215 ), 'partial 4' 216 217 @pytest.mark.skip('not yet') 218 def test_access_log_partial_5(self): 219 self.load('empty') 220 221 assert self.post()['status'] == 200, 'init' 222 223 self.get(headers={'Connection': 'close'}) 224 225 assert ( 226 self.wait_for_record(r'"GET / HTTP/1.1" 400 \d+ "-" "-"') 227 is not None 228 ), 'partial 5' 229 230 def test_access_log_get_parameters(self): 231 self.load('empty') 232 233 self.get(url='/?blah&var=val') 234 235 assert ( 236 self.wait_for_record( 237 r'"GET /\?blah&var=val HTTP/1.1" 200 0 "-" "-"' 238 ) 239 is not None 240 ), 'get parameters' 241 242 def test_access_log_delete(self): 243 self.load('empty') 244 245 assert 'success' in self.conf_delete('access_log') 246 247 self.get(url='/delete') 248 249 assert self.search_in_log(r'/delete', 'access.log') is None, 'delete' 250 251 def test_access_log_change(self, temp_dir): 252 self.load('empty') 253 254 self.get() 255 256 assert 'success' in self.conf( 257 '"' + option.temp_dir + '/new.log"', 'access_log' 258 ) 259 260 self.get() 261 262 assert ( 263 self.wait_for_record(r'"GET / HTTP/1.1" 200 0 "-" "-"', 'new.log') 264 is not None 265 ), 'change' 266