Spaces:
Running
Running
File size: 2,135 Bytes
5796881 9841346 5796881 9841346 5796881 21042c7 5796881 9841346 5796881 21042c7 9841346 | 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 | import logging
from utils import log_gathered_exceptions
def test_logs_only_the_exceptions_with_their_query(caplog):
results = ["ok result", ValueError("boom"), "another ok"]
with caplog.at_level(logging.WARNING):
log_gathered_exceptions(results, "test context", ["q1", "q2", "q3"])
warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
assert len(warnings) == 1
assert "q2" in warnings[0].message
assert "boom" in warnings[0].message
assert "test context" in warnings[0].message
def test_logs_nothing_when_all_results_succeeded(caplog):
with caplog.at_level(logging.WARNING):
log_gathered_exceptions(["ok", "also ok"], "test context", ["q1", "q2"])
assert [r for r in caplog.records if r.levelno == logging.WARNING] == []
def test_uses_its_own_logger_not_the_asyncio_one(caplog):
"""Was `from asyncio.log import logger` (flagged in review), which
misattributed these warnings to the 'asyncio' logging namespace instead
of the app's own. Fixed to `logging.getLogger(__name__)`.
"""
with caplog.at_level(logging.WARNING):
log_gathered_exceptions([ValueError("boom")], "ctx", ["q1"])
assert caplog.records[0].name == "utils"
def test_takes_a_plain_sequence_of_queries(caplog):
"""This helper is the lowest layer in the codebase; it used to import
SerpQuery from serp.py purely to read one attribute, inverting the
dependency. It now accepts any iterable of query strings, so the patent
and OPS paths can use it too.
"""
with caplog.at_level(logging.WARNING):
log_gathered_exceptions([ValueError("boom")], "ctx", iter(["only-query"]))
assert "only-query" in caplog.records[0].message
def test_extra_results_without_a_matching_query_are_ignored(caplog):
"""zip stops at the shorter sequence - pinning that so a mismatch can't
raise inside an error-logging path.
"""
with caplog.at_level(logging.WARNING):
log_gathered_exceptions([ValueError("a"), ValueError("b")], "ctx", ["q1"])
assert len([r for r in caplog.records if r.levelno == logging.WARNING]) == 1
|