chemenu/tools/chemenu/tests/test_cli.py

"""The CLI dispatch wrapper: the budget/trace hook every command runs
through (`cli.main`/`cli._run_traced`), and the broken-pipe handling that
sits alongside it.

Gitea #110's SIGPIPE side finding: a successful call whose output is cut off
by a closed pipe (`wikitool types describe source | head -1`) used to record
`exit_code: 1` in the trace - indistinguishable from a real tool failure, and
reproduced verbatim by the very next, unpiped retry of the same command
showing `exit_code: 0`. `cli._BrokenPipeSwallow` and `cli._pacify_real_fd`
exist to close that gap; these tests exercise them without depending on a
real OS pipe or Click's own internal handling, which is exactly the moving
part being routed around.
"""
import errno
import json
import sys

import pytest

from chemenu import cli


def read_lines(path):
    return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]


class _FailingStream:
    """Raises EPIPE on the `fail_on`-th write - a fake stream standing in for
    a real pipe whose reader has already closed."""

    def __init__(self, fail_on=1):
        self.fail_on = fail_on
        self.calls = 0
        self.written = []
        self.flushed = False

    def write(self, data):
        self.calls += 1
        if self.calls >= self.fail_on:
            raise OSError(errno.EPIPE, "Broken pipe")
        self.written.append(data)
        return len(data)

    def flush(self):
        self.flushed = True

    def isatty(self):
        return False


# --- _BrokenPipeSwallow ---

def test_broken_pipe_swallow_absorbs_epipe_on_write():
    swallow = cli._BrokenPipeSwallow(_FailingStream(fail_on=1))
    n = swallow.write("hello")
    assert n == len("hello")
    assert swallow.sigpipe is True


def test_broken_pipe_swallow_absorbs_epipe_on_flush():
    class _FlushFails:
        def flush(self):
            raise OSError(errno.EPIPE, "Broken pipe")

    swallow = cli._BrokenPipeSwallow(_FlushFails())
    swallow.flush()  # does not raise
    assert swallow.sigpipe is True


def test_broken_pipe_swallow_reraises_unrelated_oserrors():
    class _Explodes:
        def write(self, data):
            raise OSError(errno.ENOSPC, "No space left on device")

    swallow = cli._BrokenPipeSwallow(_Explodes())
    with pytest.raises(OSError):
        swallow.write("x")
    assert swallow.sigpipe is False


def test_broken_pipe_swallow_passes_through_normal_writes():
    wrapped = _FailingStream(fail_on=99)
    swallow = cli._BrokenPipeSwallow(wrapped)
    swallow.write("hello")
    assert wrapped.written == ["hello"]
    assert swallow.sigpipe is False


def test_broken_pipe_swallow_proxies_unknown_attributes():
    wrapped = _FailingStream()
    swallow = cli._BrokenPipeSwallow(wrapped)
    assert swallow.isatty() is False


# --- _pacify_real_fd ---

def test_pacify_real_fd_is_a_best_effort_noop_without_a_real_descriptor():
    class _RaisesOSError:
        def fileno(self):
            raise OSError("not a real fd in this test")

    class _HasNoFilenoAtAll:
        pass

    cli._pacify_real_fd(_RaisesOSError())  # must not raise
    cli._pacify_real_fd(_HasNoFilenoAtAll())  # must not raise either


# --- _run_traced: the trace records what actually happened ---

def test_a_write_cut_off_by_a_closed_pipe_is_not_recorded_as_an_error(monkeypatch, tmp_path):
    monkeypatch.setenv("WIKI_TRACE_DIR", str(tmp_path))
    monkeypatch.setenv("WIKITOOL_SESSION_ID", "sigpipe-unit")
    stub = _FailingStream(fail_on=2)  # first write succeeds, second hits EPIPE
    monkeypatch.setattr(sys, "stdout", stub)

    def fake_app(**_kwargs):
        sys.stdout.write("line one\n")
        sys.stdout.write("line two\n")  # truncated here, like `| head -1`
        raise SystemExit(0)

    monkeypatch.setattr(cli, "app", fake_app)

    with pytest.raises(SystemExit) as exc:
        cli._run_traced("types", ["describe", "source"])
    assert exc.value.code == 0

    records = read_lines(tmp_path / "sigpipe-unit" / "trace.jsonl")
    call = next(r for r in records if r["event"] == "wikitool.call")
    assert call["attrs"]["exit_code"] == 0
    assert call["attrs"]["stdout_truncated"] is True


def test_a_real_failure_is_still_recorded_as_one(monkeypatch, tmp_path):
    """The unrelated-error path stays exactly as before: an actual failure
    keeps its exit code and carries no truncation flag."""
    monkeypatch.setenv("WIKI_TRACE_DIR", str(tmp_path))
    monkeypatch.setenv("WIKITOOL_SESSION_ID", "real-failure-unit")

    def fake_app(**_kwargs):
        raise SystemExit(1)

    monkeypatch.setattr(cli, "app", fake_app)

    with pytest.raises(SystemExit) as exc:
        cli._run_traced("new", ["entity", "--name", ""])
    assert exc.value.code == 1

    records = read_lines(tmp_path / "real-failure-unit" / "trace.jsonl")
    call = next(r for r in records if r["event"] == "wikitool.call")
    assert call["attrs"]["exit_code"] == 1
    assert "stdout_truncated" not in call["attrs"]


def test_an_ordinary_call_restores_the_real_streams_afterwards(monkeypatch, tmp_path):
    monkeypatch.setenv("WIKI_TRACE_DIR", str(tmp_path))
    monkeypatch.setenv("WIKITOOL_SESSION_ID", "restore-unit")
    real_stdout, real_stderr = sys.stdout, sys.stderr

    def fake_app(**_kwargs):
        raise SystemExit(0)

    monkeypatch.setattr(cli, "app", fake_app)
    with pytest.raises(SystemExit):
        cli._run_traced("lint", [])
    assert sys.stdout is real_stdout
    assert sys.stderr is real_stderr


# --- help rendering from cli_contract (Gitea #121 B5) ---

import io  # noqa: E402

from chemenu import cli_contract  # noqa: E402

FRAME_CHARS = set("│╭╰─╯╮┃┏┗━┓┛")
SECTION_ORDER = ("NAME", "SYNOPSIS", "PROPERTIES", "OPTIONS", "EXIT STATUS", "ON FAILURE", "NOTES")


def _run(monkeypatch, args, tty=False):
    """Drive the real app the way `main()` does, capturing both streams.
    `tty=True` makes stdout claim to be a terminal - what a `wikitool` run in
    an interactive shell sees, as opposed to an agent's pipe."""
    out, err = io.StringIO(), io.StringIO()
    if tty:
        out.isatty = lambda: True
    monkeypatch.setattr(sys, "stdout", out)
    monkeypatch.setattr(sys, "stderr", err)
    monkeypatch.setattr(sys, "argv", ["whatever-argv0-is", *args])
    with pytest.raises(SystemExit):
        cli.app(prog_name=cli.PROG_NAME)
    return out.getvalue(), err.getvalue()


def test_publish_help_is_byte_identical_with_and_without_a_tty(monkeypatch):
    piped, _ = _run(monkeypatch, ["publish", "-h"])
    terminal, _ = _run(monkeypatch, ["publish", "-h"], tty=True)
    assert piped == terminal


def test_publish_help_carries_the_record_in_man_page_order(monkeypatch):
    text, _ = _run(monkeypatch, ["publish", "-h"])
    assert text.startswith("NAME")
    positions = [text.index(section) for section in SECTION_ORDER]
    assert positions == sorted(positions)
    exit_block = text[text.index("EXIT STATUS"):text.index("ON FAILURE")]
    assert "\n    1 " in exit_block
    assert "\n    42 " in exit_block


def test_every_command_renders_the_same_for_h_and_help_without_frames(monkeypatch):
    for path in sorted(cli_contract.all_records()):
        short, _ = _run(monkeypatch, [*path.split(" "), "-h"])
        long, _ = _run(monkeypatch, [*path.split(" "), "--help"])
        assert short == long, path
        assert short.startswith("NAME"), path
        assert not FRAME_CHARS & set(short), path


def test_top_level_help_is_the_index_without_frames(monkeypatch):
    text, _ = _run(monkeypatch, ["-h"])
    assert text.startswith("Usage: wikitool")
    assert not FRAME_CHARS & set(text)
    non_idempotent = {
        line.split("  ")[0].strip() for line in text.splitlines() if " non-idempotent " in line
    }
    assert {
        "new", "log append", "publish", "upstream merge",
        "version bump", "version release", "migrate done",
    } <= non_idempotent


def test_a_usage_error_is_unframed_and_names_wikitool(monkeypatch):
    out, err = _run(monkeypatch, ["publish", "--bogus"])
    text = out + err
    assert "Usage: wikitool publish" in text
    assert not FRAME_CHARS & set(text)