1# -*- coding: utf-8 -*- 2# test_git.py 3# Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors 4# 5# This module is part of GitPython and is released under 6# the BSD License: http://www.opensource.org/licenses/bsd-license.php 7import os 8import subprocess 9import sys 10 11from git import ( 12 Git, 13 refresh, 14 GitCommandError, 15 GitCommandNotFound, 16 Repo, 17 cmd 18) 19from git.compat import PY3, is_darwin 20from git.test.lib import ( 21 TestBase, 22 patch, 23 raises, 24 assert_equal, 25 assert_true, 26 assert_match, 27 fixture_path 28) 29from git.test.lib import with_rw_directory 30from git.util import finalize_process 31 32import os.path as osp 33 34 35try: 36 from unittest import mock 37except ImportError: 38 import mock 39 40from git.compat import is_win 41 42 43class TestGit(TestBase): 44 45 @classmethod 46 def setUpClass(cls): 47 super(TestGit, cls).setUpClass() 48 cls.git = Git(cls.rorepo.working_dir) 49 50 def tearDown(self): 51 import gc 52 gc.collect() 53 54 @patch.object(Git, 'execute') 55 def test_call_process_calls_execute(self, git): 56 git.return_value = '' 57 self.git.version() 58 assert_true(git.called) 59 assert_equal(git.call_args, ((['git', 'version'],), {})) 60 61 def test_call_unpack_args_unicode(self): 62 args = Git._Git__unpack_args(u'Unicode€™') 63 if PY3: 64 mangled_value = 'Unicode\u20ac\u2122' 65 else: 66 mangled_value = 'Unicode\xe2\x82\xac\xe2\x84\xa2' 67 assert_equal(args, [mangled_value]) 68 69 def test_call_unpack_args(self): 70 args = Git._Git__unpack_args(['git', 'log', '--', u'Unicode€™']) 71 if PY3: 72 mangled_value = 'Unicode\u20ac\u2122' 73 else: 74 mangled_value = 'Unicode\xe2\x82\xac\xe2\x84\xa2' 75 assert_equal(args, ['git', 'log', '--', mangled_value]) 76 77 @raises(GitCommandError) 78 def test_it_raises_errors(self): 79 self.git.this_does_not_exist() 80 81 def test_it_transforms_kwargs_into_git_command_arguments(self): 82 assert_equal(["-s"], self.git.transform_kwargs(**{'s': True})) 83 assert_equal(["-s", "5"], self.git.transform_kwargs(**{'s': 5})) 84 85 assert_equal(["--max-count"], self.git.transform_kwargs(**{'max_count': True})) 86 assert_equal(["--max-count=5"], self.git.transform_kwargs(**{'max_count': 5})) 87 88 # Multiple args are supported by using lists/tuples 89 assert_equal(["-L", "1-3", "-L", "12-18"], self.git.transform_kwargs(**{'L': ('1-3', '12-18')})) 90 assert_equal(["-C", "-C"], self.git.transform_kwargs(**{'C': [True, True]})) 91 92 # order is undefined 93 res = self.git.transform_kwargs(**{'s': True, 't': True}) 94 self.assertEqual({'-s', '-t'}, set(res)) 95 96 def test_it_executes_git_to_shell_and_returns_result(self): 97 assert_match(r'^git version [\d\.]{2}.*$', self.git.execute(["git", "version"])) 98 99 def test_it_accepts_stdin(self): 100 filename = fixture_path("cat_file_blob") 101 with open(filename, 'r') as fh: 102 assert_equal("70c379b63ffa0795fdbfbc128e5a2818397b7ef8", 103 self.git.hash_object(istream=fh, stdin=True)) 104 105 @patch.object(Git, 'execute') 106 def test_it_ignores_false_kwargs(self, git): 107 # this_should_not_be_ignored=False implies it *should* be ignored 108 self.git.version(pass_this_kwarg=False) 109 assert_true("pass_this_kwarg" not in git.call_args[1]) 110 111 def test_it_accepts_environment_variables(self): 112 filename = fixture_path("ls_tree_empty") 113 with open(filename, 'r') as fh: 114 tree = self.git.mktree(istream=fh) 115 env = { 116 'GIT_AUTHOR_NAME': 'Author Name', 117 'GIT_AUTHOR_EMAIL': 'author@example.com', 118 'GIT_AUTHOR_DATE': '1400000000+0000', 119 'GIT_COMMITTER_NAME': 'Committer Name', 120 'GIT_COMMITTER_EMAIL': 'committer@example.com', 121 'GIT_COMMITTER_DATE': '1500000000+0000', 122 } 123 commit = self.git.commit_tree(tree, m='message', env=env) 124 assert_equal(commit, '4cfd6b0314682d5a58f80be39850bad1640e9241') 125 126 def test_persistent_cat_file_command(self): 127 # read header only 128 import subprocess as sp 129 hexsha = "b2339455342180c7cc1e9bba3e9f181f7baa5167" 130 g = self.git.cat_file(batch_check=True, istream=sp.PIPE, as_process=True) 131 g.stdin.write(b"b2339455342180c7cc1e9bba3e9f181f7baa5167\n") 132 g.stdin.flush() 133 obj_info = g.stdout.readline() 134 135 # read header + data 136 g = self.git.cat_file(batch=True, istream=sp.PIPE, as_process=True) 137 g.stdin.write(b"b2339455342180c7cc1e9bba3e9f181f7baa5167\n") 138 g.stdin.flush() 139 obj_info_two = g.stdout.readline() 140 self.assertEqual(obj_info, obj_info_two) 141 142 # read data - have to read it in one large chunk 143 size = int(obj_info.split()[2]) 144 g.stdout.read(size) 145 g.stdout.read(1) 146 147 # now we should be able to read a new object 148 g.stdin.write(b"b2339455342180c7cc1e9bba3e9f181f7baa5167\n") 149 g.stdin.flush() 150 self.assertEqual(g.stdout.readline(), obj_info) 151 152 # same can be achieved using the respective command functions 153 hexsha, typename, size = self.git.get_object_header(hexsha) 154 hexsha, typename_two, size_two, data = self.git.get_object_data(hexsha) # @UnusedVariable 155 self.assertEqual(typename, typename_two) 156 self.assertEqual(size, size_two) 157 158 def test_version(self): 159 v = self.git.version_info 160 self.assertIsInstance(v, tuple) 161 for n in v: 162 self.assertIsInstance(n, int) 163 # END verify number types 164 165 def test_cmd_override(self): 166 prev_cmd = self.git.GIT_PYTHON_GIT_EXECUTABLE 167 exc = GitCommandNotFound 168 try: 169 # set it to something that doens't exist, assure it raises 170 type(self.git).GIT_PYTHON_GIT_EXECUTABLE = osp.join( 171 "some", "path", "which", "doesn't", "exist", "gitbinary") 172 self.failUnlessRaises(exc, self.git.version) 173 finally: 174 type(self.git).GIT_PYTHON_GIT_EXECUTABLE = prev_cmd 175 # END undo adjustment 176 177 def test_refresh(self): 178 # test a bad git path refresh 179 self.assertRaises(GitCommandNotFound, refresh, "yada") 180 181 # test a good path refresh 182 which_cmd = "where" if is_win else "which" 183 path = os.popen("{0} git".format(which_cmd)).read().strip().split('\n')[0] 184 refresh(path) 185 186 def test_options_are_passed_to_git(self): 187 # This work because any command after git --version is ignored 188 git_version = self.git(version=True).NoOp() 189 git_command_version = self.git.version() 190 self.assertEquals(git_version, git_command_version) 191 192 def test_persistent_options(self): 193 git_command_version = self.git.version() 194 # analog to test_options_are_passed_to_git 195 self.git.set_persistent_git_options(version=True) 196 git_version = self.git.NoOp() 197 self.assertEquals(git_version, git_command_version) 198 # subsequent calls keep this option: 199 git_version_2 = self.git.NoOp() 200 self.assertEquals(git_version_2, git_command_version) 201 202 # reset to empty: 203 self.git.set_persistent_git_options() 204 self.assertRaises(GitCommandError, self.git.NoOp) 205 206 def test_single_char_git_options_are_passed_to_git(self): 207 input_value = 'TestValue' 208 output_value = self.git(c='user.name=%s' % input_value).config('--get', 'user.name') 209 self.assertEquals(input_value, output_value) 210 211 def test_change_to_transform_kwargs_does_not_break_command_options(self): 212 self.git.log(n=1) 213 214 def test_insert_after_kwarg_raises(self): 215 # This isn't a complete add command, which doesn't matter here 216 self.failUnlessRaises(ValueError, self.git.remote, 'add', insert_kwargs_after='foo') 217 218 def test_env_vars_passed_to_git(self): 219 editor = 'non_existent_editor' 220 with mock.patch.dict('os.environ', {'GIT_EDITOR': editor}): # @UndefinedVariable 221 self.assertEqual(self.git.var("GIT_EDITOR"), editor) 222 223 @with_rw_directory 224 def test_environment(self, rw_dir): 225 # sanity check 226 self.assertEqual(self.git.environment(), {}) 227 228 # make sure the context manager works and cleans up after itself 229 with self.git.custom_environment(PWD='/tmp'): 230 self.assertEqual(self.git.environment(), {'PWD': '/tmp'}) 231 232 self.assertEqual(self.git.environment(), {}) 233 234 old_env = self.git.update_environment(VARKEY='VARVALUE') 235 # The returned dict can be used to revert the change, hence why it has 236 # an entry with value 'None'. 237 self.assertEqual(old_env, {'VARKEY': None}) 238 self.assertEqual(self.git.environment(), {'VARKEY': 'VARVALUE'}) 239 240 new_env = self.git.update_environment(**old_env) 241 self.assertEqual(new_env, {'VARKEY': 'VARVALUE'}) 242 self.assertEqual(self.git.environment(), {}) 243 244 path = osp.join(rw_dir, 'failing-script.sh') 245 with open(path, 'wt') as stream: 246 stream.write("#!/usr/bin/env sh\n" 247 "echo FOO\n") 248 os.chmod(path, 0o777) 249 250 rw_repo = Repo.init(osp.join(rw_dir, 'repo')) 251 remote = rw_repo.create_remote('ssh-origin', "ssh://git@server/foo") 252 253 with rw_repo.git.custom_environment(GIT_SSH=path): 254 try: 255 remote.fetch() 256 except GitCommandError as err: 257 if sys.version_info[0] < 3 and is_darwin: 258 self.assertIn('ssh-orig, ' in str(err)) 259 self.assertEqual(err.status, 128) 260 else: 261 self.assertIn('FOO', str(err)) 262 263 def test_handle_process_output(self): 264 from git.cmd import handle_process_output 265 266 line_count = 5002 267 count = [None, 0, 0] 268 269 def counter_stdout(line): 270 count[1] += 1 271 272 def counter_stderr(line): 273 count[2] += 1 274 275 cmdline = [sys.executable, fixture_path('cat_file.py'), str(fixture_path('issue-301_stderr'))] 276 proc = subprocess.Popen(cmdline, 277 stdin=None, 278 stdout=subprocess.PIPE, 279 stderr=subprocess.PIPE, 280 shell=False, 281 creationflags=cmd.PROC_CREATIONFLAGS, 282 ) 283 284 handle_process_output(proc, counter_stdout, counter_stderr, finalize_process) 285 286 self.assertEqual(count[1], line_count) 287 self.assertEqual(count[2], line_count) 288