from __future__ import annotations

import inspect
import io
import builtins
from pathlib import Path

import pytest

import dlp_io
import dlp_io as api_module


class _FakeSession:
    def __init__(self, payload: bytes = b"bridged data") -> None:
        self.payload = payload
        self.paths: list[str] = []
        self.closed = False

    def open_raw(self, path):
        self.paths.append(str(path))
        return io.BytesIO(self.payload)

    def close(self) -> None:
        self.closed = True


@pytest.fixture(autouse=True)
def reset_default_session():
    dlp_io.shutdown()
    yield
    dlp_io.shutdown()


def test_open_signature_matches_io_open() -> None:
    assert inspect.signature(dlp_io.open) == inspect.signature(io.open)


def test_write_mode_uses_native_open(tmp_path: Path) -> None:
    output = tmp_path / "output.txt"

    with dlp_io.open(output, "w", encoding="utf-8") as file:
        file.write("本地写入")

    assert output.read_bytes() == "本地写入".encode("utf-8")


def test_path_update_mode_is_explicitly_unsupported(tmp_path: Path) -> None:
    path = tmp_path / "data.bin"
    path.write_bytes(b"data")

    with pytest.raises(io.UnsupportedOperation):
        dlp_io.open(path, "r+b")


def test_path_rejects_closefd_false(tmp_path: Path) -> None:
    path = tmp_path / "data.bin"

    with pytest.raises(ValueError):
        dlp_io.open(path, "rb", closefd=False)


def test_custom_opener_uses_native_open(tmp_path: Path) -> None:
    path = tmp_path / "native.txt"
    path.write_text("native", encoding="utf-8")
    calls = []

    with api_module._ORIGINAL_OPEN(path, "rb") as expected:
        expected_bytes = expected.read()

    # Keep the returned fd alive independently; os.open is the normal opener contract.
    import os

    def os_opener(name, flags):
        calls.append((name, flags))
        return os.open(name, flags)

    with dlp_io.open(path, "rb", opener=os_opener) as file:
        assert file.read() == expected_bytes

    assert len(calls) == 1


def test_custom_opener_update_mode_uses_native_open(tmp_path: Path) -> None:
    import os

    path = tmp_path / "native-update.bin"
    path.write_bytes(b"abc")

    with dlp_io.open(path, "r+b", opener=os.open) as file:
        assert file.read(1) == b"a"
        file.seek(0)
        file.write(b"z")

    assert path.read_bytes() == b"zbc"


def test_integer_fd_uses_native_open(tmp_path: Path) -> None:
    import os

    path = tmp_path / "fd.bin"
    path.write_bytes(b"fd data")
    descriptor = os.open(path, os.O_RDONLY)

    with dlp_io.open(descriptor, "rb", closefd=True) as file:
        assert file.read() == b"fd data"


def test_read_lazily_starts_and_reuses_default_session(monkeypatch) -> None:
    created: list[tuple[object, float, _FakeSession]] = []

    def fake_start(*, python_cmd=None, timeout=60):
        session = _FakeSession()
        created.append((python_cmd, timeout, session))
        return session

    monkeypatch.setattr(api_module.DlpSession, "start", fake_start)
    dlp_io.configure(python_executable=r"C:\Approved\python.exe", startup_timeout=12)

    with dlp_io.open(r"D:\data\first.bin", "rb") as first:
        assert first.read() == b"bridged data"
    with dlp_io.open(r"D:\data\second.bin", "rb") as second:
        assert second.read() == b"bridged data"

    assert len(created) == 1
    assert tuple(Path(item) for item in created[0][0]) == (
        Path(r"C:\Approved\python.exe"),
    )
    assert created[0][1] == 12
    assert tuple(Path(item) for item in created[0][2].paths) == (
        Path(r"D:\data\first.bin"),
        Path(r"D:\data\second.bin"),
    )


def test_configure_rejects_active_session_and_allows_after_shutdown(monkeypatch) -> None:
    session = _FakeSession()
    monkeypatch.setattr(api_module.DlpSession, "start", lambda **kwargs: session)

    with dlp_io.open(r"D:\data\input.bin", "rb") as file:
        assert file.read() == b"bridged data"

    with pytest.raises(dlp_io.DlpConfigurationError):
        dlp_io.configure(python_executable=r"C:\Other\python.exe")

    dlp_io.shutdown()
    assert session.closed

    dlp_io.configure(python_executable=r"C:\Other\python.exe")


def test_binary_buffering_zero_returns_raw_stream(monkeypatch) -> None:
    session = _FakeSession(b"raw")
    monkeypatch.setattr(api_module.DlpSession, "start", lambda **kwargs: session)

    with dlp_io.open(r"D:\data\raw.bin", "rb", buffering=0) as file:
        assert isinstance(file, io.BytesIO)
        assert file.read() == b"raw"


def test_binary_default_returns_buffered_reader(monkeypatch) -> None:
    session = _FakeSession(b"buffered")
    monkeypatch.setattr(api_module.DlpSession, "start", lambda **kwargs: session)

    with dlp_io.open(r"D:\data\buffered.bin", "rb") as file:
        assert isinstance(file, io.BufferedReader)
        assert file.read() == b"buffered"


def test_text_mode_preserves_encoding_errors_and_newline(monkeypatch) -> None:
    session = _FakeSession("第一行\r\n第二行".encode("utf-8"))
    monkeypatch.setattr(api_module.DlpSession, "start", lambda **kwargs: session)

    with dlp_io.open(
        r"D:\data\text.txt",
        "r",
        encoding="utf-8",
        errors="strict",
        newline="",
    ) as file:
        assert file.read().encode("utf-8") == session.payload


@pytest.mark.parametrize(
    "kwargs",
    [
        {"encoding": "utf-8"},
        {"errors": "strict"},
        {"newline": ""},
    ],
)
def test_binary_mode_rejects_text_arguments_before_starting_helper(
    monkeypatch, kwargs
) -> None:
    monkeypatch.setattr(
        api_module.DlpSession,
        "start",
        lambda **unused: pytest.fail("invalid arguments must not start the helper"),
    )

    with pytest.raises(ValueError):
        dlp_io.open(r"D:\data\payload.bin", "rb", **kwargs)


@pytest.mark.parametrize("mode", ["", "rr", "rbt", "U", "q"])
def test_invalid_read_mode_is_rejected_before_starting_helper(
    monkeypatch, mode: str
) -> None:
    monkeypatch.setattr(
        api_module.DlpSession,
        "start",
        lambda **unused: pytest.fail("invalid mode must not start the helper"),
    )

    with pytest.raises(ValueError):
        dlp_io.open(r"D:\data\payload.bin", mode)


@pytest.mark.parametrize("buffering", [None, "invalid"])
def test_invalid_buffering_is_rejected_before_starting_helper(
    monkeypatch, buffering
) -> None:
    monkeypatch.setattr(
        api_module.DlpSession,
        "start",
        lambda **unused: pytest.fail("invalid buffering must not start the helper"),
    )

    with pytest.raises(TypeError):
        dlp_io.open(r"D:\data\payload.bin", "rb", buffering=buffering)


def test_negative_buffering_is_rejected_before_starting_helper(monkeypatch) -> None:
    monkeypatch.setattr(
        api_module.DlpSession,
        "start",
        lambda **unused: pytest.fail("invalid buffering must not start the helper"),
    )

    with pytest.raises(ValueError):
        dlp_io.open(r"D:\data\payload.bin", "rb", buffering=-2)


def test_invalid_pathlike_is_rejected_before_starting_helper(monkeypatch) -> None:
    class InvalidPath:
        def __fspath__(self):
            raise TypeError("invalid path")

    monkeypatch.setattr(
        api_module.DlpSession,
        "start",
        lambda **unused: pytest.fail("invalid path must not start the helper"),
    )

    with pytest.raises(TypeError):
        dlp_io.open(InvalidPath(), "rb")


def test_open_patch_routes_reads_and_keeps_native_writes(tmp_path, monkeypatch) -> None:
    session = _FakeSession(b"patched")
    monkeypatch.setattr(api_module.DlpSession, "start", lambda **kwargs: session)
    original_builtin = builtins.open
    original_io = io.open

    dlp_io.install_open_patch()
    dlp_io.install_open_patch()
    try:
        with builtins.open(r"D:\data\patched.bin", "rb") as file:
            assert file.read() == b"patched"
        output = tmp_path / "native.txt"
        with builtins.open(output, "w", encoding="utf-8") as file:
            file.write("native")
        with original_builtin(output, "rb") as file:
            assert file.read() == b"native"
    finally:
        dlp_io.uninstall_open_patch()
        dlp_io.uninstall_open_patch()

    assert builtins.open is original_builtin
    assert io.open is original_io