"""SP18 — JsonFormatter + CycloneDevFormatter tests. Covers the structural shape of log records, exception handling, and the ``extra`` kwarg passthrough. """ from __future__ import annotations import io import json import logging import pytest from cyclone.logging_config import ( CycloneDevFormatter, JsonFormatter, setup_logging, ) def _make_record( msg: str = "hello", args: tuple = (), level: int = logging.INFO, name: str = "test.logger", extras: dict | None = None, exc_info=None, ) -> logging.LogRecord: record = logging.getLogger(name).makeRecord( name=name, level=level, fn="t.py", lno=1, msg=msg, args=args, exc_info=exc_info, ) if extras: for k, v in extras.items(): setattr(record, k, v) return record # --------------------------------------------------------------------------- # JsonFormatter # --------------------------------------------------------------------------- def test_json_formatter_basic_shape(): f = JsonFormatter() line = f.format(_make_record(msg="hello %s", args=("cyclone",))) parsed = json.loads(line) assert parsed["level"] == "INFO" assert parsed["logger"] == "test.logger" assert parsed["msg"] == "hello cyclone" assert "ts" in parsed # ts must be ISO 8601 with milliseconds + Z suffix. assert parsed["ts"].endswith("Z") or "+" in parsed["ts"] def test_json_formatter_includes_extras(): f = JsonFormatter() line = f.format(_make_record( msg="processed", extras={"input_filename": "foo.x12", "parser_kind": "parse_999", "claims": 3}, )) parsed = json.loads(line) assert parsed["extra"] == { "input_filename": "foo.x12", "parser_kind": "parse_999", "claims": 3, } def test_json_formatter_handles_exception_info(): f = JsonFormatter() try: raise ValueError("boom") except ValueError: import sys rec = _make_record(msg="oops", exc_info=sys.exc_info()) line = f.format(rec) parsed = json.loads(line) assert "traceback" in parsed assert "ValueError: boom" in parsed["traceback"] def test_json_formatter_no_extras_key_when_none(): f = JsonFormatter() line = f.format(_make_record(msg="plain")) parsed = json.loads(line) assert "extra" not in parsed def test_json_formatter_handles_non_serializable_extras(): """Non-JSON-serializable extras go through ``default=str``.""" class Opaque: def __str__(self): return "opaque-string" f = JsonFormatter() line = f.format(_make_record(msg="x", extras={"thing": Opaque()})) parsed = json.loads(line) assert parsed["extra"]["thing"] == "opaque-string" def test_json_formatter_preserves_warning_level(): f = JsonFormatter() line = f.format(_make_record(msg="careful", level=logging.WARNING)) parsed = json.loads(line) assert parsed["level"] == "WARNING" # --------------------------------------------------------------------------- # CycloneDevFormatter # --------------------------------------------------------------------------- def test_dev_formatter_basic_shape(): f = CycloneDevFormatter() line = f.format(_make_record(msg="hello %s", args=("cyclone",))) assert "INFO" in line assert "test.logger" in line assert "hello cyclone" in line def test_dev_formatter_includes_extras(): f = CycloneDevFormatter() line = f.format(_make_record( msg="processed", extras={"input_filename": "foo.x12", "claims": 3}, )) assert "input_filename='foo.x12'" in line assert "claims=3" in line def test_dev_formatter_handles_exception(): f = CycloneDevFormatter() try: raise RuntimeError("nope") except RuntimeError: import sys rec = _make_record(msg="oops", exc_info=sys.exc_info()) line = f.format(rec) assert "RuntimeError: nope" in line # --------------------------------------------------------------------------- # setup_logging (light — deeper coverage in test_logging_setup.py) # --------------------------------------------------------------------------- def test_setup_logging_attaches_handler_to_root(): setup_logging(level="DEBUG", json_format=True) root = logging.getLogger() assert len(root.handlers) >= 1 assert isinstance(root.handlers[0].formatter, JsonFormatter) def test_setup_logging_is_idempotent(): """Re-calling clears handlers; the formatter toggle takes effect.""" setup_logging(level="INFO", json_format=True) setup_logging(level="INFO", json_format=False) root = logging.getLogger() assert len(root.handlers) == 1 assert isinstance(root.handlers[0].formatter, CycloneDevFormatter) # Reset back to JSON for the rest of the test suite. setup_logging(level="INFO", json_format=True)