test/git_test.py

   1 #!/usr/bin/env python
   2 """Tests various operations using the cola.git module
   3 """
   4 from __future__ import unicode_literals
   5
   6 import time
   7 import signal
   8 import unittest
   9
  10 from cola import git
  11 from cola.compat import WIN32
  12 from cola.git import STDOUT
  13
  14
  15 class GitCommandTest(unittest.TestCase):
  16     """Runs tests using a git.Git instance"""
  17
  18     def setUp(self):
  19         """Creates a git.Git instance for later use"""
  20         self.git = git.Git()
  21
  22     def test_transform_kwargs(self):
  23         expect = []
  24         actual = self.git.transform_kwargs(foo=None, bar=False)
  25         self.assertEqual(expect, actual)
  26
  27         expect = ['-a']
  28         actual = self.git.transform_kwargs(a=True)
  29         self.assertEqual(expect, actual)
  30
  31         expect = ['--abc']
  32         actual = self.git.transform_kwargs(abc=True)
  33         self.assertEqual(expect, actual)
  34
  35         expect = ['-a1']
  36         actual = self.git.transform_kwargs(a=1)
  37         self.assertEqual(expect, actual)
  38
  39         expect = ['--abc=1']
  40         actual = self.git.transform_kwargs(abc=1)
  41         self.assertEqual(expect, actual)
  42
  43         expect = ['-abc']
  44         actual = self.git.transform_kwargs(a='bc')
  45         self.assertEqual(expect, actual)
  46
  47         expect = ['--abc=def']
  48         actual = self.git.transform_kwargs(abc='def')
  49         self.assertEqual(expect, actual)
  50
  51     def test_version(self):
  52         """Test running 'git version'"""
  53         version = self.git.version()[STDOUT]
  54         self.failUnless(version.startswith('git version'))
  55
  56     def test_tag(self):
  57         """Test running 'git tag'"""
  58         tags = self.git.tag()[STDOUT].splitlines()
  59         self.failUnless( 'v1.0.0' in tags )
  60
  61     def test_show(self):
  62         """Test running 'git show'"""
  63         sha = '1b9742bda5d26a4f250fa64657f66ed20624a084'
  64         contents = self.git.show(sha)[STDOUT].splitlines()
  65         self.failUnless(contents[0] == '/build')
  66
  67     def test_stdout(self):
  68         """Test overflowing the stdout buffer"""
  69         # Write to stdout only
  70         code = ('import sys;'
  71                 's = "\\0" * (1024 * 16 + 1);'
  72                 'sys.stdout.write(s);')
  73         status, out, err = git.Git.execute(['python', '-c', code], _raw=True)
  74         self.assertEqual(status, 0)
  75         self.assertEqual(len(out), 1024 * 16 + 1)
  76         self.assertEqual(len(err), 0)
  77
  78     def test_stderr(self):
  79         """Test that stderr is seen"""
  80         # Write to stderr and capture it
  81         code = ('import sys;'
  82                 's = "\\0" * (1024 * 16 + 1);'
  83                 'sys.stderr.write(s);')
  84         status, out, err = git.Git.execute(['python', '-c', code], _raw=True)
  85         self.assertEqual(status, 0)
  86         self.assertEqual(len(out), 0)
  87         self.assertEqual(len(err), 1024 * 16 + 1)
  88
  89     def test_stdout_and_stderr(self):
  90         """Test ignoring stderr when stdout+stderr are provided (v2)"""
  91         # Write to stdout and stderr but only capture stdout
  92         code = ('import sys;'
  93                 's = "\\0" * (1024 * 16 + 1);'
  94                 'sys.stdout.write(s);'
  95                 'sys.stderr.write(s);')
  96         status, out, err = git.Git.execute(['python', '-c', code], _raw=True)
  97         self.assertEqual(status, 0)
  98         self.assertEqual(len(out), 1024 * 16 + 1)
  99         self.assertEqual(len(err), 1024 * 16 + 1)
 100
 101     def test_it_doesnt_deadlock(self):
 102         """Test that we don't deadlock with both stderr and stdout"""
 103         # 16k+1 bytes to exhaust any output buffers
 104         code = ('import sys;'
 105                 's = "\\0" * (1024 * 16 + 1);'
 106                 'sys.stderr.write(s);'
 107                 'sys.stdout.write(s);')
 108         status, out, err = git.Git.execute(['python', '-c', code], _raw=True)
 109         self.assertEqual(status, 0)
 110         self.assertEqual(out, '\0' * (1024 * 16 + 1))
 111         self.assertEqual(err, '\0' * (1024 * 16 + 1))
 112
 113     def test_it_handles_interrupted_syscalls(self):
 114         """Test that we handle interrupted system calls"""
 115         # send ourselves a signal that causes EINTR
 116         if WIN32:
 117             # SIGALRM not supported on Windows
 118             return
 119         prev_handler = signal.signal(signal.SIGALRM, lambda x, y: 1)
 120         signal.alarm(1)
 121         time.sleep(0.1)
 122         status, out, err = git.Git.execute(['sleep', '1'])
 123         self.assertEqual(status, 0)
 124
 125         signal.signal(signal.SIGALRM, prev_handler)
 126
 127
 128 if __name__ == '__main__':
 129     unittest.main()