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