1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
|
from __future__ import annotations
import logging
import os
import time
from typing import Optional, Type, Union
import pytest
from hypercorn.config import Config
from hypercorn.logging import AccessLogAtoms, Logger
from hypercorn.typing import HTTPScope, ResponseSummary
@pytest.mark.parametrize(
"target, expected_name, expected_handler_type",
[
("-", "hypercorn.access", logging.StreamHandler),
("/tmp/path", "hypercorn.access", logging.FileHandler),
(logging.getLogger("test_special"), "test_special", None),
(None, None, None),
],
)
def test_access_logger_init(
target: Union[logging.Logger, str, None],
expected_name: Optional[str],
expected_handler_type: Optional[Type[logging.Handler]],
) -> None:
config = Config()
config.accesslog = target
config.access_log_format = "%h"
logger = Logger(config)
assert logger.access_log_format == "%h"
assert logger.getEffectiveLevel() == logging.INFO
if target is None:
assert logger.access_logger is None
elif expected_name is None:
assert logger.access_logger.handlers == []
else:
assert logger.access_logger.name == expected_name
if expected_handler_type is None:
assert logger.access_logger.handlers == []
else:
assert isinstance(logger.access_logger.handlers[0], expected_handler_type)
@pytest.mark.parametrize(
"level, expected",
[
(logging.getLevelName(level_name), level_name)
for level_name in range(logging.DEBUG, logging.CRITICAL + 1, 10)
],
)
def test_loglevel_option(level: Optional[str], expected: int) -> None:
config = Config()
config.loglevel = level
logger = Logger(config)
assert logger.error_logger.getEffectiveLevel() == expected
@pytest.fixture(name="response")
def _response_scope() -> dict:
return {"status": 200, "headers": [(b"Content-Length", b"5"), (b"X-Hypercorn", b"Hypercorn")]}
def test_access_log_standard_atoms(http_scope: HTTPScope, response: ResponseSummary) -> None:
atoms = AccessLogAtoms(http_scope, response, 0.000_023)
assert atoms["h"] == "127.0.0.1:80"
assert atoms["l"] == "-"
assert time.strptime(atoms["t"], "[%d/%b/%Y:%H:%M:%S %z]")
assert int(atoms["s"]) == 200
assert atoms["m"] == "GET"
assert atoms["U"] == "/"
assert atoms["q"] == "a=b"
assert atoms["H"] == "2"
assert int(atoms["b"]) == 5
assert int(atoms["B"]) == 5
assert atoms["f"] == "hypercorn"
assert atoms["a"] == "Hypercorn"
assert atoms["p"] == f"<{os.getpid()}>"
assert atoms["not-atom"] == "-"
assert int(atoms["T"]) == 0
assert int(atoms["D"]) == 23
assert atoms["L"] == "0.000023"
assert atoms["r"] == "GET / 2"
assert atoms["R"] == "GET /?a=b 2"
assert atoms["Uq"] == "/?a=b"
assert atoms["st"] == "OK"
def test_access_log_header_atoms(http_scope: HTTPScope, response: ResponseSummary) -> None:
atoms = AccessLogAtoms(http_scope, response, 0)
assert atoms["{X-Hypercorn}i"] == "Hypercorn"
assert atoms["{X-HYPERCORN}i"] == "Hypercorn"
assert atoms["{not-atom}i"] == "-"
assert atoms["{X-Hypercorn}o"] == "Hypercorn"
assert atoms["{X-HYPERCORN}o"] == "Hypercorn"
def test_access_no_log_header_atoms(http_scope: HTTPScope) -> None:
atoms = AccessLogAtoms(http_scope, {"status": 200, "headers": []}, 0)
assert atoms["{X-Hypercorn}i"] == "Hypercorn"
assert atoms["{X-HYPERCORN}i"] == "Hypercorn"
assert atoms["{not-atom}i"] == "-"
assert not any(key.startswith("{") and key.endswith("}o") for key in atoms.keys())
def test_access_log_environ_atoms(http_scope: HTTPScope, response: ResponseSummary) -> None:
os.environ["Random"] = "Environ"
atoms = AccessLogAtoms(http_scope, response, 0)
assert atoms["{random}e"] == "Environ"
def test_nonstandard_status_code(http_scope: HTTPScope) -> None:
atoms = AccessLogAtoms(http_scope, {"status": 441, "headers": []}, 0)
assert atoms["st"] == "<???441???>"
|
