test/test_cola_git.py

   1 #!/usr/bin/env python
   2 """Tests various operations using the cola.git module
   3 """
   4
   5 import time
   6 import signal
   7 import unittest
   8
   9 from cola import git
  10
  11
  12 class GitCommandTest(unittest.TestCase):
  13     """Runs tests using a git.Git instance"""
  14
  15     def setUp(self):
  16         """Creates a git.Git instance for later use"""
  17         self.git = git.Git()
  18
  19     def test_version(self):
  20         """Test running 'git version'"""
  21         version = self.git.version()
  22         self.failUnless(version.startswith('git version'))
  23
  24     def test_tag(self):
  25         """Test running 'git tag'"""
  26         tags = self.git.tag().splitlines()
  27         self.failUnless( 'v1.0.0' in tags )
  28
  29     def test_show(self):
  30         """Test running 'git show'"""
  31         sha = '1b9742bda5d26a4f250fa64657f66ed20624a084'
  32         contents = self.git.show(sha).splitlines()
  33         self.failUnless(contents[0] == '/build')
  34
  35     def test_stdout(self):
  36         """Test overflowing the stdout buffer"""
  37         # Write to stdout only
  38         code = ('import sys;'
  39                 's = "\\0" * (1024 * 16 + 1);'
  40                 'sys.stdout.write(s);')
  41         status, out = git.Git.execute(['python', '-c', code],
  42                                       with_status=True,
  43                                       with_raw_output=True)
  44         self.assertEqual(status, 0)
  45         self.assertEqual(len(out), 1024 * 16 + 1)
  46
  47     def test_stderr_empty(self):
  48         """Test that stderr is ignored by execute() without with_stderr"""
  49         # Write to stderr but ignore it
  50         code = ('import sys;'
  51                 's = "\\0" * (1024 * 16 + 1);'
  52                 'sys.stderr.write(s);')
  53         status, out = git.Git.execute(['python', '-c', code],
  54                                       with_status=True,
  55                                       with_raw_output=True)
  56         self.assertEqual(status, 0)
  57         self.assertEqual(len(out), 0)
  58
  59     def test_stderr_nonempty_with_stderr(self):
  60         """Test that with_stderr makes execute() see stderr"""
  61         # Write to stderr and capture it
  62         code = ('import sys;'
  63                 's = "\\0" * (1024 * 16 + 1);'
  64                 'sys.stderr.write(s);')
  65         status, out = git.Git.execute(['python', '-c', code],
  66                                       with_status=True,
  67                                       with_stderr=True,
  68                                       with_raw_output=True)
  69         self.assertEqual(status, 0)
  70         self.assertEqual(len(out), 1024 * 16 + 1)
  71
  72     def test_stdout_and_stderr_ignores_stderr(self):
  73         """Test ignoring stderr when stdout+stderr are provided"""
  74         # Write to stdout only
  75         code = ('import sys;'
  76                 's = "\\0" * (1024 * 16 + 1);'
  77                 'sys.stdout.write(s);')
  78         status, out = git.Git.execute(['python', '-c', code],
  79                                       with_status=True,
  80                                       with_raw_output=True)
  81         self.assertEqual(status, 0)
  82         self.assertEqual(len(out), 1024 * 16 + 1)
  83
  84     def test_stdout_and_stderr_ignores_stderr_v2(self):
  85         """Test ignoring stderr when stdout+stderr are provided (v2)"""
  86         # Write to stdout and stderr but only capture stdout
  87         code = ('import sys;'
  88                 's = "\\0" * (1024 * 16 + 1);'
  89                 'sys.stdout.write(s);'
  90                 'sys.stderr.write(s);')
  91         status, out = git.Git.execute(['python', '-c', code],
  92                                       with_status=True,
  93                                       with_raw_output=True)
  94         self.assertEqual(status, 0)
  95         self.assertEqual(len(out), 1024 * 16 + 1)
  96
  97     def test_stdout_and_stderr_sees_stderr(self):
  98         """Test seeing both stderr and stdout when both are available"""
  99         # Write to stdout and stderr and capture both
 100         code = ('import sys;'
 101                 's = "\\0" * (1024 * 16 + 1);'
 102                 'sys.stdout.write(s);'
 103                 'sys.stderr.write(s);')
 104         status, out = git.Git.execute(['python', '-c', code],
 105                                       with_status=True,
 106                                       with_stderr=True,
 107                                       with_raw_output=True)
 108         self.assertEqual(status, 0)
 109         self.assertEqual(len(out), 1024 * 16 * 2 + 2)
 110
 111     def test_stdout_and_stderr_sees_stderr_v2(self):
 112         """Test seeing both stderr and stdout when both are available (v2)."""
 113         # Write to stderr and stdout (swapped) and capture both
 114         code = ('import sys;'
 115                 's = "\\0" * (1024 * 16 + 1);'
 116                 'sys.stderr.write(s);'
 117                 'sys.stdout.write(s);')
 118         status, out = git.Git.execute(['python', '-c', code],
 119                                       # otherwise, same as above
 120                                       with_status=True,
 121                                       with_stderr=True,
 122                                       with_raw_output=True)
 123         self.assertEqual(status, 0)
 124         self.assertEqual(len(out), 1024 * 16 * 2 + 2)
 125
 126     def test_it_doesnt_deadlock(self):
 127         """Test that we don't deadlock with both stderr and stdout"""
 128         # 16k+1 bytes to exhaust any output buffers
 129         code = ('import sys;'
 130                 's = "\\0" * (1024 * 16 + 1);'
 131                 'sys.stderr.write(s);'
 132                 'sys.stdout.write(s);')
 133         out = git.Git.execute(['python', '-c', code])
 134         self.assertEqual(out, '\0' * (1024 * 16 + 1))
 135
 136     def test_it_handles_interrupted_syscalls(self):
 137         """Test that we handle interrupted system calls"""
 138         # send ourselves a signal that causes EINTR
 139         prev_handler = signal.signal(signal.SIGALRM, lambda x,y: 1)
 140         signal.alarm(1)
 141         time.sleep(0.5)
 142
 143         status, output = git.Git.execute(['sleep', '1'],
 144                                          with_status=True)
 145         self.assertEqual(status, 0)
 146
 147         signal.signal(signal.SIGALRM, prev_handler)
 148
 149 if __name__ == '__main__':
 150     unittest.main()