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 from cola.git import STDOUT
  11
  12
  13 class GitCommandTest(unittest.TestCase):
  14     """Runs tests using a git.Git instance"""
  15
  16     def setUp(self):
  17         """Creates a git.Git instance for later use"""
  18         self.git = git.Git()
  19
  20     def test_version(self):
  21         """Test running 'git version'"""
  22         version = self.git.version()[STDOUT]
  23         self.failUnless(version.startswith('git version'))
  24
  25     def test_tag(self):
  26         """Test running 'git tag'"""
  27         tags = self.git.tag()[STDOUT].splitlines()
  28         self.failUnless( 'v1.0.0' in tags )
  29
  30     def test_show(self):
  31         """Test running 'git show'"""
  32         sha = '1b9742bda5d26a4f250fa64657f66ed20624a084'
  33         contents = self.git.show(sha)[STDOUT].splitlines()
  34         self.failUnless(contents[0] == '/build')
  35
  36     def test_stdout(self):
  37         """Test overflowing the stdout buffer"""
  38         # Write to stdout only
  39         code = ('import sys;'
  40                 's = "\\0" * (1024 * 16 + 1);'
  41                 'sys.stdout.write(s);')
  42         status, out, err = git.Git.execute(['python', '-c', code], _raw=True)
  43         self.assertEqual(status, 0)
  44         self.assertEqual(len(out), 1024 * 16 + 1)
  45         self.assertEqual(len(err), 0)
  46
  47     def test_stderr(self):
  48         """Test that stderr is seen"""
  49         # Write to stderr and capture it
  50         code = ('import sys;'
  51                 's = "\\0" * (1024 * 16 + 1);'
  52                 'sys.stderr.write(s);')
  53         status, out, err = git.Git.execute(['python', '-c', code], _raw=True)
  54         self.assertEqual(status, 0)
  55         self.assertEqual(len(out), 0)
  56         self.assertEqual(len(err), 1024 * 16 + 1)
  57
  58     def test_stdout_and_stderr(self):
  59         """Test ignoring stderr when stdout+stderr are provided (v2)"""
  60         # Write to stdout and stderr but only capture stdout
  61         code = ('import sys;'
  62                 's = "\\0" * (1024 * 16 + 1);'
  63                 'sys.stdout.write(s);'
  64                 'sys.stderr.write(s);')
  65         status, out, err = git.Git.execute(['python', '-c', code], _raw=True)
  66         self.assertEqual(status, 0)
  67         self.assertEqual(len(out), 1024 * 16 + 1)
  68         self.assertEqual(len(err), 1024 * 16 + 1)
  69
  70     def test_it_doesnt_deadlock(self):
  71         """Test that we don't deadlock with both stderr and stdout"""
  72         # 16k+1 bytes to exhaust any output buffers
  73         code = ('import sys;'
  74                 's = "\\0" * (1024 * 16 + 1);'
  75                 'sys.stderr.write(s);'
  76                 'sys.stdout.write(s);')
  77         status, out, err = git.Git.execute(['python', '-c', code], _raw=True)
  78         self.assertEqual(status, 0)
  79         self.assertEqual(out, '\0' * (1024 * 16 + 1))
  80         self.assertEqual(err, '\0' * (1024 * 16 + 1))
  81
  82     def test_it_handles_interrupted_syscalls(self):
  83         """Test that we handle interrupted system calls"""
  84         # send ourselves a signal that causes EINTR
  85         prev_handler = signal.signal(signal.SIGALRM, lambda x,y: 1)
  86         signal.alarm(1)
  87         time.sleep(0.5)
  88         status, out, err = git.Git.execute(['sleep', '1'])
  89         self.assertEqual(status, 0)
  90
  91         signal.signal(signal.SIGALRM, prev_handler)
  92
  93 if __name__ == '__main__':
  94     unittest.main()