stack/tests/cms/test_log.py

"""Tests for cms.log — JSONL structured logging."""

from __future__ import annotations

import json
import logging

import pytest

from cms.log import JsonlHandler, setup


@pytest.fixture()
def log_path(tmp_path):
    return tmp_path / "test.jsonl"


class TestJsonlHandler:
    def test_creates_parent_dirs(self, tmp_path) -> None:
        path = tmp_path / "nested" / "dir" / "log.jsonl"
        handler = JsonlHandler(path)
        assert path.parent.exists()
        handler.close()

    def test_writes_json_line(self, log_path) -> None:
        handler = JsonlHandler(log_path)
        logger = logging.getLogger("test.jsonl")
        logger.addHandler(handler)
        logger.setLevel(logging.DEBUG)
        logger.info("hello world")
        handler.close()

        lines = log_path.read_text().strip().split("\n")
        assert len(lines) == 1
        entry = json.loads(lines[0])
        assert entry["message"] == "hello world"
        assert entry["level"] == "INFO"
        assert entry["logger"] == "test.jsonl"
        assert "ts" in entry
        logger.removeHandler(handler)

    def test_multiple_records(self, log_path) -> None:
        handler = JsonlHandler(log_path)
        logger = logging.getLogger("test.multi")
        logger.addHandler(handler)
        logger.setLevel(logging.DEBUG)
        logger.info("first")
        logger.warning("second")
        logger.debug("third")
        handler.close()

        lines = log_path.read_text().strip().split("\n")
        assert len(lines) == 3
        levels = [json.loads(l)["level"] for l in lines]
        assert levels == ["INFO", "WARNING", "DEBUG"]
        logger.removeHandler(handler)

    def test_custom_attrs(self, log_path) -> None:
        handler = JsonlHandler(log_path)
        logger = logging.getLogger("test.custom")
        logger.addHandler(handler)
        logger.setLevel(logging.DEBUG)
        logger.info("with extra", extra={"job_id": "abc123", "count": 42})
        handler.close()

        entry = json.loads(log_path.read_text().strip())
        assert entry["job_id"] == "abc123"
        assert entry["count"] == 42
        logger.removeHandler(handler)

    def test_exception_info(self, log_path) -> None:
        handler = JsonlHandler(log_path)
        logger = logging.getLogger("test.exc")
        logger.addHandler(handler)
        logger.setLevel(logging.DEBUG)
        try:
            raise ValueError("test error")
        except ValueError:
            logger.exception("caught error")
        handler.close()

        entry = json.loads(log_path.read_text().strip())
        assert entry["message"] == "caught error"
        assert "exception" in entry
        assert any("ValueError" in line for line in entry["exception"])
        logger.removeHandler(handler)

    def test_append_mode(self, log_path) -> None:
        log_path.write_text('{"existing": true}\n')
        handler = JsonlHandler(log_path)
        logger = logging.getLogger("test.append")
        logger.addHandler(handler)
        logger.setLevel(logging.DEBUG)
        logger.info("appended")
        handler.close()

        lines = log_path.read_text().strip().split("\n")
        assert len(lines) == 2
        logger.removeHandler(handler)


class TestSetup:
    def test_returns_logger(self, log_path) -> None:
        logger = setup(log_path)
        assert isinstance(logger, logging.Logger)
        assert logger.name == "cms"
        # Cleanup
        for h in logger.handlers[:]:
            if isinstance(h, JsonlHandler):
                h.close()
                logger.removeHandler(h)

    def test_sets_level(self, log_path) -> None:
        logger = setup(log_path, level=logging.WARNING)
        assert logger.level <= logging.WARNING
        for h in logger.handlers[:]:
            if isinstance(h, JsonlHandler):
                h.close()
                logger.removeHandler(h)

    def test_idempotent_same_path(self, log_path) -> None:
        logger1 = setup(log_path)
        count_before = sum(1 for h in logger1.handlers if isinstance(h, JsonlHandler))
        logger2 = setup(log_path)
        count_after = sum(1 for h in logger2.handlers if isinstance(h, JsonlHandler))
        assert count_before == count_after
        assert logger1 is logger2
        for h in logger1.handlers[:]:
            if isinstance(h, JsonlHandler):
                h.close()
                logger1.removeHandler(h)

    def test_different_paths_add_handlers(self, tmp_path) -> None:
        p1 = tmp_path / "a.jsonl"
        p2 = tmp_path / "b.jsonl"
        logger = setup(p1)
        before = sum(1 for h in logger.handlers if isinstance(h, JsonlHandler))
        setup(p2)
        after = sum(1 for h in logger.handlers if isinstance(h, JsonlHandler))
        assert after == before + 1
        for h in logger.handlers[:]:
            if isinstance(h, JsonlHandler):
                h.close()
                logger.removeHandler(h)


class TestJsonlHandlerEdgeCases:
    """Cover remaining edge-case lines in JsonlHandler."""

    def test_non_serializable_extra_uses_repr(self, log_path) -> None:
        """Line 77-78: non-JSON-serializable extras fall back to repr."""
        handler = JsonlHandler(log_path)
        logger = logging.getLogger("test.repr_fallback")
        logger.addHandler(handler)
        logger.setLevel(logging.DEBUG)

        # sets are not JSON-serializable
        logger.info("with set", extra={"data": {1, 2, 3}})
        handler.close()

        entry = json.loads(log_path.read_text().strip())
        assert entry["message"] == "with set"
        # repr(set) produces something like {1, 2, 3}
        assert isinstance(entry["data"], str)
        assert "1" in entry["data"]
        logger.removeHandler(handler)

    def test_extra_key_collides_with_entry(self, log_path) -> None:
        """Line 72-73: extra key already in entry dict is skipped."""
        handler = JsonlHandler(log_path)
        logger = logging.getLogger("test.collision")
        logger.addHandler(handler)
        logger.setLevel(logging.DEBUG)
        # 'level' is already in the entry dict but NOT in _BUILTIN_ATTRS,
        # so it passes the first check (line 70-71) but hits the
        # 'if key in entry: continue' check (line 72-73).
        logger.info("collision", extra={"level": "CUSTOM"})
        handler.close()

        entry = json.loads(log_path.read_text().strip())
        # The entry should keep the original "INFO", not the extra
        assert entry["level"] == "INFO"
        logger.removeHandler(handler)

    def test_emit_handles_error_on_write_failure(self, log_path) -> None:
        """Lines 85-86: handleError is called when emit raises."""
        handler = JsonlHandler(log_path)
        logger = logging.getLogger("test.handle_error")
        logger.addHandler(handler)
        logger.setLevel(logging.DEBUG)

        # Close the file to force a write error
        handler._file.close()
        # This should trigger handleError rather than crashing
        logger.info("this will fail to write")
        logger.removeHandler(handler)

    def test_close_exception_handling(self, tmp_path) -> None:
        """Lines 92-93: close() swallows exceptions."""
        log_file = tmp_path / "close_test.jsonl"
        handler = JsonlHandler(log_file)
        # Close the file first so the handler.close() flush/close
        # will raise, but should be silently swallowed
        handler._file.close()
        # Should not raise
        handler.close()