chopratejas--headroom/tests/test_compaction_markdown_kv.py

"""Markdown-KV compaction formatter — opt-in serialization-aware output.

Covers the plumbing added for issue #858:

- ``headroom._core.SmartCrusher.with_compaction_format`` renders uniform
  arrays as Markdown-KV when the lossless gate passes.
- The high-level ``SmartCrusher`` exposes the knob via the
  ``compaction_format`` kwarg and the ``HEADROOM_COMPACTION_FORMAT`` env
  var, defaulting to the unchanged ``csv-schema`` path.
- Unknown format names fail loudly (``ValueError``) instead of silently
  falling back.

The formatter's rendering rules themselves (missing-cell omission,
string quoting, CCR marker contract, buckets) are covered by the Rust
unit tests in ``compaction/formatter.rs``.
"""

from __future__ import annotations

import json

import pytest

from headroom._core import SmartCrusher as RustSmartCrusher
from headroom._core import SmartCrusherConfig as RustSmartCrusherConfig
from headroom.transforms.smart_crusher import SmartCrusher
from headroom.transforms.smart_crusher import SmartCrusherConfig as PySmartCrusherConfig


def _tabular_json(n: int = 50) -> str:
    return json.dumps(
        [
            {
                "id": i,
                "name": f"user_{i}",
                "email": f"user_{i}@example.com",
                "status": "ok" if i % 3 == 0 else "pending",
            }
            for i in range(n)
        ]
    )


# ── Rust bridge: with_compaction_format ──


def test_markdown_kv_renders_kv_lines() -> None:
    # Lower the lossless gate: Markdown-KV repeats field names per row,
    # so its savings vs raw JSON are real but below the 30% default.
    cfg = RustSmartCrusherConfig(lossless_min_savings_ratio=0.01)
    crusher = RustSmartCrusher.with_compaction_format(cfg, "markdown-kv")
    result = crusher.crush(_tabular_json(), "", 1.0)
    assert result.was_modified
    assert "lossless" in result.strategy
    # Columns are schema-sorted, so `email` opens each row and `id`
    # renders as a continuation line.
    assert "- email: user_0@example.com" in result.compressed
    assert "id: 0" in result.compressed
    assert "name: user_1" in result.compressed
    # Declaration line survives (shared with the CSV formatter).
    assert "[50]{" in result.compressed


def test_csv_schema_format_name_matches_default() -> None:
    cfg = RustSmartCrusherConfig(lossless_min_savings_ratio=0.01)
    via_name = RustSmartCrusher.with_compaction_format(cfg, "csv-schema")
    via_default = RustSmartCrusher(cfg)
    content = _tabular_json()
    assert (
        via_name.crush(content, "", 1.0).compressed
        == via_default.crush(content, "", 1.0).compressed
    )


def test_unknown_format_name_raises() -> None:
    with pytest.raises(ValueError, match="markdown-kv"):
        RustSmartCrusher.with_compaction_format(None, "toon")


# ── High-level SmartCrusher knob ──


def test_default_format_is_csv_schema() -> None:
    crusher = SmartCrusher()
    assert crusher._compaction_format == "csv-schema"


def test_kwarg_opts_into_markdown_kv() -> None:
    crusher = SmartCrusher(compaction_format="markdown-kv")
    assert crusher._compaction_format == "markdown-kv"


def test_env_var_opts_in(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("HEADROOM_COMPACTION_FORMAT", "markdown-kv")
    crusher = SmartCrusher()
    assert crusher._compaction_format == "markdown-kv"


def test_kwarg_overrides_env_var(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("HEADROOM_COMPACTION_FORMAT", "json")
    crusher = SmartCrusher(compaction_format="markdown-kv")
    assert crusher._compaction_format == "markdown-kv"


def test_unknown_format_kwarg_raises() -> None:
    with pytest.raises(ValueError):
        SmartCrusher(compaction_format="bogus")


def test_without_compaction_ignores_format() -> None:
    crusher = SmartCrusher(with_compaction=False, compaction_format="markdown-kv")
    assert crusher._compaction_format is None


def test_without_compaction_still_validates_format() -> None:
    # An explicit bogus format is a misconfiguration even when the knob
    # is ignored on this path — fail loudly, don't silently accept.
    with pytest.raises(ValueError, match="bogus"):
        SmartCrusher(with_compaction=False, compaction_format="bogus")


def test_end_to_end_crush_emits_markdown_kv() -> None:
    # Through the high-level Python SmartCrusher, not the Rust bridge:
    # proves the kwarg changes crush() output, not just the stored
    # attribute. Same lowered gate as the bridge test — KV's savings on
    # minified JSON sit below the 30% default.
    config = PySmartCrusherConfig(lossless_min_savings_ratio=0.01)
    crusher = SmartCrusher(config=config, compaction_format="markdown-kv")
    result = crusher.crush(_tabular_json())
    assert result.was_modified
    assert "- email: user_0@example.com" in result.compressed
    assert "[50]{" in result.compressed


def test_default_output_unchanged_by_feature() -> None:
    # The default constructor path must stay byte-identical to an
    # explicit csv-schema opt-in — proves the gate is truly default-off.
    content = _tabular_json()
    default_out = SmartCrusher().crush(content)
    explicit_out = SmartCrusher(compaction_format="csv-schema").crush(content)
    assert default_out.compressed == explicit_out.compressed