"""Tests for esphome.espidf.runner."""

from __future__ import annotations

import io
import os
from pathlib import Path
import subprocess
import sys
import threading

import pytest

from esphome.espidf import runner

# A flushing runner delivers the first line in well under a second; this is
# only ever waited out when the shim has gone back to buffering, so keep it
# just long enough to cover interpreter startup on a loaded CI machine.
FIRST_LINE_TIMEOUT = 10.0


def _prepare_main(
    monkeypatch: pytest.MonkeyPatch, probe: Path, *args: str
) -> tuple[io.BytesIO, io.TextIOWrapper]:
    """Point ``runner.main()`` at *probe* with a buffered fake stdout.

    ``main`` rewrites ``sys.path``, ``sys.argv``, both std streams and
    ``os.get_terminal_size``; every one of those is monkeypatched so it is
    put back afterwards. The fake stdout is block buffered like a pipe, so
    the caller can tell whether the shim flushed. The wrapper comes back with
    the buffer because dropping it would close the buffer underneath us.
    """
    buf = io.BytesIO()
    stream = io.TextIOWrapper(buf, encoding="utf-8", newline="\n", line_buffering=False)

    monkeypatch.setattr(sys, "path", list(sys.path))
    monkeypatch.setattr(sys, "argv", ["runner.py", str(probe), *args])
    monkeypatch.setattr(sys, "stdout", stream)
    monkeypatch.setattr(sys, "stderr", stream)
    monkeypatch.setattr(os, "get_terminal_size", os.get_terminal_size)

    return buf, stream


def _run_main(
    monkeypatch: pytest.MonkeyPatch, probe: Path, *args: str
) -> tuple[io.BytesIO, io.TextIOWrapper]:
    """Run ``runner.main()`` against *probe* and expect a clean exit."""
    buf, stream = _prepare_main(monkeypatch, probe, *args)
    assert runner.main() == 0
    return buf, stream


def test_main_filters_noise_and_flushes_each_write(
    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
) -> None:
    """Useful lines reach the stream right away; noisy ones are dropped."""
    buf, _stream = _run_main(
        monkeypatch, fixture_path / "espidf" / "filtering_probe.py"
    )

    # Read before any flush of our own: the shim has to have flushed.
    output = buf.getvalue().decode("utf-8")

    assert "Compiling main.cpp\n" in output
    assert "[2/9] Building C object\n" in output
    # Matched by FILTER_IDF_LINES, so they never leave the runner.
    assert "Project build complete." not in output
    assert "-- Component paths:" not in output
    # Held back until the end because no terminator arrived.
    assert output.endswith("still going\n")


def test_main_keeps_output_after_a_form_feed(
    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
) -> None:
    """A form feed is text, not a line break, so nothing after it is lost."""
    buf, _stream = _run_main(monkeypatch, fixture_path / "espidf" / "formfeed_probe.py")

    assert buf.getvalue().decode("utf-8") == (
        "Compiling main.cpp\npage one\x0cpage two\n[2/9] Building C object\n"
    )


def test_main_drains_a_partial_line_when_the_build_dies(
    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
) -> None:
    """A build that stops mid line must still show that line.

    This is the whole point of draining: the message explaining why the
    build failed is exactly the one most likely to arrive without a
    trailing newline.
    """
    buf, _stream = _prepare_main(
        monkeypatch, fixture_path / "espidf" / "crashing_probe.py"
    )

    with pytest.raises(SystemExit) as excinfo:
        runner.main()

    assert excinfo.value.code == 2
    assert buf.getvalue().decode("utf-8") == "FATAL: ld returned 1 exit status\n"


def test_main_reports_rather_than_raises_when_draining_fails(
    monkeypatch: pytest.MonkeyPatch,
    fixture_path: Path,
    capfd: pytest.CaptureFixture[str],
) -> None:
    """A stream that closed under us must not crash the runner's cleanup.

    The drain runs from a ``finally``, so an exception there would replace
    whatever exit code the build was carrying back.
    """
    _prepare_main(monkeypatch, fixture_path / "espidf" / "closing_probe.py")

    assert runner.main() == 0
    reported = capfd.readouterr().err
    assert "Could not write out remaining output" in reported
    # The held line has to come along; the stream it was meant for is gone.
    assert "partial before close" in reported


def test_main_survives_a_drain_failure_with_nowhere_to_report_it(
    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
) -> None:
    """With no real stderr to report to, cleanup still must not raise.

    ``sys.__stderr__`` is None on some interpreters, and ``print(file=None)``
    falls back to ``sys.stdout``, which here is the shim wrapping the stream
    that just failed.
    """
    monkeypatch.setattr(sys, "__stderr__", None)
    _prepare_main(monkeypatch, fixture_path / "espidf" / "closing_probe.py")

    assert runner.main() == 0


def test_main_still_filters_a_drained_partial_line(
    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
) -> None:
    """Releasing a held line does not smuggle noise past the filter."""
    buf, _stream = _run_main(
        monkeypatch, fixture_path / "espidf" / "partial_noise_probe.py"
    )

    assert buf.getvalue().decode("utf-8") == "Compiling main.cpp\n"


def test_main_keeps_everything_in_verbose_mode(
    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
) -> None:
    """``-v`` turns the filter off so the noisy lines survive."""
    buf, _stream = _run_main(
        monkeypatch, fixture_path / "espidf" / "filtering_probe.py", "-v"
    )

    output = buf.getvalue().decode("utf-8")

    assert "Project build complete.\n" in output
    assert "-- Component paths: /a /b /c\n" in output
    # With no filter there is no line buffering, so the partial line goes
    # straight through as well.
    assert output.endswith("still going")


def test_runner_streams_output_before_the_build_finishes(
    fixture_path: Path, probe_env: dict[str, str]
) -> None:
    """The runner must flush, or a dashboard build looks frozen.

    ``toolchain.py`` spawns the runner as a plain script with no ``-u``, and
    hands it a pipe when esphome itself is running under the dashboard. A
    pipe is block buffered, so without a flush in the shim's ``write()`` the
    output sits in the child until 8 KiB piles up or the build ends.
    """
    runner_py = Path(runner.__file__)
    probe = fixture_path / "espidf" / "streaming_probe.py"

    with subprocess.Popen(
        [sys.executable, str(runner_py), str(probe)],
        stdout=subprocess.PIPE,
        # Keep stderr: if the runner dies on startup, its traceback is the
        # only clue about why no line showed up.
        stderr=subprocess.PIPE,
        env=probe_env,
        text=True,
    ) as proc:
        assert proc.stdout is not None
        assert proc.stderr is not None
        first_line: list[str] = []
        reader = threading.Thread(
            target=lambda: first_line.append(proc.stdout.readline()), daemon=True
        )
        try:
            reader.start()
            reader.join(FIRST_LINE_TIMEOUT)
            still_running = proc.poll() is None

            # The probe sleeps for a minute after writing, so reaching us at
            # all means the line was flushed rather than released at exit.
            assert first_line == ["Compiling main.cpp\n"], (
                f"runner stderr: {'' if still_running else proc.stderr.read()}"
            )
            assert still_running
        finally:
            proc.kill()
            proc.wait()
            # Join before leaving the block, so the reader is done rather than
            # racing ``Popen`` closing the pipe under it.
            reader.join(1.0)


class _FakeKernel32:
    """Stand-in for the Windows kernel32 console code page calls."""

    def __init__(self, input_cp: int, output_cp: int) -> None:
        self.input_cp = input_cp
        self.output_cp = output_cp
        self.calls: list[tuple[str, int]] = []

    def GetConsoleCP(self) -> int:  # noqa: N802
        return self.input_cp

    def GetConsoleOutputCP(self) -> int:  # noqa: N802
        return self.output_cp

    def SetConsoleCP(self, codepage: int) -> int:  # noqa: N802
        self.calls.append(("SetConsoleCP", codepage))
        self.input_cp = codepage
        return 1

    def SetConsoleOutputCP(self, codepage: int) -> int:  # noqa: N802
        self.calls.append(("SetConsoleOutputCP", codepage))
        self.output_cp = codepage
        return 1


def test_main_runs_the_build_with_a_utf8_console(
    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
) -> None:
    """An attached console is switched to UTF-8 and then put back."""
    kernel32 = _FakeKernel32(850, 850)
    monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32)

    _run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py")

    assert kernel32.calls == [
        ("SetConsoleCP", runner.UTF8_CODEPAGE),
        ("SetConsoleOutputCP", runner.UTF8_CODEPAGE),
        ("SetConsoleCP", 850),
        ("SetConsoleOutputCP", 850),
    ]


def test_main_restores_the_console_when_the_build_dies(
    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
) -> None:
    """A failing build must not leave the user's console on UTF-8."""
    kernel32 = _FakeKernel32(437, 437)
    monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32)
    _prepare_main(monkeypatch, fixture_path / "espidf" / "crashing_probe.py")

    with pytest.raises(SystemExit):
        runner.main()

    assert (kernel32.input_cp, kernel32.output_cp) == (437, 437)


def test_main_restores_the_console_when_the_switch_fails_part_way(
    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
) -> None:
    """A failed output page switch must not strand the changed input page."""
    kernel32 = _FakeKernel32(850, 850)

    def _refuse(codepage: int) -> int:
        kernel32.calls.append(("SetConsoleOutputCP", codepage))
        return 0

    kernel32.SetConsoleOutputCP = _refuse  # type: ignore[method-assign]
    monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32)

    _run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py")

    assert kernel32.input_cp == 850
    assert kernel32.calls[-2:] == [("SetConsoleCP", 850), ("SetConsoleOutputCP", 850)]


def test_main_leaves_the_console_alone_when_there_is_none(
    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
) -> None:
    """Without a console the code page calls return 0 and nothing is set."""
    kernel32 = _FakeKernel32(0, 0)
    monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32)

    _run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py")

    assert kernel32.calls == []


def test_main_leaves_a_console_already_on_utf8_alone(
    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
) -> None:
    """An overlapping build must not save UTF-8 as the page to restore."""
    kernel32 = _FakeKernel32(runner.UTF8_CODEPAGE, runner.UTF8_CODEPAGE)
    monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32)

    _run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py")

    assert kernel32.calls == []


@pytest.mark.skipif(sys.platform == "win32", reason="kernel32 exists on Windows")
def test_get_kernel32_is_none_off_windows() -> None:
    assert runner._get_kernel32() is None