test/git_test.py

   1 #!/usr/bin/env python
   2 """Tests various operations using the cola.git module
   3 """
   4 from __future__ import absolute_import, division, unicode_literals
   5
   6 import os
   7 import signal
   8 import time
   9 import unittest
  10
  11 try:
  12     from unittest.mock import patch
  13 except ImportError:
  14     from mock import patch
  15
  16 from cola import git
  17 from cola.compat import WIN32
  18 from cola.git import STDOUT
  19
  20
  21 class GitModuleTestCase(unittest.TestCase):
  22
  23     @patch('cola.git.is_git_dir')
  24     def test_find_git_dir_None(self, is_git_dir):
  25
  26         paths = git.find_git_directory(None)
  27
  28         self.assertFalse(is_git_dir.called)
  29         self.assertEqual(None, paths.git_dir)
  30         self.assertEqual(None, paths.git_file)
  31         self.assertEqual(None, paths.worktree)
  32
  33     @patch('cola.git.is_git_dir')
  34     def test_find_git_dir_empty_string(self, is_git_dir):
  35
  36         paths = git.find_git_directory('')
  37
  38         self.assertFalse(is_git_dir.called)
  39         self.assertEqual(None, paths.git_dir)
  40         self.assertEqual(None, paths.git_file)
  41         self.assertEqual(None, paths.worktree)
  42
  43     @patch('cola.git.is_git_dir')
  44     def test_find_git_dir_never_found(self, is_git_dir):
  45         is_git_dir.return_value = False
  46
  47         paths = git.find_git_directory('/does/not/exist')
  48
  49         self.assertTrue(is_git_dir.called)
  50         self.assertEqual(None, paths.git_dir)
  51         self.assertEqual(None, paths.git_file)
  52         self.assertEqual(None, paths.worktree)
  53
  54         self.assertEqual(8, is_git_dir.call_count)
  55         kwargs = {}
  56         is_git_dir.assert_has_calls([
  57             (('/does/not/exist',), kwargs),
  58             (('/does/not/exist/.git',), kwargs),
  59             (('/does/not',), kwargs),
  60             (('/does/not/.git',), kwargs),
  61             (('/does',), kwargs),
  62             (('/does/.git',), kwargs),
  63             (('/',), kwargs),
  64             (('/.git',), kwargs),
  65         ])
  66
  67     @patch('cola.git.is_git_dir')
  68     def test_find_git_dir_found_right_away(self, is_git_dir):
  69         git_dir = '/seems/to/exist/.git'
  70         worktree = '/seems/to/exist'
  71         is_git_dir.return_value = True
  72
  73         paths = git.find_git_directory(git_dir)
  74
  75         self.assertTrue(is_git_dir.called)
  76         self.assertEqual(git_dir, paths.git_dir)
  77         self.assertEqual(None, paths.git_file)
  78         self.assertEqual(worktree, paths.worktree)
  79
  80     @patch('cola.git.is_git_dir')
  81     def test_find_git_does_discovery(self, is_git_dir):
  82         git_dir = '/the/root/.git'
  83         worktree = '/the/root'
  84         is_git_dir.side_effect = lambda x: x == git_dir
  85
  86         paths = git.find_git_directory('/the/root/sub/dir')
  87
  88         self.assertEqual(git_dir, paths.git_dir)
  89         self.assertEqual(None, paths.git_file)
  90         self.assertEqual(worktree, paths.worktree)
  91
  92     @patch('cola.git.read_git_file')
  93     @patch('cola.git.is_git_file')
  94     @patch('cola.git.is_git_dir')
  95     def test_find_git_honors_git_files(self,
  96                                        is_git_dir,
  97                                        is_git_file,
  98                                        read_git_file):
  99         git_file = '/the/root/.git'
 100         worktree = '/the/root'
 101         git_dir = '/super/module/.git/modules/root'
 102
 103         is_git_dir.side_effect = lambda x: x == git_file
 104         is_git_file.side_effect = lambda x: x == git_file
 105         read_git_file.return_value = git_dir
 106
 107         paths = git.find_git_directory('/the/root/sub/dir')
 108
 109         self.assertEqual(git_dir, paths.git_dir)
 110         self.assertEqual(git_file, paths.git_file)
 111         self.assertEqual(worktree, paths.worktree)
 112
 113         kwargs = {}
 114         self.assertEqual(6, is_git_dir.call_count)
 115         is_git_dir.assert_has_calls([
 116             (('/the/root/sub/dir',), kwargs),
 117             (('/the/root/sub/dir/.git',), kwargs),
 118             (('/the/root/sub',), kwargs),
 119             (('/the/root/sub/.git',), kwargs),
 120             (('/the/root',), kwargs),
 121             (('/the/root/.git',), kwargs),
 122         ])
 123         read_git_file.assert_called_once_with('/the/root/.git')
 124
 125     @patch('cola.core.getenv')
 126     @patch('cola.git.is_git_dir')
 127     def test_find_git_honors_ceiling_dirs(self, is_git_dir, getenv):
 128
 129         git_dir = '/ceiling/.git'
 130         ceiling = '/tmp:/ceiling:/other/ceiling'
 131         is_git_dir.side_effect = lambda x: x == git_dir
 132
 133         def mock_getenv(k, v=None):
 134             if k == 'GIT_CEILING_DIRECTORIES':
 135                 return ceiling
 136             return v
 137
 138         getenv.side_effect = mock_getenv
 139
 140         paths = git.find_git_directory('/ceiling/sub/dir')
 141
 142         self.assertEqual(None, paths.git_dir)
 143         self.assertEqual(None, paths.git_file)
 144         self.assertEqual(None, paths.worktree)
 145
 146         self.assertEqual(4, is_git_dir.call_count)
 147         kwargs = {}
 148         is_git_dir.assert_has_calls([
 149             (('/ceiling/sub/dir',), kwargs),
 150             (('/ceiling/sub/dir/.git',), kwargs),
 151             (('/ceiling/sub',), kwargs),
 152             (('/ceiling/sub/.git',), kwargs),
 153         ])
 154
 155     @patch('cola.core.islink')
 156     @patch('cola.core.isdir')
 157     @patch('cola.core.isfile')
 158     def test_is_git_dir_finds_linked_repository(self, isfile, isdir, islink):
 159         dirs = set([
 160             '/foo',
 161             '/foo/.git',
 162             '/foo/.git/refs',
 163             '/foo/.git/objects',
 164             '/foo/.git/worktrees',
 165             '/foo/.git/worktrees/foo',
 166         ])
 167         files = set([
 168             '/foo/.git/HEAD',
 169             '/foo/.git/worktrees/foo/HEAD',
 170             '/foo/.git/worktrees/foo/index',
 171             '/foo/.git/worktrees/foo/commondir',
 172             '/foo/.git/worktrees/foo/gitdir',
 173         ])
 174         islink.return_value = False
 175         isfile.side_effect = lambda x: x in files
 176         isdir.side_effect = lambda x: x in dirs
 177
 178         self.assertTrue(git.is_git_dir('/foo/.git/worktrees/foo'))
 179         self.assertTrue(git.is_git_dir('/foo/.git'))
 180
 181
 182 class GitCommandTest(unittest.TestCase):
 183     """Runs tests using a git.Git instance"""
 184
 185     def setUp(self):
 186         """Creates a git.Git instance for later use"""
 187         self.git = git.Git()
 188
 189     def test_transform_kwargs(self):
 190         expect = []
 191         actual = self.git.transform_kwargs(foo=None, bar=False)
 192         self.assertEqual(expect, actual)
 193
 194         expect = ['-a']
 195         actual = self.git.transform_kwargs(a=True)
 196         self.assertEqual(expect, actual)
 197
 198         expect = ['--abc']
 199         actual = self.git.transform_kwargs(abc=True)
 200         self.assertEqual(expect, actual)
 201
 202         expect = ['-a1']
 203         actual = self.git.transform_kwargs(a=1)
 204         self.assertEqual(expect, actual)
 205
 206         expect = ['--abc=1']
 207         actual = self.git.transform_kwargs(abc=1)
 208         self.assertEqual(expect, actual)
 209
 210         expect = ['-abc']
 211         actual = self.git.transform_kwargs(a='bc')
 212         self.assertEqual(expect, actual)
 213
 214         expect = ['--abc=def']
 215         actual = self.git.transform_kwargs(abc='def')
 216         self.assertEqual(expect, actual)
 217
 218     def test_version(self):
 219         """Test running 'git version'"""
 220         version = self.git.version()[STDOUT]
 221         self.failUnless(version.startswith('git version'))
 222
 223     def test_tag(self):
 224         """Test running 'git tag'"""
 225         tags = self.git.tag()[STDOUT].splitlines()
 226         if os.getenv('GIT_COLA_NO_HISTORY', False):
 227             return
 228         self.failUnless('v1.0.0' in tags)
 229
 230     def test_show(self):
 231         """Test running 'git show'"""
 232         sha = 'HEAD'
 233         content = self.git.show(sha)[STDOUT]
 234         self.failUnless(content.startswith('commit '))
 235
 236     def test_stdout(self):
 237         """Test overflowing the stdout buffer"""
 238         # Write to stdout only
 239         code = ('import sys;'
 240                 's = "\\0" * (1024 * 16 + 1);'
 241                 'sys.stdout.write(s);')
 242         status, out, err = git.Git.execute(['python', '-c', code], _raw=True)
 243         self.assertEqual(status, 0)
 244         self.assertEqual(len(out), 1024 * 16 + 1)
 245         self.assertEqual(len(err), 0)
 246
 247     def test_stderr(self):
 248         """Test that stderr is seen"""
 249         # Write to stderr and capture it
 250         code = ('import sys;'
 251                 's = "\\0" * (1024 * 16 + 1);'
 252                 'sys.stderr.write(s);')
 253         status, out, err = git.Git.execute(['python', '-c', code], _raw=True)
 254         self.assertEqual(status, 0)
 255         self.assertEqual(len(out), 0)
 256         self.assertEqual(len(err), 1024 * 16 + 1)
 257
 258     def test_stdout_and_stderr(self):
 259         """Test ignoring stderr when stdout+stderr are provided (v2)"""
 260         # Write to stdout and stderr but only capture stdout
 261         code = ('import sys;'
 262                 's = "\\0" * (1024 * 16 + 1);'
 263                 'sys.stdout.write(s);'
 264                 'sys.stderr.write(s);')
 265         status, out, err = git.Git.execute(['python', '-c', code], _raw=True)
 266         self.assertEqual(status, 0)
 267         self.assertEqual(len(out), 1024 * 16 + 1)
 268         self.assertEqual(len(err), 1024 * 16 + 1)
 269
 270     def test_it_doesnt_deadlock(self):
 271         """Test that we don't deadlock with both stderr and stdout"""
 272         # 16k+1 bytes to exhaust any output buffers
 273         code = ('import sys;'
 274                 's = "\\0" * (1024 * 16 + 1);'
 275                 'sys.stderr.write(s);'
 276                 'sys.stdout.write(s);')
 277         status, out, err = git.Git.execute(['python', '-c', code], _raw=True)
 278         self.assertEqual(status, 0)
 279         self.assertEqual(out, '\0' * (1024 * 16 + 1))
 280         self.assertEqual(err, '\0' * (1024 * 16 + 1))
 281
 282     def test_it_handles_interrupted_syscalls(self):
 283         """Test that we handle interrupted system calls"""
 284         # send ourselves a signal that causes EINTR
 285         if WIN32:
 286             # SIGALRM not supported on Windows
 287             return
 288         prev_handler = signal.signal(signal.SIGALRM, lambda x, y: 1)
 289         signal.alarm(1)
 290         time.sleep(0.1)
 291         status, out, err = git.Git.execute(['sleep', '1'])
 292         self.assertEqual(status, 0)
 293
 294         signal.signal(signal.SIGALRM, prev_handler)
 295
 296
 297 if __name__ == '__main__':
 298     unittest.main()