RustPython/scripts/update_lib/tests/test_migrate.py

"""Tests for migrate.py - file migration operations."""

import contextlib
import io
import json
import pathlib
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from update_lib.cmd_migrate import (
    _preserve_block,
    main,
    migrate_package,
    patch_directory,
    patch_file,
    patch_single_content,
    retire_legacy_paths,
)
from update_lib.patch_spec import (
    CANONICAL_LOAD_TESTS,
    COMMENT,
    PatchSpec,
    UtMethod,
    extract_patches,
    iter_patches,
    patches_to_json,
    select_remaining_patches,
)


class TestRetireLegacyPaths(unittest.TestCase):
    @contextlib.contextmanager
    def trees(self):
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            source = root / "cpython" / "Lib" / "test"
            source.mkdir(parents=True)
            local = root / "rustpython"
            (local / "Lib" / "test").mkdir(parents=True)
            with contextlib.chdir(local):
                yield source

    def test_flat_map_retires_only_explicit_paths_without_parsing(self):
        with self.trees() as source:
            mapping = {}
            for name, content in (
                ("bad.py", b"invalid syntax !"),
                ("data.bin", b"\xff\x00"),
            ):
                (source / "new").mkdir(exist_ok=True)
                (source / "new" / name).write_bytes(content)
                Path("Lib/test/new").mkdir(exist_ok=True)
                Path("Lib/test/new", name).write_bytes(content)
                Path("Lib/test", name).write_bytes(content)
                mapping[f"Lib/test/{name}"] = f"Lib/test/new/{name}"
            Path("Lib/test/removed.py").write_text("old = True\n")
            mapping["Lib/test/removed.py"] = None
            Path("Lib/test/original.py").write_text("preserve = True\n")
            Path("map.json").write_text(json.dumps(mapping))

            self.assertEqual(main([str(source), "--legacy-map", "map.json"]), 0)

            for old in mapping:
                self.assertFalse(Path(old).exists())
            self.assertEqual(
                Path("Lib/test/original.py").read_text(), "preserve = True\n"
            )
            self.assertEqual(Path("Lib/test/new/data.bin").read_bytes(), b"\xff\x00")
            self.assertEqual(
                Path("Lib/test/new/bad.py").read_bytes(), b"invalid syntax !"
            )

    def test_late_missing_or_adapted_replacement_preserves_all_legacy_files(self):
        for replacement in (None, b"local adaptation"):
            with self.subTest(replacement=replacement), self.trees() as source:
                Path("Lib/test/first.py").write_text("first")
                Path("Lib/test/second.py").write_text("second")
                (source / "new.py").write_bytes(b"canonical")
                if replacement is not None:
                    Path("Lib/test/new.py").write_bytes(replacement)
                with self.assertRaisesRegex(ValueError, "Replacement is not canonical"):
                    retire_legacy_paths(
                        source,
                        {
                            "Lib/test/first.py": None,
                            "Lib/test/second.py": "Lib/test/new.py",
                        },
                    )
                self.assertTrue(Path("Lib/test/first.py").exists())
                self.assertTrue(Path("Lib/test/second.py").exists())

    def test_file_still_upstream_is_not_retired(self):
        with self.trees() as source:
            (source / "active.py").write_text("active")
            Path("Lib/test/active.py").write_text("active")
            with self.assertRaisesRegex(ValueError, "upstream-absent"):
                retire_legacy_paths(source, {"Lib/test/active.py": None})
            self.assertTrue(Path("Lib/test/active.py").exists())

    def test_outside_paths_and_missing_source_are_rejected(self):
        with self.trees() as source:
            Path("Lib/test/removed.py").write_text("preserve")
            for name in ("outside.py", "Lib/../outside.py", "/tmp/outside.py"):
                with self.subTest(name=name), self.assertRaises(ValueError):
                    retire_legacy_paths(source, {name: None})
            for missing in (
                source / "missing",
                source.parent.parent / "missing/Lib/test",
            ):
                with (
                    self.subTest(missing=missing),
                    self.assertRaisesRegex(ValueError, "Canonical Lib"),
                ):
                    retire_legacy_paths(missing, {"Lib/test/removed.py": None})
                self.assertEqual(Path("Lib/test/removed.py").read_text(), "preserve")


class TestPatchSingleContent(unittest.TestCase):
    """Tests for patch_single_content function."""

    def test_explicit_blocks_preserve_existing_changes(self):
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            source, target = root / "canonical.py", root / "Lib/test/test_example.py"
            target.parent.mkdir(parents=True)
            source.write_text("import unittest\nvalue = 2\n")
            target.write_text("import sys\nimport unittest\nvalue = 1\n")
            document = {
                "target": "Lib/test/test_example.py",
                "patches": {},
                "blocks": [["import unittest\n", "import sys\nimport unittest\n"]],
            }
            with patch(
                "update_lib.cmd_migrate.__file__",
                str(root / "scripts/update_lib/cmd_migrate.py"),
            ):
                self.assertEqual(
                    patch_single_content(source, target), source.read_text()
                )
                result = patch_single_content(source, target, document)
                self.assertEqual(result, "import sys\nimport unittest\nvalue = 2\n")
                for blocks in (
                    None,
                    [[]],
                    [["import unittest\n", "import missing\n"]],
                    [["value = 2\n", "import sys\n"], ["import sys\n", "value = 1\n"]],
                ):
                    with self.subTest(blocks=blocks), self.assertRaises(ValueError):
                        patch_file(
                            source,
                            target,
                            verbose=False,
                            patches_document={**document, "blocks": blocks},
                        )
                    self.assertEqual(
                        target.read_text(), "import sys\nimport unittest\nvalue = 1\n"
                    )

    def test_nested_fragments_and_ambiguous_inputs(self):
        canonical, local = "    value = 1\n", "    value = 2\n"
        self.assertEqual(
            _preserve_block(
                "class X:\n" + local, "class X:\n" + canonical, canonical, local
            ),
            "class X:\n" + local,
        )
        for old, new in (
            (local * 2, canonical),
            (local, canonical * 2),
            ("", canonical),
        ):
            with self.subTest(old=old, new=new), self.assertRaises(ValueError):
                _preserve_block(old, new, canonical, local)

    def test_invalid_result_does_not_write_production_file(self):
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            source, target = root / "canonical.py", root / "Lib/example.py"
            target.parent.mkdir()
            source.write_text("x = 1\n")
            target.write_text("x = 'a'\n")
            document = {
                "target": "Lib/example.py",
                "patches": {},
                "blocks": [["1", "'a"]],
            }
            with patch(
                "update_lib.cmd_migrate.__file__",
                str(root / "scripts/update_lib/cmd_migrate.py"),
            ):
                with self.assertRaisesRegex(ValueError, "invalid Python"):
                    patch_file(source, target, verbose=False, patches_document=document)
                self.assertEqual(target.read_text(), "x = 'a'\n")

    def migrate_doctest(self, selection=None, explicit=False):
        old = '''"""
>>> raise X, Y  # TODO: RUSTPYTHON; old mismatch # doctest: +EXPECTED_FAILURE
SyntaxError: old result
>>> raise Z  # TODO: RUSTPYTHON; obsolete mismatch # doctest: +EXPECTED_FAILURE
SyntaxError: obsolete result
"""
'''
        canonical = (
            '''"""
>>> raise X, Y
SyntaxError: canonical result
>>> raise Z
SyntaxError: canonical second result
"""
'''
            + CANONICAL_LOAD_TESTS
            + "\n"
        )
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary) / "repository"
            target = root / "Lib/test/test_example.py"
            target.parent.mkdir(parents=True)
            target.write_text(old)
            source = Path(temporary) / "canonical/Lib/test/test_example.py"
            source.parent.mkdir(parents=True)
            source.write_text(canonical)
            args = [str(source)]
            if explicit:
                metadata = Path(temporary) / "selection.json"
                metadata.write_text(
                    json.dumps(
                        {
                            "target": "Lib/test/test_example.py",
                            "patches": {},
                            "doctests": selection,
                        }
                    )
                )
                args += ["--patches", str(metadata)]
            with (
                contextlib.chdir(root),
                patch(
                    "update_lib.cmd_migrate.__file__",
                    str(root / "scripts/update_lib/cmd_migrate.py"),
                ),
                contextlib.redirect_stdout(io.StringIO()),
                contextlib.redirect_stderr(io.StringIO()),
            ):
                result = main(args)
            return result, target.read_text(), old, canonical

    def test_doctest_default_requires_explicit_selection(self):
        status, result, old, _ = self.migrate_doctest()
        self.assertEqual(status, 1)
        self.assertEqual(result, old)

    def test_doctest_cli_selection_keeps_only_reviewed_guard(self):
        status, result, _, _ = self.migrate_doctest(
            {"raise X, Y\n": "known mismatch"}, True
        )
        self.assertEqual(status, 0)
        self.assertEqual(result.count("+EXPECTED_FAILURE"), 1)
        self.assertIn("SyntaxError: canonical result", result)
        self.assertIn("SyntaxError: canonical second result", result)
        self.assertNotIn("obsolete mismatch", result)

    def test_doctest_empty_selection_keeps_canonical_unchanged(self):
        status, result, _, canonical = self.migrate_doctest({}, True)
        self.assertEqual(status, 0)
        self.assertEqual(result, canonical)

    def test_doctest_invalid_selection_writes_nothing(self):
        for selection in (
            None,
            [],
            {"raise unknown\n": "unrecognized"},
            {"raise X, Y\n": "line\nbreak"},
        ):
            with self.subTest(selection=selection):
                status, result, old, _ = self.migrate_doctest(selection, True)
                self.assertEqual(status, 1)
                self.assertEqual(result, old)

    def test_patch_with_no_existing_file(self):
        """Test patching when lib file doesn't exist."""
        with tempfile.TemporaryDirectory() as tmpdir:
            tmpdir = pathlib.Path(tmpdir)

            # Create source file
            src_path = tmpdir / "src.py"
            src_path.write_text("""import unittest

class TestFoo(unittest.TestCase):
    def test_one(self):
        pass
""")

            # Non-existent lib path
            lib_path = tmpdir / "lib.py"

            result = patch_single_content(src_path, lib_path)

            # Should return source content unchanged
            self.assertIn("def test_one(self):", result)
            self.assertNotIn(COMMENT, result)

    def test_patch_with_existing_patches(self):
        """Test patching preserves existing patches."""
        with tempfile.TemporaryDirectory() as tmpdir:
            tmpdir = pathlib.Path(tmpdir)

            # Create source file (new version)
            src_path = tmpdir / "src.py"
            src_path.write_text("""import unittest

class TestFoo(unittest.TestCase):
    def test_one(self):
        pass

    def test_two(self):
        pass
""")

            # Create lib file with existing patch
            lib_path = tmpdir / "lib.py"
            lib_path.write_text(f"""import unittest

class TestFoo(unittest.TestCase):
    # {COMMENT}
    @unittest.expectedFailure
    def test_one(self):
        pass
""")

            result = patch_single_content(src_path, lib_path)

            # Should have patch on test_one
            self.assertIn("@unittest.expectedFailure", result)
            self.assertIn(COMMENT, result)
            # Should have test_two from source
            self.assertIn("def test_two(self):", result)


class TestPatchFile(unittest.TestCase):
    """Tests for patch_file function."""

    def test_patch_file_creates_output(self):
        """Test that patch_file writes output file."""
        with tempfile.TemporaryDirectory() as tmpdir:
            tmpdir = pathlib.Path(tmpdir)

            # Create source file
            src_path = tmpdir / "src.py"
            src_path.write_text("""import unittest

class TestFoo(unittest.TestCase):
    def test_one(self):
        pass
""")

            # Output path
            lib_path = tmpdir / "Lib" / "test.py"

            patch_file(src_path, lib_path, verbose=False)

            # File should exist
            self.assertTrue(lib_path.exists())
            content = lib_path.read_text()
            self.assertIn("def test_one(self):", content)

    def test_patch_file_preserves_patches(self):
        """Test that patch_file preserves existing patches."""
        with tempfile.TemporaryDirectory() as tmpdir:
            tmpdir = pathlib.Path(tmpdir)

            # Create source file
            src_path = tmpdir / "src.py"
            src_path.write_text("""import unittest

class TestFoo(unittest.TestCase):
    def test_one(self):
        pass
""")

            # Create existing lib file with patch
            lib_path = tmpdir / "lib.py"
            lib_path.write_text(f"""import unittest

class TestFoo(unittest.TestCase):
    # {COMMENT}
    @unittest.expectedFailure
    def test_one(self):
        pass
""")

            patch_file(src_path, lib_path, verbose=False)

            content = lib_path.read_text()
            self.assertIn("@unittest.expectedFailure", content)


class TestPatchDirectory(unittest.TestCase):
    """Tests for patch_directory function."""

    def test_empty_canonical_fixture_stays_byte_empty(self):
        with tempfile.TemporaryDirectory() as temporary:
            source, target = Path(temporary) / "canonical", Path(temporary) / "local"
            source.mkdir()
            target.mkdir()
            (source / "empty.py").write_bytes(b"")
            (target / "empty.py").write_bytes(b"\n")
            for _ in range(2):
                patch_directory(source, target, verbose=False)
                self.assertEqual((target / "empty.py").read_bytes(), b"")

    def test_excludes_cache_directories_but_preserves_canonical_assets(self):
        with tempfile.TemporaryDirectory() as temporary:
            root = pathlib.Path(temporary)
            source = root / "source"
            destination = root / "destination"
            for relative in ("__pycache__", "nested/__pycache__", "assets"):
                (source / relative).mkdir(parents=True)
            files = {
                "module.py": b"# canonical source\n",
                "__pycache__/module.cpython-315.pyc": b"stale cache",
                "nested/__pycache__/module.cpython-315.pyc": b"nested cache",
                "assets/__pycache__": b"ordinary canonical file",
                "assets/fixture.pyc": b"canonical bytecode fixture",
            }
            for relative, content in files.items():
                (source / relative).write_bytes(content)
            patch_directory(source, destination, verbose=False)
            expected = {
                relative: content
                for relative, content in files.items()
                if "__pycache__" not in pathlib.Path(relative).parts[:-1]
            }
            actual = {
                path.relative_to(destination).as_posix(): path.read_bytes()
                for path in destination.rglob("*")
                if path.is_file()
            }
            self.assertEqual(actual, expected)
            for relative, content in files.items():
                self.assertEqual((source / relative).read_bytes(), content)

    def test_patch_directory_all_files(self):
        """Test that patch_directory processes all .py files."""
        with tempfile.TemporaryDirectory() as tmpdir:
            tmpdir = pathlib.Path(tmpdir)

            # Create source directory with files
            src_dir = tmpdir / "src"
            src_dir.mkdir()
            (src_dir / "test_a.py").write_text("# test_a")
            (src_dir / "test_b.py").write_text("# test_b")
            (src_dir / "subdir").mkdir()
            (src_dir / "subdir" / "test_c.py").write_text("# test_c")

            # Output directory
            lib_dir = tmpdir / "lib"

            patch_directory(src_dir, lib_dir, verbose=False)

            # All files should exist
            self.assertTrue((lib_dir / "test_a.py").exists())
            self.assertTrue((lib_dir / "test_b.py").exists())
            self.assertTrue((lib_dir / "subdir" / "test_c.py").exists())

    def test_patch_directory_preserves_patches(self):
        """Test that patch_directory preserves patches in existing files."""
        with tempfile.TemporaryDirectory() as tmpdir:
            tmpdir = pathlib.Path(tmpdir)

            # Create source directory
            src_dir = tmpdir / "src"
            src_dir.mkdir()
            (src_dir / "test_a.py").write_text("""import unittest

class TestA(unittest.TestCase):
    def test_one(self):
        pass
""")

            # Create lib directory with patched file
            lib_dir = tmpdir / "lib"
            lib_dir.mkdir()
            (lib_dir / "test_a.py").write_text(f"""import unittest

class TestA(unittest.TestCase):
    # {COMMENT}
    @unittest.expectedFailure
    def test_one(self):
        pass
""")

            patch_directory(src_dir, lib_dir, verbose=False)

            content = (lib_dir / "test_a.py").read_text()
            self.assertIn("@unittest.expectedFailure", content)


OVERRIDE_CLEAN = """import unittest
class Example(unittest.TestCase):
    def test_known(self):
        self.assertEqual(1, 1)
    def test_guarded(self):
        self.assertEqual(2, 2)
"""
OVERRIDE_OLD = OVERRIDE_CLEAN.replace(
    "    def test_known",
    "    @unittest.expectedFailure  # TODO: RUSTPYTHON; old failure\n    def test_known",
).replace(
    "    def test_guarded",
    '    @unittest.skip("TODO: RUSTPYTHON; retained guard")\n    def test_guarded',
)


class MarkerOverrideTests(unittest.TestCase):
    def invoke(self, mutation=None, directory=False, override=True, legacy=False):
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary) / "repository"
            output = root / "Lib" / "test" / "test_example.py"
            output.parent.mkdir(parents=True)
            output.write_text(OVERRIDE_OLD)
            source = Path(temporary) / "cpython" / "Lib" / "test" / "test_example.py"
            source.parent.mkdir(parents=True)
            source.write_text(OVERRIDE_CLEAN)
            markers = patches_to_json(extract_patches(OVERRIDE_OLD))
            del markers["Example"]["test_known"]
            document = {"target": "Lib/test/test_example.py", "patches": markers}
            if mutation:
                mutation(document)
            metadata = Path(temporary) / "remaining.json"
            metadata.write_text(json.dumps(document))
            args = [str(source.parent if directory else source)]
            if override:
                args += ["--patches", str(metadata)]
            if legacy:
                args += ["--legacy-map", str(metadata)]
            with (
                contextlib.chdir(root),
                patch(
                    "update_lib.cmd_migrate.__file__",
                    str(root / "scripts/update_lib/cmd_migrate.py"),
                ),
                contextlib.redirect_stdout(io.StringIO()),
                contextlib.redirect_stderr(io.StringIO()),
            ):
                status = main(args)
            return status, output.read_text()

    def test_cli_override_removes_only_requested_expected_failure(self):
        status, result = self.invoke()
        self.assertEqual(status, 0)
        self.assertNotIn("expectedFailure", result)
        self.assertIn('unittest.skip("TODO: RUSTPYTHON; retained guard")', result)
        self.assertIn("self.assertEqual(1, 1)", result)
        self.assertIn("self.assertEqual(2, 2)", result)

    def test_default_migration_preserves_all_markers(self):
        status, result = self.invoke(override=False)
        self.assertEqual(status, 0)
        self.assertIn("expectedFailure", result)
        self.assertIn("retained guard", result)

    def test_invalid_target_does_not_write(self):
        status, result = self.invoke(
            lambda document: document.update(target="Lib/test/other.py")
        )
        self.assertEqual(status, 1)
        self.assertEqual(result, OVERRIDE_OLD)

    def test_removing_skip_does_not_write(self):
        status, result = self.invoke(
            lambda document: document["patches"]["Example"].clear()
        )
        self.assertEqual(status, 1)
        self.assertEqual(result, OVERRIDE_OLD)

    def test_directory_override_does_not_write(self):
        status, result = self.invoke(directory=True)
        self.assertEqual(status, 1)
        self.assertEqual(result, OVERRIDE_OLD)

    def test_legacy_combination_does_not_write(self):
        status, result = self.invoke(legacy=True)
        self.assertEqual(status, 1)
        self.assertEqual(result, OVERRIDE_OLD)

    def test_added_changed_or_reordered_markers_fail(self):
        failure = PatchSpec(UtMethod.ExpectedFailure, reason="known failure")
        success = PatchSpec(UtMethod.ExpectedSuccess, reason="known success")
        current = {"Example": {"test_known": [failure, success]}}
        for markers in (
            [failure, success, failure],
            [PatchSpec(UtMethod.ExpectedFailure, reason="changed"), success],
            [success, failure],
        ):
            document = {
                "target": "Lib/test/test_example.py",
                "patches": patches_to_json({"Example": {"test_known": markers}}),
            }
            with self.subTest(markers=markers), self.assertRaises(ValueError):
                select_remaining_patches(current, document, Path(document["target"]))

    def test_cli_narrows_unconditional_failure(self):
        def narrow(document):
            document["patches"]["Example"]["test_known"] = [
                {
                    "ut_method": "expectedFailureIf",
                    "cond": "MS_WINDOWS",
                    "reason": "old failure",
                }
            ]

        status, result = self.invoke(narrow)
        self.assertEqual(status, 0)
        self.assertIn(
            'unittest.expectedFailureIf(MS_WINDOWS, "TODO: RUSTPYTHON; old failure")',
            result,
        )
        self.assertIn('unittest.skip("TODO: RUSTPYTHON; retained guard")', result)
        self.assertIn("self.assertEqual(1, 1)", result)

    def test_narrowing_preserves_reason_order_and_existing_targets(self):
        failure = PatchSpec(UtMethod.ExpectedFailure, reason="known failure")
        narrowed = failure._replace(
            ut_method=UtMethod.ExpectedFailureIf, cond="MS_WINDOWS"
        )
        success = PatchSpec(UtMethod.ExpectedSuccess, reason="known success")
        current = {"Example": {"test_known": [failure, success]}}
        for markers in (
            [narrowed, success, narrowed],
            [narrowed._replace(reason="changed"), success],
            [success, narrowed],
            [narrowed._replace(ut_method=UtMethod.SkipIf), success],
            [narrowed._replace(ut_method=UtMethod.ExpectedSuccess), success],
            [narrowed._replace(cond=None), success],
        ):
            document = {
                "target": "Lib/test/test_example.py",
                "patches": patches_to_json({"Example": {"test_known": markers}}),
            }
            with self.subTest(markers=markers), self.assertRaises(ValueError):
                select_remaining_patches(current, document, Path(document["target"]))

        document["patches"] = patches_to_json({"Example": {"test_new": [narrowed]}})
        with self.assertRaises(ValueError):
            select_remaining_patches(current, document, Path(document["target"]))

    def test_conditional_failure_cannot_be_broadened_or_changed(self):
        conditional = PatchSpec(UtMethod.ExpectedFailureIf, cond="MS_WINDOWS")
        current = {"Example": {"test_known": [conditional]}}
        for marker in (
            conditional._replace(ut_method=UtMethod.ExpectedFailure, cond=None),
            conditional._replace(cond="True"),
            conditional._replace(cond="not MS_WINDOWS"),
        ):
            document = {
                "target": "Lib/test/test_example.py",
                "patches": patches_to_json({"Example": {"test_known": [marker]}}),
            }
            with self.subTest(marker=marker), self.assertRaises(ValueError):
                select_remaining_patches(current, document, Path(document["target"]))


class ConditionalMarkerOverrideTests(unittest.TestCase):
    decorator = (
        '    @unittest.expectedFailureIf(not support.is_android, "TODO: RUSTPYTHON")\n'
    )

    def fixture(self):
        old = OVERRIDE_CLEAN.replace(
            "    def test_known", self.decorator + "    def test_known"
        )
        current = extract_patches(old)
        document = {
            "target": "Lib/test/test_example.py",
            "patches": {},
            "retire_conditional": [
                {
                    "class": "Example",
                    "method": "test_known",
                    "decorator": self.decorator,
                }
            ],
        }
        return old, current, document

    def select(self, old, current, document):
        return select_remaining_patches(
            current,
            document,
            Path(document["target"]),
            old_content=old,
        )

    def test_explicit_conditional_retirement(self):
        old, current, document = self.fixture()
        self.assertEqual(self.select(old, current, document), {})

    def test_omitted_conditional_without_authorization_fails(self):
        old, current, document = self.fixture()
        del document["retire_conditional"]
        with self.assertRaises(ValueError):
            self.select(old, current, document)

    def test_absent_duplicate_modified_and_retained_fail_closed(self):
        for change in (
            "absent",
            "duplicate",
            "text",
            "condition",
            "retained",
            "duplicate_source",
        ):
            old, current, document = self.fixture()
            with self.subTest(change=change), self.assertRaises(ValueError):
                if change == "absent":
                    document["retire_conditional"][0]["method"] = "missing"
                elif change == "duplicate":
                    document["retire_conditional"] *= 2
                elif change == "text":
                    document["retire_conditional"][0]["decorator"] = (
                        self.decorator.replace('"', "'")
                    )
                elif change == "condition":
                    old = old.replace("not support.is_android", "support.is_android")
                    current = extract_patches(old)
                elif change == "retained":
                    document["patches"] = patches_to_json(current)
                elif change == "duplicate_source":
                    old = old.replace(self.decorator, self.decorator * 2)
                    current = extract_patches(old)
                self.select(old, current, document)

    def test_other_guards_and_original_source_required(self):
        for replacement in ("skipIf", "expectedFailureIfWindows"):
            old, current, document = self.fixture()
            old = old.replace("expectedFailureIf", replacement)
            document["retire_conditional"][0]["decorator"] = self.decorator.replace(
                "expectedFailureIf", replacement
            )
            with self.subTest(replacement=replacement), self.assertRaises(ValueError):
                self.select(old, extract_patches(old), document)
        old, current, document = self.fixture()
        with self.assertRaises(ValueError):
            self.select(None, current, document)

    def test_cli_conditional_retirement_preserves_remaining_markers(self):
        old = OVERRIDE_OLD.replace(
            "    @unittest.expectedFailure  # TODO: RUSTPYTHON; old failure\n",
            self.decorator,
        )

        def authorize(document):
            document["retire_conditional"] = [
                {
                    "class": "Example",
                    "method": "test_known",
                    "decorator": self.decorator,
                }
            ]

        with patch(__name__ + ".OVERRIDE_OLD", old):
            status, result = MarkerOverrideTests().invoke(authorize)
            self.assertEqual(status, 0)
            self.assertNotIn("expectedFailureIf", result)
            self.assertIn('unittest.skip("TODO: RUSTPYTHON; retained guard")', result)
            status, result = MarkerOverrideTests().invoke()
            self.assertEqual(status, 1)
            self.assertEqual(result, old)


PACKAGE_MARKED = """import unittest
class Example:
    @unittest.expectedFailure  # TODO: RUSTPYTHON; retained reason
    def test_one(self):
        pass
"""
PACKAGE_CANONICAL = """import unittest
class Example:
    def test_one(self):
        pass
"""


class TestMigratePackage(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        self.src = self.root / "canonical" / "package"
        self.src.mkdir(parents=True)
        (self.src / "__init__.py").write_text("# canonical package\n")
        (self.src / "test_part.py").write_text(PACKAGE_CANONICAL)
        self.dest = self.root / "Lib" / "test" / "package"
        self.dest.parent.mkdir(parents=True)
        self.old = self.dest.parent / "old.py"
        self.old.write_text(PACKAGE_MARKED)
        self.manifest = self.root / "map.json"
        self.mapping = {"old.py": {"Example": "test_part.py"}}

    def run_migration(self):
        self.manifest.write_text(json.dumps(self.mapping))
        migrate_package(self.src, self.dest, self.manifest)

    def assert_rejected(self):
        before = self.old.read_bytes()
        with self.assertRaises((ValueError, FileNotFoundError, SyntaxError)):
            self.run_migration()
        self.assertFalse(self.dest.exists())
        self.assertEqual(self.old.read_bytes(), before)

    def test_preserves_marker_and_canonical_asset(self):
        (self.src / "asset.dat").write_bytes(b"\x00canonical")
        self.run_migration()
        self.assertFalse(self.old.exists())
        self.assertEqual((self.dest / "asset.dat").read_bytes(), b"\x00canonical")
        self.assertEqual(
            list(iter_patches((self.dest / "test_part.py").read_text())),
            list(iter_patches(PACKAGE_MARKED)),
        )

    def test_preserves_inherited_override(self):
        (self.src / "test_part.py").write_text(
            PACKAGE_CANONICAL.replace("class Example:", "class Base:")
            + "\nclass Example(Base):\n    pass\n"
        )
        self.run_migration()
        text = (self.dest / "test_part.py").read_text()
        self.assertIn("return super().test_one()", text)
        self.assertEqual(list(iter_patches(text)), list(iter_patches(PACKAGE_MARKED)))

    def test_excludes_cache_directories_but_preserves_canonical_assets(self):
        for relative in ("__pycache__", "nested/__pycache__", "assets"):
            (self.src / relative).mkdir(parents=True)
        files = {
            "__pycache__/module.cpython-315.pyc": b"stale cache",
            "nested/__pycache__/module.cpython-315.pyc": b"nested cache",
            "assets/__pycache__": b"ordinary canonical file",
            "assets/fixture.pyc": b"canonical bytecode fixture",
        }
        for relative, content in files.items():
            (self.src / relative).write_bytes(content)
        self.run_migration()
        self.assertFalse((self.dest / "__pycache__").exists())
        self.assertFalse((self.dest / "nested" / "__pycache__").exists())
        for relative in ("assets/__pycache__", "assets/fixture.pyc"):
            self.assertEqual((self.dest / relative).read_bytes(), files[relative])
        for relative, content in files.items():
            self.assertEqual((self.src / relative).read_bytes(), content)

    def test_missing_method(self):
        (self.src / "test_part.py").write_text(
            PACKAGE_CANONICAL.replace("test_one", "test_other")
        )
        self.assert_rejected()

    def test_missing_class(self):
        (self.src / "test_part.py").write_text(
            PACKAGE_CANONICAL.replace("Example", "Other")
        )
        self.assert_rejected()

    def test_ambiguous_class(self):
        (self.src / "test_part.py").write_text(PACKAGE_CANONICAL + PACKAGE_CANONICAL)
        self.assert_rejected()

    def test_ambiguous_method(self):
        (self.src / "test_part.py").write_text(
            PACKAGE_CANONICAL + "    def test_one(self):\n        pass\n"
        )
        self.assert_rejected()

    def test_missing_destination(self):
        self.mapping["old.py"]["Example"] = "missing.py"
        self.assert_rejected()

    def test_unused_mapping(self):
        self.mapping["old.py"]["Unused"] = "test_part.py"
        self.assert_rejected()

    def test_unmapped_marker(self):
        self.mapping["old.py"] = {}
        self.assert_rejected()

    def test_target_traversal(self):
        self.mapping["old.py"]["Example"] = "../outside.py"
        self.assert_rejected()

    def test_legacy_traversal(self):
        self.mapping["../old.py"] = self.mapping.pop("old.py")
        self.assert_rejected()

    def test_duplicate_marker(self):
        (self.old.parent / "second.py").write_text(PACKAGE_MARKED)
        self.mapping["second.py"] = self.mapping["old.py"]
        self.assert_rejected()
        self.assertEqual((self.old.parent / "second.py").read_text(), PACKAGE_MARKED)

    def test_preserves_ordered_markers_from_one_legacy_method(self):
        marked = PACKAGE_MARKED.replace(
            "    def test_one",
            '    @unittest.skip("TODO: RUSTPYTHON; retained skip")\n    def test_one',
        )
        self.old.write_text(marked)
        self.run_migration()
        result = (self.dest / "test_part.py").read_text()
        self.assertEqual(list(iter_patches(result)), list(iter_patches(marked)))
        self.assertLess(result.index("expectedFailure"), result.index("unittest.skip"))

    def test_duplicate_legacy_owner_is_rejected(self):
        for source in (
            PACKAGE_MARKED + PACKAGE_MARKED,
            PACKAGE_MARKED + "    def test_one(self):\n        pass\n",
        ):
            with self.subTest(source=source):
                self.old.write_text(source)
                self.assert_rejected()

    def test_unmarked_legacy_retired(self):
        self.old.write_text(PACKAGE_CANONICAL)
        self.mapping["old.py"] = {}
        self.run_migration()
        self.assertFalse(self.old.exists())
        self.assertEqual((self.dest / "test_part.py").read_text(), PACKAGE_CANONICAL)

    def test_legacy_doctest_guards_prevent_all_writes(self):
        self.old.write_text(
            '"""\n'
            ">>> raise ValueError  # TODO: RUSTPYTHON; mismatch # doctest: +EXPECTED_FAILURE\n"
            "ValueError\n"
            '"""\n' + PACKAGE_MARKED
        )
        self.assert_rejected()

    def test_validation_failure_after_valid_entry_writes_nothing(self):
        self.mapping["missing.py"] = {}
        self.assert_rejected()

    def test_existing_destination_is_not_overwritten(self):
        self.dest.mkdir()
        with self.assertRaises(ValueError):
            self.run_migration()
        self.assertTrue(self.old.exists())
        self.assertEqual(list(self.dest.iterdir()), [])

    def test_source_escape_rejected(self):
        outside = self.root / "outside.py"
        outside.write_text(PACKAGE_CANONICAL)
        (self.src / "test_part.py").unlink()
        (self.src / "test_part.py").symlink_to(outside)
        self.assert_rejected()

    def test_legacy_symlink_rejected(self):
        outside = self.root / "outside.py"
        outside.write_text(PACKAGE_MARKED)
        self.old.unlink()
        self.old.symlink_to(outside)
        self.assert_rejected()

    def test_inheritance_cycle_rejected(self):
        (self.src / "test_part.py").write_text(
            "class Example(Base):\n    pass\nclass Base(Example):\n    pass\n"
        )
        self.assert_rejected()

    def test_canonical_marker_collision_rejected(self):
        (self.src / "test_part.py").write_text(PACKAGE_MARKED)
        self.assert_rejected()

    def test_unmarked_syntax_error_prevents_all_writes(self):
        (self.src / "broken.py").write_text("class :\n")
        self.assert_rejected()


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