tests/test_utils.py

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