unit/test/test_access_log.py

import os
import re
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')

        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')

        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)

        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')

        self.stop()

        self.assertIsNotNone(
            self.search_in_log(
                r'::1 - - \[.+\] "GET / HTTP/1.1" 200 0 "-" "-"'), 'ipv6')

    def test_access_log_referer(self):
        self.load('empty')

        self.get(headers={
            'Host': 'localhost',
            'Referer': 'referer-value',
            'Connection': 'close'
        })

        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'
        })

        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)

        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)

        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)

        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)

        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)

        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')

        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')

        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()

        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()

        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')

        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()