281 lines
8.1 KiB
Python
281 lines
8.1 KiB
Python
from __future__ import annotations
|
|
|
|
import inspect
|
|
import io
|
|
import builtins
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
import dlp_io
|
|
import dlp_io._api 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
|