feat(ocr): expose Prometheus /metrics endpoint with OCR-domain counters #653
@@ -605,6 +605,29 @@ async def test_ocr_jobs_total_not_incremented_when_pdf_download_fails_in_stream(
|
|||||||
)._value.get() == 0.0
|
)._value.get() == 0.0
|
||||||
|
|
||||||
|
|
||||||
|
def test_uvicorn_access_log_filter_fails_open_on_short_or_missing_args():
|
||||||
|
"""The filter must default-allow records when args is None or shorter than expected.
|
||||||
|
|
||||||
|
Locks in fail-open behavior: if uvicorn ever changes its format we keep
|
||||||
|
forwarding records to the handler rather than silently dropping logs.
|
||||||
|
"""
|
||||||
|
import logging as _logging
|
||||||
|
from main import MetricsPathFilter
|
||||||
|
|
||||||
|
filt = MetricsPathFilter()
|
||||||
|
none_record = _logging.LogRecord(
|
||||||
|
name="uvicorn.access", level=_logging.INFO, pathname="", lineno=0,
|
||||||
|
msg="some message", args=None, exc_info=None,
|
||||||
|
)
|
||||||
|
short_record = _logging.LogRecord(
|
||||||
|
name="uvicorn.access", level=_logging.INFO, pathname="", lineno=0,
|
||||||
|
msg="%s %s", args=("a", "b"), exc_info=None,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert filt.filter(none_record) is True
|
||||||
|
assert filt.filter(short_record) is True
|
||||||
|
|
||||||
|
|
||||||
def test_uvicorn_access_log_filter_skips_metrics_path():
|
def test_uvicorn_access_log_filter_skips_metrics_path():
|
||||||
"""The MetricsPathFilter drops uvicorn.access log records that target /metrics."""
|
"""The MetricsPathFilter drops uvicorn.access log records that target /metrics."""
|
||||||
import logging as _logging
|
import logging as _logging
|
||||||
|
|||||||
Reference in New Issue
Block a user