AnyPS5/tools/tests/test_pr_overlap_encoding.py

from contextlib import contextmanager
import io
import json
import os
from pathlib import Path
import subprocess
import sys
from tempfile import TemporaryDirectory
import unittest
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import pr_overlap

RUN = subprocess.run
UNICODE = "Quoted \u201cmetadata\u201d, Y\u014dtei and \U0001f3ae"
TREE = "a" * 40


@contextmanager
def simulated_cp1252():
    if hasattr(subprocess, "_text_encoding"):
        with patch.object(subprocess, "_text_encoding", return_value="cp1252"):
            yield
    else:
        wrapper = io.TextIOWrapper

        def fake_wrapper(buffer, encoding=None, *args, **kwargs):
            if encoding == "locale":
                encoding = "cp1252"
            return wrapper(buffer, encoding=encoding, *args, **kwargs)

        with patch.object(io, "TextIOWrapper", new=fake_wrapper):
            yield


@contextmanager
def command_output(stdout, stderr="", returncode=0):
    code = ("import os, sys\n"
            f"os.write(1, {stdout.encode('utf-8')!r})\n"
            f"os.write(2, {stderr.encode('utf-8')!r})\n"
            f"sys.exit({returncode})\n")

    def run(command, **kwargs):
        return RUN([sys.executable, "-c", code], **kwargs)

    with simulated_cp1252(), \
            patch.object(subprocess, "run", side_effect=run):
        yield


class EncodingTests(unittest.TestCase):
    def test_gh_json_preserves_unicode(self):
        response = [{"number": 1, "body": UNICODE}]
        with command_output(json.dumps(response, ensure_ascii=False)):
            self.assertEqual(json.loads(pr_overlap.gh("pr", "list", "--json", "number,body")), response)

    def test_ascii_gh_json(self):
        response = [{"number": 1, "body": "Ordinary metadata"}]
        with command_output(json.dumps(response)):
            self.assertEqual(json.loads(pr_overlap.gh("pr", "list", "--json", "number,body")), response)

    def test_gh_error_preserves_unicode(self):
        with command_output("", UNICODE, 1):
            with self.assertRaises(subprocess.CalledProcessError) as raised:
                pr_overlap.gh("pr", "list")
        self.assertEqual(raised.exception.returncode, 1)
        self.assertEqual(raised.exception.stderr, UNICODE)

    def test_git_diff_preserves_unicode(self):
        diff = "@@ -1 +1 @@\n-old\n+" + UNICODE + "\n"
        with command_output(diff):
            self.assertEqual(pr_overlap.git("diff", "--no-color"), diff)

    def test_git_error_preserves_unicode(self):
        with command_output("", UNICODE, 128):
            with self.assertRaises(subprocess.CalledProcessError) as raised:
                pr_overlap.git("show", "HEAD")
        self.assertEqual(raised.exception.returncode, 128)
        self.assertEqual(raised.exception.stderr, UNICODE)

    def test_merge_conflict_preserves_unicode_paths(self):
        path = UNICODE + ".txt"
        with command_output(TREE + "\n" + path + "\n", returncode=1):
            self.assertEqual(pr_overlap.merge("main", "topic"), (TREE, {path}))

    def test_merge_success(self):
        with command_output(TREE + "\n"):
            self.assertEqual(pr_overlap.merge("main", "topic"), (TREE, set()))

    def test_merge_error_preserves_unicode(self):
        with command_output("", UNICODE, 128):
            with self.assertRaisesRegex(RuntimeError, UNICODE):
                pr_overlap.merge("main", "topic")

    def test_default_encoding_uses_simulated_cp1252(self):
        code = "import os;os.write(1," + repr(UNICODE.encode("utf-8")) + ")"
        with simulated_cp1252():
            default = RUN([sys.executable, "-c", code], capture_output=True, text=True, errors="replace")
            self.assertEqual(default.stdout, UNICODE.encode("utf-8").decode("cp1252", errors="replace"))
            explicit = RUN([sys.executable, "-c", code], capture_output=True, text=True, encoding="utf-8")
            self.assertEqual(explicit.stdout, UNICODE)

    def test_git_diff_survives_invalid_utf8(self):
        with TemporaryDirectory() as directory:
            repo = Path(directory)
            source = repo / "core" / "libs" / "prx" / "x" / "a.cpp"
            source.parent.mkdir(parents=True)
            ident = ["-c", "user.name=t", "-c", "user.email=t@example.test"]
            for args in (["init", "-q"], ["commit", "-q", "--allow-empty", "-m", "base"]):
                subprocess.run(["git", *ident, *args], cwd=repo, check=True, capture_output=True)
            source.write_bytes(b"int APS5_VABI f() {} // caf\xe9\n")
            for args in (["add", "."], ["commit", "-q", "-m", "add"]):
                subprocess.run(["git", *ident, *args], cwd=repo, check=True, capture_output=True)
            old = os.getcwd()
            os.chdir(repo)
            try:
                diff = pr_overlap.git("diff", "-U0", "--no-color", "HEAD~1", "HEAD", "--", "core/libs/prx/*.cpp")
            finally:
                os.chdir(old)
            self.assertEqual(pr_overlap.exports(diff), {"f"})

    def test_report_is_written_as_utf8(self):
        with TemporaryDirectory() as directory:
            path = Path(directory) / "1.md"
            write_text = Path.write_text

            def write_with_utf8(path, text, *args, **kwargs):
                self.assertEqual(kwargs.get("encoding"), "utf-8")
                return write_text(path, text, *args, **kwargs)

            with patch.object(Path, "write_text", new=write_with_utf8), \
                    patch.object(pr_overlap, "report", return_value=UNICODE):
                pr_overlap.write_reports(Path(directory), {1: {}}, "https://example.test")
            self.assertEqual(path.read_bytes(), UNICODE.encode("utf-8"))


if __name__ == "__main__":
    unittest.main()