test/test_cola_git.py

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