Implement _ProvidedDiffCache.GetOldContents

Get old contents for a file using the provided diff.

Bug: b/323243527
Change-Id: Ib0e42cb1efc218002fae0b07c248b2806816b7c3
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/tools/depot_tools/+/5304396
Commit-Queue: Gavin Mak <gavinmak@google.com>
Reviewed-by: Josip Sokcevic <sokcevic@chromium.org>

presubmit_support.py

@@ -27,6 +27,7 @@ import os  # Somewhat exposed through the API.
 import pathlib
 import random
 import re  # Exposed through the API.
+import shutil
 import signal
 import sys  # Parts exposed through API.
 import tempfile  # Exposed through the API.
@@ -977,8 +978,39 @@ class _ProvidedDiffCache(_DiffCache):
 
     def GetOldContents(self, path, local_root):
         """Get the old version for a particular path."""
-        # TODO(gavinmak): Implement with self._diff.
-        return ''
+        full_path = os.path.join(local_root, path)
+        diff = self.GetDiff(path, local_root)
+        is_file = os.path.isfile(full_path)
+        if not diff:
+            if is_file:
+                return gclient_utils.FileRead(full_path)
+            return ''
+
+        with gclient_utils.temporary_file() as diff_file:
+            gclient_utils.FileWrite(diff_file, diff)
+            try:
+                scm.GIT.Capture(['apply', '--reverse', '--check', diff_file],
+                                cwd=local_root)
+            except subprocess.CalledProcessError:
+                raise RuntimeError('Provided diff does not apply cleanly.')
+
+            # Apply the reverse diff to a temporary file and read its contents.
+            with gclient_utils.temporary_directory() as tmp_dir:
+                copy_dst = os.path.join(tmp_dir, path)
+                os.makedirs(os.path.dirname(copy_dst), exist_ok=True)
+                if is_file:
+                    shutil.copyfile(full_path, copy_dst)
+                scm.GIT.Capture([
+                    'apply', '--reverse', '--directory', tmp_dir,
+                    '--unsafe-paths', diff_file
+                ],
+                                cwd=tmp_dir)
+                # Applying the patch can create a new file if the file at
+                # full_path was deleted, so check if the new file at copy_dst
+                # exists.
+                if os.path.isfile(copy_dst):
+                    return gclient_utils.FileRead(copy_dst)
+                return ''
 
 
 class AffectedFile(object):
@@ -2064,10 +2096,10 @@ def _parse_change(parser, options):
         options.name, options.description, options.root, change_files,
         options.issue, options.patchset, options.author
     ]
-    if change_scm == 'git':
-        return GitChange(*change_args, upstream=options.upstream)
     if diff:
         return ProvidedDiffChange(*change_args, diff=diff)
+    if change_scm == 'git':
+        return GitChange(*change_args, upstream=options.upstream)
     return Change(*change_args)
 
 
@@ -2164,6 +2196,7 @@ def _process_diff_file(diff_file):
         change_files.append((action, file))
     return diff, change_files
 
+
 @contextlib.contextmanager
 def setup_environ(kv: Mapping[str, str]):
     """Update environment while in context, and reset back to original on exit.

tests/presubmit_support_test.py

@@ -5,14 +5,16 @@
 
 import os.path
 import sys
-import tempfile
 import unittest
+from unittest import mock
 
 ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 sys.path.insert(0, ROOT_DIR)
 
 import gclient_utils
 import presubmit_support
+import subprocess2
+from testing_support import fake_repos
 
 
 class PresubmitSupportTest(unittest.TestCase):
@@ -25,6 +27,91 @@ class PresubmitSupportTest(unittest.TestCase):
         self.assertIsNone(os.environ.get('PRESUBMIT_FOO_ENV', None))
 
 
+class ProvidedDiffChangeFakeRepo(fake_repos.FakeReposBase):
+
+    NB_GIT_REPOS = 1
+
+    def populateGit(self):
+        self._commit_git(
+            'repo_1', {
+                'to_be_modified': 'please change me\n',
+                'to_be_deleted': 'delete\nme\n',
+                'somewhere/else': 'not a top level file!\n',
+            })
+        self._commit_git(
+            'repo_1', {
+                'to_be_modified': 'changed me!\n',
+                'to_be_deleted': None,
+                'somewhere/else': 'still not a top level file!\n',
+                'added': 'a new file\n',
+            })
+
+
+class ProvidedDiffChangeTest(fake_repos.FakeReposTestBase):
+
+    FAKE_REPOS_CLASS = ProvidedDiffChangeFakeRepo
+
+    def setUp(self):
+        super(ProvidedDiffChangeTest, self).setUp()
+        self.enabled = self.FAKE_REPOS.set_up_git()
+        if not self.enabled:
+            self.skipTest('git fake repos not available')
+        self.repo = os.path.join(self.FAKE_REPOS.git_base, 'repo_1')
+        diff = subprocess2.check_output(['git', 'show'],
+                                        cwd=self.repo).decode('utf-8')
+        self.change = self._create_change(diff)
+
+    def _create_change(self, diff):
+        with gclient_utils.temporary_file() as tmp:
+            gclient_utils.FileWrite(tmp, diff)
+            options = mock.Mock(root=self.repo,
+                                all_files=False,
+                                description='description',
+                                files=None,
+                                diff_file=tmp)
+            change = presubmit_support._parse_change(None, options)
+            assert isinstance(change, presubmit_support.ProvidedDiffChange)
+            return change
+
+    def _get_affected_file_from_name(self, change, name):
+        for file in change._affected_files:
+            if file.LocalPath() == os.path.normpath(name):
+                return file
+        self.fail(f'No file named {name}')
+
+    def _test(self, name, old, new):
+        affected_file = self._get_affected_file_from_name(self.change, name)
+        self.assertEqual(affected_file.OldContents(), old)
+        self.assertEqual(affected_file.NewContents(), new)
+
+    def test_old_contents_of_added_file_returns_empty(self):
+        self._test('added', [], ['a new file'])
+
+    def test_old_contents_of_deleted_file_returns_whole_file(self):
+        self._test('to_be_deleted', ['delete', 'me'], [])
+
+    def test_old_contents_of_modified_file(self):
+        self._test('to_be_modified', ['please change me'], ['changed me!'])
+
+    def test_old_contents_of_file_with_nested_dirs(self):
+        self._test('somewhere/else', ['not a top level file!'],
+                   ['still not a top level file!'])
+
+    def test_old_contents_of_bad_diff_raises_runtimeerror(self):
+        diff = """
+diff --git a/foo b/foo
+new file mode 100644
+index 0000000..9daeafb
+--- /dev/null
++++ b/foo
+@@ -0,0 +1 @@
++add
+"""
+        change = self._create_change(diff)
+        with self.assertRaises(RuntimeError):
+            change._affected_files[0].OldContents()
+
+
 class TestParseDiff(unittest.TestCase):
     """A suite of tests related to diff parsing and processing."""