"""Bounded fixture processes only; no project servers or external transports."""

from pathlib import Path
from dataclasses import FrozenInstanceError
import sys
import threading
import time
from unittest.mock import patch

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools/devkit"))
from govoplan_devkit.process import _OutputBuffer, require_capture, run_captured


def python(code, **kwargs):
    return run_captured([sys.executable, "-c", code], **kwargs)


def test_large_stdin_and_separate_outputs_are_multiplexed():
    result = python(
        "import sys; sys.stderr.write('diagnostic'); print(len(sys.stdin.buffer.read()))",
        input_bytes=b"x" * 400000,
    )
    assert result.status == "passed"
    assert result.stdout == b"400000\n"
    assert result.stderr == b"diagnostic"


def test_live_output_limit_terminates_unbounded_producer():
    result = python(
        "import os;\nwhile True: os.write(1, b'x'*65536)", max_stdout=2048, timeout=3
    )
    assert result.status == "output_limit"
    assert result.truncated
    assert len(result.stdout) == 2048


def test_drain_mode_caps_memory_without_turning_success_into_failure():
    result = python("print('x'*100000)", max_stdout=1024, terminate_on_limit=False)
    assert result.status == "passed"
    assert result.truncated and len(result.stdout) == 1024


def test_deadline_applies_after_command_closes_both_output_streams():
    started = time.monotonic()
    result = python(
        "import os,time; os.close(1); os.close(2); time.sleep(30)", timeout=0.15
    )
    assert result.status == "timed_out"
    assert time.monotonic() - started < 5


def test_cancellation_stops_owned_process():
    cancelled = threading.Event()
    timer = threading.Timer(0.1, cancelled.set)
    timer.start()
    try:
        result = python("import time; time.sleep(30)", cancelled=cancelled)
    finally:
        timer.cancel()
        timer.join()
    assert result.status == "interrupted"


@pytest.mark.parametrize("redirected", [False, True])
def test_outliving_child_is_not_success_and_cannot_keep_running(redirected):
    code = (
        "import subprocess,sys; child=subprocess.Popen([sys.executable,'-c','import time; time.sleep(30)']"
        + (",stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL" if redirected else "")
        + "); print(child.pid,flush=True)"
    )
    result = python(code, timeout=4)
    assert result.status == "leaked_process"
    child_pid = int(result.stdout.strip())
    # A terminated adopted child may remain a zombie until the host's init reaps it.
    proc = Path(f"/proc/{child_pid}/stat")
    if proc.exists():
        assert proc.read_text().split(")", 1)[1].strip().split()[0] == "Z"


def test_required_capture_rejects_nonordinary_completion():
    with pytest.raises(ValueError, match="output_limit"):
        require_capture([sys.executable, "-c", "print('x'*10000)"], max_stdout=32)


def test_ordinary_nonzero_exit_retains_diagnostics():
    result = python("import sys; print('problem',file=sys.stderr); sys.exit(7)")
    assert (result.status, result.returncode, result.stderr) == (
        "failed",
        7,
        b"problem\n",
    )


def test_selector_setup_failure_still_terminates_spawned_process():
    from govoplan_devkit import process

    original = process.subprocess.Popen
    spawned = []

    def capture(*args, **kwargs):
        child = original(*args, **kwargs)
        spawned.append(child)
        return child

    with (
        patch.object(process.subprocess, "Popen", side_effect=capture),
        patch.object(
            process.selectors,
            "DefaultSelector",
            side_effect=OSError("fixture selector unavailable"),
        ),
    ):
        with pytest.raises(OSError, match="selector unavailable"):
            python("import time; time.sleep(30)")
    assert len(spawned) == 1 and spawned[0].poll() is not None


def test_head_tail_keeps_the_actual_failure_tail_within_original_bound():
    output = b"BEGIN" + b"middle" * 100 + b"FINAL ERROR"
    result = python(
        f"import os; os.write(1, {output!r}); raise SystemExit(7)",
        max_stdout=32,
        capture_mode="head_tail",
        terminate_on_limit=False,
    )
    assert (result.status, result.returncode) == ("failed", 7)
    assert result.stdout == output[:16] + output[-16:]
    assert result.stdout_head_bytes == 16
    assert result.omitted_stdout_bytes == len(output) - 32
    assert result.snapshot().final
    assert "FINAL ERROR" in result.snapshot().text()
    assert str(len(output) - 32) + " output bytes omitted" in result.snapshot().text()


@pytest.mark.parametrize("limit", [0, 1, 2, 3, 31, 1024])
def test_rolling_buffers_remain_bounded_for_every_chunk(limit):
    buffer = _OutputBuffer(limit, "head_tail")
    original = b""
    for data in (b"a", b"bcdef", b"x" * 65536, b"last error"):
        original += data
        buffer.append(data)
        assert len(buffer.head) + len(buffer.tail) <= limit
        assert buffer.omitted == max(0, len(original) - limit)
    if len(original) > limit:
        head = limit // 2
        tail = limit - head
        expected = original[:head] + (original[-tail:] if tail else b"")
        assert buffer.value() == expected


def test_prefix_capture_remains_exact_and_counts_omitted_bytes():
    result = python(
        "import os; os.write(1,b'0123456789')", max_stdout=4, terminate_on_limit=False
    )
    assert result.stdout == b"0123"
    assert result.stdout_head_bytes == 4
    assert result.omitted_stdout_bytes == 6


def test_callback_exposes_bounded_immutable_initial_and_final_snapshots():
    snapshots = []
    result = python(
        "import os,time; os.write(1,b'first\\n'); time.sleep(.15); "
        "os.write(1,b'x'*10000+b'FINAL\\n'); time.sleep(.1)",
        max_stdout=32,
        capture_mode="head_tail",
        terminate_on_limit=False,
        on_output=snapshots.append,
    )
    assert not snapshots[0].final
    assert snapshots[-1] == result.snapshot()
    assert snapshots[0].stdout == b"first\n"
    assert snapshots[-1].stdout.endswith(b"FINAL\n")
    assert all(len(item.stdout) <= 32 for item in snapshots)
    with pytest.raises(FrozenInstanceError):
        snapshots[0].final = True


def test_callback_updates_dirty_output_while_child_becomes_quiet():
    snapshots = []
    python(
        "import os,time; os.write(1,b'first\\n'); time.sleep(.15); "
        "os.write(1,b'second\\n'); time.sleep(1.15)",
        on_output=lambda item: snapshots.append((time.monotonic(), item)),
    )
    assert len(snapshots) >= 3
    assert snapshots[0][1].stdout == b"first\n"
    assert not snapshots[1][1].final
    assert snapshots[1][1].stdout.endswith(b"second\n")
    assert snapshots[1][0] - snapshots[0][0] >= 0.95
    assert snapshots[-1][1].final


def test_quiet_process_has_one_final_empty_snapshot():
    snapshots = []
    python("pass", on_output=snapshots.append)
    assert len(snapshots) == 1
    assert snapshots[0].final and snapshots[0].stdout == b""


def test_snapshot_text_handles_cut_and_invalid_utf8_without_malformed_strings():
    result = python(
        "import os; os.write(1, 'Ä😊Z'.encode()*20+b'\\xff\\xfe')",
        max_stdout=9,
        capture_mode="head_tail",
        terminate_on_limit=False,
    )
    text = result.snapshot().text()
    text.encode("utf-8", errors="strict")
    assert "output bytes omitted" in text
    assert len(result.stdout) == 9
    with pytest.raises(ValueError, match="stream"):
        result.snapshot().text("other")


def test_callback_failure_terminates_owned_process_and_propagates():
    from govoplan_devkit import process

    original = process.subprocess.Popen
    spawned = []

    def capture(*args, **kwargs):
        child = original(*args, **kwargs)
        spawned.append(child)
        return child

    def fail(_snapshot):
        raise ValueError("fixture callback failed")

    with patch.object(process.subprocess, "Popen", side_effect=capture):
        with pytest.raises(ValueError, match="fixture callback failed"):
            python(
                "import time; print('ready',flush=True); time.sleep(30)", on_output=fail
            )
    assert len(spawned) == 1 and spawned[0].poll() is not None


@pytest.mark.parametrize(
    "kwargs", [{"capture_mode": "other"}, {"max_stdout": -1}, {"max_stderr": True}]
)
def test_invalid_capture_configuration_fails_before_starting_a_process(kwargs):
    from govoplan_devkit import process

    with patch.object(process.subprocess, "Popen") as popen:
        with pytest.raises(ValueError):
            python("pass", **kwargs)
        popen.assert_not_called()