opensquilla--opensquilla/tests/test_subprocess_encoding.py

"""Tests for subprocess output decoding and UTF-8 child-env forcing (issue #336).

The Windows fallback path is exercised deterministically on any OS by passing an
explicit ``fallback_encoding`` and by monkeypatching ``os.name`` -- the CI hosts
are POSIX, so we cannot rely on the ambient code page.
"""

from __future__ import annotations

import pytest

from opensquilla import subprocess_encoding
from opensquilla.subprocess_encoding import apply_utf8_child_env, decode_subprocess_output

_HELLO = "你好世界"


class TestDecodeSubprocessOutput:
    def test_gbk_bytes_with_fallback_decode_correctly(self) -> None:
        raw = _HELLO.encode("gbk")
        assert decode_subprocess_output(raw, fallback_encoding="gbk") == _HELLO

    def test_utf8_bytes_win_over_fallback(self) -> None:
        # Strict UTF-8 is tried first, so genuine UTF-8 output is preserved even
        # when a legacy fallback encoding is configured.
        raw = _HELLO.encode("utf-8")
        assert decode_subprocess_output(raw, fallback_encoding="gbk") == _HELLO

    def test_no_fallback_matches_legacy_behavior(self) -> None:
        # POSIX / fallback disabled must behave exactly like the previous code:
        # UTF-8 with replacement.
        raw = _HELLO.encode("gbk")
        assert decode_subprocess_output(raw, fallback_encoding=None) == raw.decode(
            "utf-8", errors="replace"
        )

    def test_valid_utf8_with_no_fallback_is_exact(self) -> None:
        raw = _HELLO.encode("utf-8")
        assert decode_subprocess_output(raw, fallback_encoding=None) == _HELLO

    def test_empty_and_none(self) -> None:
        assert decode_subprocess_output(b"", fallback_encoding="gbk") == ""
        assert decode_subprocess_output(None, fallback_encoding="gbk") == ""

    def test_undecodable_in_both_never_raises(self) -> None:
        # 0xFF is invalid UTF-8 and (as a lone byte) invalid in utf-16 too; the
        # final replacement safety net must keep us from raising.
        result = decode_subprocess_output(b"\xff\xfe\x00bad", fallback_encoding="utf-16")
        assert isinstance(result, str)

    def test_bad_fallback_encoding_name_falls_back_to_replace(self) -> None:
        raw = _HELLO.encode("gbk")
        result = decode_subprocess_output(raw, fallback_encoding="not-a-real-codec")
        assert result == raw.decode("utf-8", errors="replace")

    def test_ascii_is_stable_across_paths(self) -> None:
        raw = b"exit_code=0\nhello\n"
        assert decode_subprocess_output(raw, fallback_encoding="gbk") == raw.decode()
        assert decode_subprocess_output(raw, fallback_encoding=None) == raw.decode()

    def test_chunk_boundary_split_reassembles(self) -> None:
        # Emulate the background_process fix: raw bytes are accumulated and the
        # whole buffer is decoded once, so a multibyte char split across a read
        # boundary is not garbled the way per-chunk decoding was.
        full = _HELLO.encode("gbk")
        first, second = full[:3], full[3:]  # splits a 2-byte GBK char
        per_chunk = first.decode("gbk", errors="replace") + second.decode("gbk", errors="replace")
        buffer = bytearray(first)
        buffer.extend(second)
        whole = decode_subprocess_output(bytes(buffer), fallback_encoding="gbk")
        assert whole == _HELLO
        assert per_chunk != _HELLO  # the old per-chunk approach was lossy

    def test_truncated_gbk_tail_does_not_garble_whole_buffer(self) -> None:
        # Byte-cap truncation cuts the last GBK character in half.  A single cut
        # trailing byte must not collapse the whole buffer to replacement chars.
        body = (_HELLO * 500).encode("gbk")
        truncated = body[:-1]
        result = decode_subprocess_output(truncated, fallback_encoding="gbk")
        assert result.startswith(_HELLO * 10)
        assert "�" not in result  # only the dropped tail is lost, nothing garbled
        assert result == (_HELLO * 500)[:-1]

    def test_truncated_utf8_tail_drops_cleanly(self) -> None:
        # UTF-8 output truncated mid-character (e.g. PYTHONUTF8 child hitting the
        # byte cap) keeps the valid prefix and drops the incomplete tail.
        body = (_HELLO * 500).encode("utf-8")
        truncated = body[:-1]
        result = decode_subprocess_output(truncated, fallback_encoding="gbk")
        assert result == (_HELLO * 500)[:-1]
        assert "�" not in result

    def test_utf8_with_invalid_middle_byte_uses_fallback(self) -> None:
        # A genuinely non-UTF-8 byte in the middle (not just a truncated tail)
        # routes to the legacy code page rather than being treated as UTF-8.
        raw = _HELLO.encode("gbk")
        assert decode_subprocess_output(raw, fallback_encoding="gbk") == _HELLO

    def test_truncated_gbk_with_utf8_lookalike_body_prefers_codepage(self) -> None:
        # 聙 is 0xC2 0x80 in GBK, which is *also* a valid UTF-8 sequence (U+0080).
        # With a truncated trailing lead byte a lenient UTF-8 decode would emit the
        # U+0080 control char and mojibake the real GBK text; scoring by
        # replacement count must keep the code-page reading instead.
        raw = b"\xc2\x80\xe4"  # 聙 + an incomplete trailing multibyte lead
        assert decode_subprocess_output(raw, fallback_encoding="gbk") == "聙"

    def test_truncated_utf8_beats_codepage_on_score(self) -> None:
        # The mirror case: genuinely UTF-8 output cut mid-character must not be
        # dragged into the code page even though the fallback is configured.
        raw = (_HELLO * 50).encode("utf-8")[:-1]
        assert decode_subprocess_output(raw, fallback_encoding="gbk") == (_HELLO * 50)[:-1]

    def test_complete_gbk_valid_as_utf8_still_prefers_codepage(self) -> None:
        # 聙聛 in GBK is 0xC2 0x80 0xC2 0x81 -- a *complete*, fully valid UTF-8
        # string (U+0080 U+0081).  The clean-UTF-8 fast path must not short-circuit
        # past scoring and hand back the C1-control mojibake.
        raw = "聙聛".encode("gbk")
        assert raw == b"\xc2\x80\xc2\x81"
        assert decode_subprocess_output(raw, fallback_encoding="gbk") == "聙聛"

    def test_realistic_output_survives_truncation_both_encodings(self) -> None:
        # Real mixed CJK+ASCII output, in either encoding, truncated by up to a
        # full multibyte char, must decode cleanly and keep the right prefix.
        text = "你好，OpenSquilla 支持中文输出！日志：正常运行 abc123。"
        for encoding in ("utf-8", "gbk"):
            body = (text * 5).encode(encoding)
            for cut in range(5):
                raw = body[: len(body) - cut]
                got = decode_subprocess_output(raw, fallback_encoding="gbk")
                assert "�" not in got, (encoding, cut)
                assert got.startswith("你好，OpenSquilla"), (encoding, cut)


class TestApplyUtf8ChildEnv:
    def test_sets_vars_on_windows(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setattr(subprocess_encoding.os, "name", "nt")
        env: dict[str, str] = {}
        result = apply_utf8_child_env(env)
        assert result is env  # mutates and returns the same dict
        assert env["PYTHONUTF8"] == "1"
        assert env["PYTHONIOENCODING"] == "utf-8"

    def test_does_not_override_existing_values(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setattr(subprocess_encoding.os, "name", "nt")
        env = {"PYTHONIOENCODING": "latin-1", "PYTHONUTF8": "0"}
        apply_utf8_child_env(env)
        assert env["PYTHONIOENCODING"] == "latin-1"
        assert env["PYTHONUTF8"] == "0"

    def test_noop_on_posix(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setattr(subprocess_encoding.os, "name", "posix")
        env: dict[str, str] = {}
        apply_utf8_child_env(env)
        assert env == {}