tests/test_drop_plugin.py

   1 import sys, os, subprocess
   2 from tempfile import TemporaryDirectory
   3 from unittest import TestCase
   4 from pathlib import Path
   5
   6
   7 class CommitDropTest(TestCase):
   8     def test_drop_single_commit_by_hash(self):
   9         hash1 = self.create_commit('commit 1')
  10         self.create_commit('commit 2')
  11
  12         self.drop(hash1)
  13
  14         self.assertEqual(['commit 2'], self.git.log())
  15
  16     def test_drop_commits_by_desc(self):
  17         self.create_commit('commit 1 is good')
  18         self.create_commit('commit 2 is bad')
  19         self.create_commit('commit 3 is good')
  20         self.create_commit('commit 4 is bad')
  21
  22         self.drop('.*bad')
  23
  24         expected = ['commit 1 is good', 'commit 3 is good']
  25         self.assertEqual(expected, self.git.log())
  26
  27     def test_drop_sequential_commits_in_single_plugin_instance(self):
  28         self.create_commit('commit 1')
  29         hash2 = self.create_commit('commit 2')
  30         hash3 = self.create_commit('commit 3')
  31         hash4 = self.create_commit('commit 4')
  32         self.create_commit('commit 5')
  33
  34         self.drop(','.join((hash2, hash3, hash4)))
  35
  36         expected = ['commit 1', 'commit 5']
  37         self.assertEqual(expected, self.git.log())
  38
  39     def test_drop_sequential_commits_in_multiple_plugin_instances(self):
  40         self.create_commit('commit 1')
  41         hash2 = self.create_commit('commit 2')
  42         hash3 = self.create_commit('commit 3')
  43         hash4 = self.create_commit('commit 4')
  44         self.create_commit('commit 5')
  45
  46         self.drop(hash2, hash3, hash4)
  47
  48         expected = ['commit 1', 'commit 5']
  49         self.assertEqual(expected, self.git.log())
  50
  51     def test_drop_nonsequential_commits(self):
  52         self.create_commit('commit 1')
  53         hash2 = self.create_commit('commit 2')
  54         self.create_commit('commit 3')
  55         hash4 = self.create_commit('commit 4')
  56
  57         self.drop(','.join((hash2, hash4)))
  58
  59         expected = ['commit 1', 'commit 3']
  60         self.assertEqual(expected, self.git.log())
  61
  62     def test_drop_head(self):
  63         self.create_commit('first')
  64         self.create_commit('middle')
  65         hash_last = self.create_commit('last')
  66
  67         self.drop(hash_last)
  68
  69         self.assertEqual(['first', 'middle'], self.git.log())
  70
  71     def test_drop_merge_commit(self):
  72         initial_hash = self.create_commit('initial')
  73         self.create_commit('branch A')
  74         self.hg.checkout(initial_hash)
  75         self.create_commit('branch B')
  76         self.hg.merge()
  77         merge_hash = self.create_commit('merge to drop')
  78         self.create_commit('last')
  79
  80         self.drop(merge_hash)
  81
  82         expected_commits = ['initial', 'branch A', 'branch B', 'last']
  83         self.assertEqual(expected_commits, self.git.log())
  84         self.assertEqual(['branch B', 'branch A'], self.git_parents('last'))
  85
  86     def test_drop_different_commits_in_multiple_plugin_instances(self):
  87         self.create_commit('good commit')
  88         bad_hash = self.create_commit('bad commit')
  89         self.create_commit('awful commit')
  90         self.create_commit('another good commit')
  91
  92         self.drop('^awful.*', bad_hash)
  93
  94         expected = ['good commit', 'another good commit']
  95         self.assertEqual(expected, self.git.log())
  96
  97     def test_drop_same_commit_in_multiple_plugin_instances(self):
  98         self.create_commit('good commit')
  99         bad_hash = self.create_commit('bad commit')
 100         self.create_commit('another good commit')
 101
 102         self.drop('^bad.*', bad_hash)
 103
 104         expected = ['good commit', 'another good commit']
 105         self.assertEqual(expected, self.git.log())
 106
 107     def setUp(self):
 108         self.tempdir = TemporaryDirectory()
 109
 110         self.hg = HgDriver(Path(self.tempdir.name) / 'hgrepo')
 111         self.hg.init()
 112
 113         self.git = GitDriver(Path(self.tempdir.name) / 'gitrepo')
 114         self.git.init()
 115
 116         self.export = ExportDriver(self.hg.repodir, self.git.repodir)
 117
 118     def tearDown(self):
 119         self.tempdir.cleanup()
 120
 121     def create_commit(self, message):
 122         self.write_file_data('Data for %r.' % message)
 123         return self.hg.commit(message)
 124
 125     def write_file_data(self, data, filename='test_file.txt'):
 126         path = self.hg.repodir / filename
 127         with path.open('w') as f:
 128             print(data, file=f)
 129
 130     def drop(self, *spec):
 131         self.export.run_with_drop(*spec)
 132
 133     def git_parents(self, message):
 134         matches = self.git.grep_log(message)
 135         if len(matches) != 1:
 136             raise Exception('No unique commit with message %r.' % message)
 137         subject, parents = self.git.details(matches[0])
 138         return [self.git.details(p)[0] for p in parents]
 139
 140
 141 class ExportDriver:
 142     def __init__(self, sourcedir, targetdir, *, quiet=True):
 143         self.sourcedir = Path(sourcedir)
 144         self.targetdir = Path(targetdir)
 145         self.quiet = quiet
 146         self.python_executable = str(
 147             Path.cwd() / os.environ.get('PYTHON', sys.executable))
 148         self.script = Path(__file__).parent / '../hg-fast-export.sh'
 149
 150     def run_with_drop(self, *plugin_args):
 151         cmd = [self.script, '-r', str(self.sourcedir)]
 152         for arg in plugin_args:
 153             cmd.extend(['--plugin', 'drop=' + arg])
 154         output = subprocess.DEVNULL if self.quiet else None
 155         subprocess.run(cmd, check=True, cwd=str(self.targetdir),
 156                        env={'PYTHON': self.python_executable},
 157                        stdout=output, stderr=output)
 158
 159
 160 class HgDriver:
 161     def __init__(self, repodir):
 162         self.repodir = Path(repodir)
 163
 164     def init(self):
 165         self.repodir.mkdir()
 166         self.run_command('init')
 167
 168     def commit(self, message):
 169         self.run_command('commit', '-A', '-m', message)
 170         return self.run_command('id', '--id', '--debug').strip()
 171
 172     def log(self):
 173         output = self.run_command('log', '-T', '{desc}\n')
 174         commits = output.strip().splitlines()
 175         commits.reverse()
 176         return commits
 177
 178     def checkout(self, rev):
 179         self.run_command('checkout', '-r', rev)
 180
 181     def merge(self):
 182         self.run_command('merge', '--tool', ':local')
 183
 184     def run_command(self, *args):
 185         p = subprocess.run(('hg', '-yq') + args,
 186                            cwd=str(self.repodir),
 187                            check=True,
 188                            text=True,
 189                            capture_output=True)
 190         return p.stdout
 191
 192
 193 class GitDriver:
 194     def __init__(self, repodir):
 195         self.repodir = Path(repodir)
 196
 197     def init(self):
 198         self.repodir.mkdir()
 199         self.run_command('init')
 200
 201     def log(self):
 202         output = self.run_command('log', '--format=%s', '--reverse')
 203         return output.strip().splitlines()
 204
 205     def grep_log(self, pattern):
 206         output = self.run_command('log', '--format=%H',
 207                                   '-F', '--grep', pattern)
 208         return output.strip().splitlines()
 209
 210     def details(self, commit_hash):
 211         fmt = '%s%n%P'
 212         output = self.run_command('show', '-s', '--format=' + fmt,
 213                                   commit_hash)
 214         subject, parents = output.splitlines()
 215         return subject, parents.split()
 216
 217     def run_command(self, *args):
 218         p = subprocess.run(('git', '--no-pager') + args,
 219                            cwd=str(self.repodir),
 220                            check=True,
 221                            text=True,
 222                            capture_output=True)
 223         return p.stdout