81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
"""unit tests: sessions.get_response_json parses a body or logs and returns None
|
|
|
|
run from the repo root:
|
|
uv run pytest tests/test_response_json.py
|
|
|
|
offline — the aioweb Response is faked at the leaf; nothing is sent.
|
|
"""
|
|
|
|
import logging
|
|
|
|
import pytest
|
|
|
|
from monitor_utils.sessions import get_response_json
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, body):
|
|
self.body = body
|
|
|
|
def json(self):
|
|
return self.body
|
|
|
|
|
|
def warnings(caplog):
|
|
return [r for r in caplog.records if r.levelno == logging.WARNING]
|
|
|
|
|
|
def test_dict_body_is_returned_without_logging(caplog):
|
|
with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
|
|
assert get_response_json(FakeResponse({"a": 1})) == {"a": 1}
|
|
assert warnings(caplog) == []
|
|
|
|
|
|
def test_none_body_logs_and_returns_none(caplog):
|
|
with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
|
|
assert get_response_json(FakeResponse(None), category="c1", page=2) is None
|
|
(record,) = warnings(caplog)
|
|
assert record.message == "unexpected response"
|
|
assert record.body_type == "NoneType"
|
|
assert record.body == "None"
|
|
assert (record.category, record.page) == ("c1", 2)
|
|
|
|
|
|
@pytest.mark.parametrize("body", [[1, 2], "text", 42, b"bytes"])
|
|
def test_wrong_shape_logs_type_and_truncated_body(body, caplog):
|
|
with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
|
|
assert get_response_json(FakeResponse(body)) is None
|
|
(record,) = warnings(caplog)
|
|
assert record.body_type == type(body).__name__
|
|
assert record.body == str(body)[:200]
|
|
|
|
|
|
def test_body_preview_is_capped_at_200_chars(caplog):
|
|
with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
|
|
get_response_json(FakeResponse("x" * 1000))
|
|
(record,) = warnings(caplog)
|
|
assert len(record.body) == 200
|
|
|
|
|
|
def test_expected_accepts_other_types(caplog):
|
|
with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
|
|
assert get_response_json(FakeResponse([1]), expected=list) == [1]
|
|
assert get_response_json(FakeResponse([1]), expected=(dict, list)) == [1]
|
|
assert get_response_json(FakeResponse({}), expected=list) is None
|
|
assert len(warnings(caplog)) == 1
|
|
|
|
|
|
def test_warning_goes_to_the_passed_logger(caplog):
|
|
with caplog.at_level(logging.WARNING):
|
|
get_response_json(
|
|
FakeResponse(None), logger=logging.getLogger("woot.woot_monitor")
|
|
)
|
|
(record,) = warnings(caplog)
|
|
assert record.name == "woot.woot_monitor"
|
|
|
|
|
|
def test_empty_dict_counts_as_a_dict(caplog):
|
|
with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
|
|
assert get_response_json(FakeResponse({})) == {}
|
|
assert warnings(caplog) == []
|