test_cpython_diff_sync.py

# Owner(s): ["module: dynamo"]

"""Unit tests for tools/cpython_diff_sync offline sync helpers.

See https://github.com/pytorch/pytorch/issues/189607

Lives at test/test_cpython_diff_sync.py (not under test/cpython/) so run_test.py
does not exclude it via the version-prefix filter, and so it is picked up by
normal pull shards rather than the Dynamo-wrapped cpython suite.

CI enforcement for the full tree also runs via the CPYTHON_DIFF_SYNC lintrunner
adapter (tools/linter/adapters/cpython_diff_sync_linter.py).
"""

from __future__ import annotations

import importlib.util
import re
import shutil
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from torch.testing._internal.common_utils import run_tests, TestCase


def _load_module(name: str, relative: str):
    path = Path(__file__).resolve().parents[1] / relative
    spec = importlib.util.spec_from_file_location(name, path)
    if spec is None or spec.loader is None:
        raise AssertionError(f"cannot load {path}")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


diff_sync = _load_module("torch_cpython_diff_sync", "tools/cpython_diff_sync.py")
regen = _load_module(
    "torch_regenerate_cpython_diffs", "tools/regenerate_cpython_diffs.py"
)

_FAKE_REL = "test/cpython/v3_13/test_fake.py"
_FAKE_TAG = "v3.13.5"
_FAKE_UPSTREAM = "Lib/test/test_fake.py"
_OUTSIDE_HUNK_LINE = b'@unittest.skip("Dynamo: not supported yet")\n'
_FAKE_HEADER = b"""\
# ======= BEGIN Dynamo patch =======
# https://raw.githubusercontent.com/python/cpython/refs/tags/v3.13.5/Lib/test/test_fake.py
# ======= END DYNAMO PATCH =======

"""
_FAKE_BODY = b"""\
def foo():
    return 1

# pad01
# pad02
# pad03
# pad04
# pad05
# pad06
# pad07
# pad08

def bar():
    return 2

# pad09
# pad10
# pad11
# pad12
# pad13
# pad14
# pad15
# pad16

def baz():
    return 3
"""


def _fake_pair_bytes():
    """Pristine / adapted with one middle hunk, plus an edit outside that hunk."""
    pristine = _FAKE_BODY
    adapted = _FAKE_HEADER + _FAKE_BODY.replace(b"    return 2", b"    return 99")
    edited = adapted.replace(b"def baz():", _OUTSIDE_HUNK_LINE + b"def baz():")
    diff_text = diff_sync.make_unified_diff(pristine, adapted, _FAKE_REL)
    return pristine, adapted, edited, diff_text


def _bool_pair():
    for diff_path in diff_sync.iter_diff_paths():
        if diff_path.name == "test_bool.diff":
            py_path = diff_path.with_suffix(".py")
            if not py_path.is_file():
                raise AssertionError(f"missing {py_path}")
            return py_path, diff_path
    raise AssertionError("test_bool.py pair not found")


@unittest.skipIf(shutil.which("git") is None, "git binary required")
class CPythonDiffSyncTests(TestCase):
    def test_verify_all_passes_on_checked_in_tree(self):
        errors = diff_sync.verify_all()
        self.assertEqual(
            errors,
            [],
            "CPython .diff sync check failed. Regenerate with:\n"
            "  python tools/regenerate_cpython_diffs.py\n"
            + "\n".join(f"{rel}: {msg}" for rel, msg in errors),
        )

    def test_parse_header_requires_source_url(self):
        with tempfile.TemporaryDirectory() as tmp:
            py = Path(tmp) / "test_fake.py"
            py.write_text("# no url here\n", encoding="utf-8")
            with self.assertRaises(ValueError):
                diff_sync.parse_header(py)

    def test_adapted_blob_drift_fails_verify(self):
        py_path, diff_path = _bool_pair()
        text = diff_path.read_text(encoding="utf-8")
        _before, after = diff_sync.parse_index(text)
        # Corrupt the after-hash in a temp copy of the diff (do not touch the tree).
        corrupted = re.sub(
            rf"(index [0-9a-f]+\.\.){re.escape(after)}",
            rf"\g<1>{'0' * len(after)}",
            text,
            count=1,
        )
        self.assertNotEqual(corrupted, text)
        with tempfile.TemporaryDirectory() as tmp:
            tmp_diff = Path(tmp) / "test_bool.diff"
            tmp_diff.write_bytes(corrupted.encode("utf-8"))
            errors = diff_sync.verify_pair(py_path, tmp_diff)
        self.assertTrue(
            any("stale" in msg or "adapted file hash" in msg for _, msg in errors),
            errors,
        )

    def test_missing_index_line_fails_verify(self):
        py_path, diff_path = _bool_pair()
        text = (
            "\n".join(
                line
                for line in diff_path.read_text(encoding="utf-8").split("\n")
                if not line.startswith("index ")
            )
            + "\n"
        )
        with tempfile.TemporaryDirectory() as tmp:
            tmp_diff = Path(tmp) / "test_bool.diff"
            tmp_diff.write_bytes(text.encode("utf-8"))
            errors = diff_sync.verify_pair(py_path, tmp_diff)
        self.assertTrue(
            any("missing git index line" in msg for _, msg in errors), errors
        )

    def test_all_pairs_share_one_upstream_tag(self):
        # Do not hardcode a CPython tag — the suite will move (e.g. to 3.15).
        tags = {
            diff_sync.parse_header(diff.with_suffix(".py"))[0]
            for diff in diff_sync.iter_diff_paths()
            if diff.with_suffix(".py").is_file()
        }
        self.assertEqual(len(tags), 1, tags)

    def test_index_hashes_match_blob_objects(self):
        py_path, diff_path = _bool_pair()
        before, after = diff_sync.parse_index(diff_path.read_text(encoding="utf-8"))
        adapted = diff_sync.normalize_bytes(py_path.read_bytes())
        self.assertEqual(diff_sync.git_hash_object(adapted), after)
        pristine = diff_sync.reverse_apply_to_pristine(py_path, diff_path)
        self.assertEqual(diff_sync.git_hash_object(pristine), before)

    def _reverse_apply_bytes(self, base: bytes, diff_text: str) -> bytes:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            py = root / _FAKE_REL
            py.parent.mkdir(parents=True)
            py.write_bytes(base)
            diff_path = py.with_suffix(".diff")
            diff_sync.write_utf8(diff_path, diff_text)
            with patch.object(diff_sync, "REPO_ROOT", root):
                return diff_sync.reverse_apply_to_pristine(py, diff_path)

    def test_check_pristine_anchor_refuses_outside_hunk_edit(self):
        _pristine, adapted, edited, diff_text = _fake_pair_bytes()
        old_before, _ = diff_sync.parse_index(diff_text)
        reconstructed = self._reverse_apply_bytes(edited, diff_text)
        self.assertIn(_OUTSIDE_HUNK_LINE, reconstructed)
        self.assertNotEqual(diff_sync.git_hash_object(reconstructed), old_before)
        with self.assertRaisesRegex(
            RuntimeError, "refusing to move the upstream anchor"
        ):
            diff_sync.check_pristine_anchor(
                reconstructed, diff_text, _FAKE_TAG, _FAKE_UPSTREAM
            )
        matching = self._reverse_apply_bytes(adapted, diff_text)
        self.assertEqual(diff_sync.git_hash_object(matching), old_before)
        diff_sync.check_pristine_anchor(matching, diff_text, _FAKE_TAG, _FAKE_UPSTREAM)
        # First full-index write: no index line, nothing to preserve.
        diff_sync.check_pristine_anchor(
            reconstructed, "diff --git a/x b/x\n", _FAKE_TAG, _FAKE_UPSTREAM
        )

    def _run_regen_on_fake_tree(
        self,
        py_bytes: bytes,
        diff_text: str,
        *,
        force: bool,
        pristine_path: Path | None,
    ):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            cpython_dir = root / "test" / "cpython" / "v3_13"
            cpython_dir.mkdir(parents=True)
            py_path = cpython_dir / "test_fake.py"
            diff_path = cpython_dir / "test_fake.diff"
            py_path.write_bytes(py_bytes)
            diff_sync.write_utf8(diff_path, diff_text)
            old_diff = diff_path.read_bytes()
            old_before, _ = diff_sync.parse_index(diff_text)
            with patch.object(regen.diff_sync, "CPYTHON_DIR", cpython_dir):
                with patch.object(regen.diff_sync, "REPO_ROOT", root):
                    with patch.object(regen, "REPO_ROOT", root):
                        rc = regen.regenerate(
                            "test_fake.py", force=force, pristine_path=pristine_path
                        )
            return (
                rc,
                old_before,
                old_diff,
                diff_path.read_bytes(),
                py_path.read_bytes(),
            )

    def test_regenerate_refuses_outside_hunk_edit_without_pristine(self):
        _pristine, _adapted, edited, diff_text = _fake_pair_bytes()
        for force in (False, True):
            with self.subTest(force=force):
                rc, old_before, old_diff, new_diff, _py = self._run_regen_on_fake_tree(
                    edited, diff_text, force=force, pristine_path=None
                )
                self.assertEqual(rc, 1)
                self.assertEqual(new_diff, old_diff)
                got_before, _ = diff_sync.parse_index(new_diff.decode("utf-8"))
                self.assertEqual(got_before, old_before)

    def test_regenerate_force_keeps_anchor_when_in_sync(self):
        _pristine, adapted, _edited, diff_text = _fake_pair_bytes()
        rc, old_before, _old_diff, new_diff, _py = self._run_regen_on_fake_tree(
            adapted, diff_text, force=True, pristine_path=None
        )
        self.assertEqual(rc, 0)
        new_before, new_after = diff_sync.parse_index(new_diff.decode("utf-8"))
        self.assertEqual(new_before, old_before)
        self.assertEqual(diff_sync.git_hash_object(adapted), new_after)

    def test_regenerate_with_pristine_records_outside_hunk_edit(self):
        with tempfile.TemporaryDirectory() as ptmp:
            pristine_file = Path(ptmp) / "pristine.py"
            pristine, _adapted, edited, diff_text = _fake_pair_bytes()
            pristine_file.write_bytes(pristine)
            rc, old_before, old_diff, new_diff, written = self._run_regen_on_fake_tree(
                edited, diff_text, force=False, pristine_path=pristine_file
            )
        self.assertEqual(rc, 0)
        self.assertEqual(written, edited)
        self.assertNotEqual(new_diff, old_diff)
        new_before, new_after = diff_sync.parse_index(new_diff.decode("utf-8"))
        self.assertEqual(new_before, old_before)
        self.assertEqual(diff_sync.git_hash_object(edited), new_after)
        self.assertIn(_OUTSIDE_HUNK_LINE.decode("ascii"), new_diff.decode("utf-8"))


if __name__ == "__main__":
    run_tests()