"""AC tests for AZ-266: shared structured logging module. Verifies the `log_record_schema` v1.0.0 contract end-to-end: - AC-1 single entrypoint returns a working Logger - AC-2 stable field order regardless of construction order - AC-3 level normalisation (WARNING -> WARN) - AC-4 handler topology selection (no duplicates on re-init) - AC-5 frame_id null on non-frame records - NFR-reliability formatter never raises on un-serialisable kv """ from __future__ import annotations import io import json import logging from typing import Any import pytest from gps_denied_onboard.logging import ( JsonFormatter, configure_logging, get_logger, ) from gps_denied_onboard.logging.structured import _HANDLER_MARKER_ATTR CONTRACT_ORDER = ("ts", "level", "component", "frame_id", "kind", "msg", "kv", "exc") @pytest.fixture def captured_logger() -> tuple[logging.Logger, io.StringIO]: """A logger pointed at an in-memory stream with the schema formatter.""" stream = io.StringIO() handler = logging.StreamHandler(stream) handler.setFormatter(JsonFormatter()) logger = logging.getLogger("test.az266") logger.handlers = [handler] logger.setLevel(logging.DEBUG) logger.propagate = False return logger, stream def _read_last_payload(stream: io.StringIO) -> dict[str, Any]: lines = [line for line in stream.getvalue().splitlines() if line.strip()] return json.loads(lines[-1]) def _read_last_raw(stream: io.StringIO) -> str: return [line for line in stream.getvalue().splitlines() if line.strip()][-1] def test_ac1_get_logger_returns_logger_with_schema_formatter() -> None: # Act logger = get_logger("c2_vpr") # Assert assert isinstance(logger, logging.Logger) root = logging.getLogger() assert any(getattr(h, _HANDLER_MARKER_ATTR, None) for h in root.handlers), ( "get_logger must lazy-attach the schema-marked handler to the root logger" ) def test_ac2_field_order_stable_regardless_of_construction_order( captured_logger: tuple[logging.Logger, io.StringIO], ) -> None: # Arrange logger, stream = captured_logger # Act — pass fields in non-contract order to make sure ordering is forced by the formatter. logger.info( "vpr query", extra={ "kv": {"backbone": "salad"}, "kind": "vpr.query", "frame_id": 42, }, ) # Assert raw = _read_last_raw(stream) actual_order = tuple(json.loads(raw).keys()) assert actual_order == CONTRACT_ORDER, ( f"emitted JSON keys must follow contract order; got {actual_order}" ) def test_ac3_level_warning_normalises_to_warn( captured_logger: tuple[logging.Logger, io.StringIO], ) -> None: # Arrange logger, stream = captured_logger # Act logger.warning("covariance spike", extra={"kind": "state.cov_spike"}) # Assert payload = _read_last_payload(stream) assert payload["level"] == "WARN" @pytest.mark.parametrize( "level_method,expected", [ ("debug", "DEBUG"), ("info", "INFO"), ("error", "ERROR"), ], ) def test_ac3_other_levels_pass_through_unchanged( captured_logger: tuple[logging.Logger, io.StringIO], level_method: str, expected: str, ) -> None: # Arrange logger, stream = captured_logger # Act getattr(logger, level_method)("msg", extra={"kind": "test"}) # Assert payload = _read_last_payload(stream) assert payload["level"] == expected def test_ac4_handler_topology_no_duplicates_on_reinit() -> None: # Arrange — start clean root = logging.getLogger() root.handlers = [h for h in root.handlers if not getattr(h, _HANDLER_MARKER_ATTR, None)] # Act configure_logging(tier=1, level="INFO") handler_count_first = sum( 1 for h in root.handlers if getattr(h, _HANDLER_MARKER_ATTR, None) == "tier1.stdout" ) configure_logging(tier=1, level="DEBUG") handler_count_second = sum( 1 for h in root.handlers if getattr(h, _HANDLER_MARKER_ATTR, None) == "tier1.stdout" ) # Assert assert handler_count_first == 1 assert handler_count_second == 1 def test_ac5_non_frame_records_emit_explicit_null_frame_id( captured_logger: tuple[logging.Logger, io.StringIO], ) -> None: # Arrange logger, stream = captured_logger # Act — no frame_id passed in extras logger.info("startup", extra={"kind": "lifecycle.start", "kv": {"version": "0.1.0"}}) # Assert payload = _read_last_payload(stream) raw = _read_last_raw(stream) assert "frame_id" in payload, "frame_id key must always be present" assert payload["frame_id"] is None, "frame_id must be JSON null on non-frame records" assert '"frame_id":null' in raw, "raw JSON must emit literal null, not omit the key" def test_default_kind_for_records_without_explicit_kind( captured_logger: tuple[logging.Logger, io.StringIO], ) -> None: # Arrange logger, stream = captured_logger # Act logger.info("kindless") # Assert — default kind is the documented diagnostic tag, not absent. payload = _read_last_payload(stream) assert payload["kind"] == "log.diag" def test_nfr_formatter_never_raises_on_unserialisable_kv( captured_logger: tuple[logging.Logger, io.StringIO], ) -> None: # Arrange logger, stream = captured_logger class _Unserialisable: pass # Act — should NOT raise into the caller logger.error( "broken kv", extra={"kind": "log.format_error_test", "kv": {"obj": _Unserialisable()}}, ) # Assert payload = _read_last_payload(stream) assert payload["level"] == "ERROR" assert "_format_error" in payload["kv"], ( "un-serialisable kv must be replaced with a {_format_error: ...} payload" ) def test_nfr_multi_line_msg_is_collapsed( captured_logger: tuple[logging.Logger, io.StringIO], ) -> None: # Arrange logger, stream = captured_logger # Act logger.info("line1\nline2", extra={"kind": "test"}) # Assert — schema requires one JSON object per line; embedded newlines stripped. raw = _read_last_raw(stream) payload = json.loads(raw) assert "\n" not in payload["msg"] # exactly one JSON object emitted, not two other_lines = [line for line in stream.getvalue().splitlines() if line.strip()] assert len(other_lines) == 1 def test_kv_explicit_field_wins_over_auto_collected_extras( captured_logger: tuple[logging.Logger, io.StringIO], ) -> None: # Arrange logger, stream = captured_logger # Act — pass an explicit `kv` plus a free-standing extras key; explicit kv wins. logger.info( "explicit kv", extra={"kind": "test", "kv": {"a": 1}, "stray": "value"}, ) # Assert payload = _read_last_payload(stream) assert payload["kv"] == {"a": 1}