| # Copyright (C) 2019 The Android Open Source Project | |
| # | |
| # Licensed under the Apache License, Version 2.0 (the "License"); | |
| # you may not use this file except in compliance with the License. | |
| # You may obtain a copy of the License at | |
| # | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, software | |
| # distributed under the License is distributed on an "AS IS" BASIS, | |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
| # See the License for the specific language governing permissions and | |
| # limitations under the License. | |
| """Unittests for the git_command.py module.""" | |
| import io | |
| import os | |
| import re | |
| import subprocess | |
| import unittest | |
| from unittest import mock | |
| import pytest | |
| import git_command | |
| import wrapper | |
| class GitCommandTest(unittest.TestCase): | |
| """Tests the GitCommand class (via git_command.git).""" | |
| def setUp(self): | |
| def realpath_mock(val): | |
| return val | |
| mock.patch.object( | |
| os.path, "realpath", side_effect=realpath_mock | |
| ).start() | |
| def tearDown(self): | |
| mock.patch.stopall() | |
| def test_alternative_setting_when_matching(self): | |
| r = git_command._build_env( | |
| objdir=os.path.join("zap", "objects"), gitdir="zap" | |
| ) | |
| self.assertIsNone(r.get("GIT_ALTERNATE_OBJECT_DIRECTORIES")) | |
| self.assertEqual( | |
| r.get("GIT_OBJECT_DIRECTORY"), os.path.join("zap", "objects") | |
| ) | |
| def test_alternative_setting_when_different(self): | |
| r = git_command._build_env( | |
| objdir=os.path.join("wow", "objects"), gitdir="zap" | |
| ) | |
| self.assertEqual( | |
| r.get("GIT_ALTERNATE_OBJECT_DIRECTORIES"), | |
| os.path.join("zap", "objects"), | |
| ) | |
| self.assertEqual( | |
| r.get("GIT_OBJECT_DIRECTORY"), os.path.join("wow", "objects") | |
| ) | |
| class GitCommandWaitTest(unittest.TestCase): | |
| """Tests the GitCommand class .Wait()""" | |
| def setUp(self): | |
| class MockPopen: | |
| rc = 0 | |
| def __init__(self): | |
| self.stdout = io.BufferedReader(io.BytesIO()) | |
| self.stderr = io.BufferedReader(io.BytesIO()) | |
| def communicate( | |
| self, input: str = None, timeout: float = None | |
| ) -> [str, str]: | |
| """Mock communicate fn.""" | |
| return ["", ""] | |
| def wait(self, timeout=None): | |
| return self.rc | |
| self.popen = popen = MockPopen() | |
| def popen_mock(*args, **kwargs): | |
| return popen | |
| def realpath_mock(val): | |
| return val | |
| mock.patch.object(subprocess, "Popen", side_effect=popen_mock).start() | |
| mock.patch.object( | |
| os.path, "realpath", side_effect=realpath_mock | |
| ).start() | |
| def tearDown(self): | |
| mock.patch.stopall() | |
| def test_raises_when_verify_non_zero_result(self): | |
| self.popen.rc = 1 | |
| r = git_command.GitCommand(None, ["status"], verify_command=True) | |
| with self.assertRaises(git_command.GitCommandError): | |
| r.Wait() | |
| def test_returns_when_no_verify_non_zero_result(self): | |
| self.popen.rc = 1 | |
| r = git_command.GitCommand(None, ["status"], verify_command=False) | |
| self.assertEqual(1, r.Wait()) | |
| def test_default_returns_non_zero_result(self): | |
| self.popen.rc = 1 | |
| r = git_command.GitCommand(None, ["status"]) | |
| self.assertEqual(1, r.Wait()) | |
| class GitCommandStreamLogsTest(unittest.TestCase): | |
| """Tests the GitCommand class stderr log streaming cases.""" | |
| def setUp(self): | |
| _ = git_command.user_agent.git | |
| self.mock_process = mock.MagicMock() | |
| self.mock_process.communicate.return_value = (None, None) | |
| self.mock_process.wait.return_value = 0 | |
| self.mock_popen = mock.MagicMock() | |
| self.mock_popen.return_value = self.mock_process | |
| mock.patch("subprocess.Popen", self.mock_popen).start() | |
| def tearDown(self): | |
| mock.patch.stopall() | |
| def test_does_not_stream_logs_when_input_is_set(self): | |
| git_command.GitCommand(None, ["status"], input="foo") | |
| self.mock_popen.assert_called_once_with( | |
| ["git", "status"], | |
| cwd=None, | |
| env=mock.ANY, | |
| encoding="utf-8", | |
| errors="backslashreplace", | |
| stdin=subprocess.PIPE, | |
| stdout=None, | |
| stderr=None, | |
| ) | |
| self.mock_process.communicate.assert_called_once_with(input="foo") | |
| self.mock_process.stderr.read1.assert_not_called() | |
| def test_does_not_stream_logs_when_stdout_is_set(self): | |
| git_command.GitCommand(None, ["status"], capture_stdout=True) | |
| self.mock_popen.assert_called_once_with( | |
| ["git", "status"], | |
| cwd=None, | |
| env=mock.ANY, | |
| encoding="utf-8", | |
| errors="backslashreplace", | |
| stdin=None, | |
| stdout=subprocess.PIPE, | |
| stderr=None, | |
| ) | |
| self.mock_process.communicate.assert_called_once_with(input=None) | |
| self.mock_process.stderr.read1.assert_not_called() | |
| def test_does_not_stream_logs_when_stderr_is_set(self): | |
| git_command.GitCommand(None, ["status"], capture_stderr=True) | |
| self.mock_popen.assert_called_once_with( | |
| ["git", "status"], | |
| cwd=None, | |
| env=mock.ANY, | |
| encoding="utf-8", | |
| errors="backslashreplace", | |
| stdin=None, | |
| stdout=None, | |
| stderr=subprocess.PIPE, | |
| ) | |
| self.mock_process.communicate.assert_called_once_with(input=None) | |
| self.mock_process.stderr.read1.assert_not_called() | |
| def test_does_not_stream_logs_when_merge_output_is_set(self): | |
| git_command.GitCommand(None, ["status"], merge_output=True) | |
| self.mock_popen.assert_called_once_with( | |
| ["git", "status"], | |
| cwd=None, | |
| env=mock.ANY, | |
| encoding="utf-8", | |
| errors="backslashreplace", | |
| stdin=None, | |
| stdout=None, | |
| stderr=subprocess.STDOUT, | |
| ) | |
| self.mock_process.communicate.assert_called_once_with(input=None) | |
| self.mock_process.stderr.read1.assert_not_called() | |
| @mock.patch("sys.stderr") | |
| def test_streams_stderr_when_no_stream_is_set(self, mock_stderr): | |
| mock_stderr.isatty.return_value = False | |
| logs = "\n".join( | |
| [ | |
| "Enumerating objects: 5, done.", | |
| "Counting objects: 100% (5/5), done.", | |
| "Writing objects: 100% (3/3), 330 bytes | 330 KiB/s, done.", | |
| "remote: Processing changes: refs: 1, new: 1, done ", | |
| "remote: SUCCESS", | |
| ] | |
| ) | |
| self.mock_process.stderr = io.BufferedReader( | |
| io.BytesIO(bytes(logs, "utf-8")) | |
| ) | |
| cmd = git_command.GitCommand(None, ["push"]) | |
| self.mock_popen.assert_called_once_with( | |
| ["git", "push"], | |
| cwd=None, | |
| env=mock.ANY, | |
| stdin=None, | |
| stdout=None, | |
| stderr=subprocess.PIPE, | |
| ) | |
| self.mock_process.communicate.assert_not_called() | |
| mock_stderr.write.assert_called_once_with(logs) | |
| self.assertEqual(cmd.stderr, logs) | |
| @mock.patch("sys.stderr") | |
| def test_push_injects_progress_only_when_tty_and_not_suppressed( | |
| self, mock_stderr: mock.MagicMock | |
| ) -> None: | |
| mock_stderr.isatty.return_value = True | |
| self.mock_process.stderr = io.BufferedReader(io.BytesIO(b"")) | |
| git_command.GitCommand(None, ["push"]) | |
| self.mock_popen.assert_called_once() | |
| cmd = self.mock_popen.call_args[0][0] | |
| self.assertEqual(cmd, ["git", "push", "--progress"]) | |
| @mock.patch("sys.stderr") | |
| def test_push_does_not_inject_progress_when_no_progress_set( | |
| self, mock_stderr: mock.MagicMock | |
| ) -> None: | |
| mock_stderr.isatty.return_value = True | |
| self.mock_process.stderr = io.BufferedReader(io.BytesIO(b"")) | |
| git_command.GitCommand(None, ["push", "--no-progress"]) | |
| self.mock_popen.assert_called_once() | |
| cmd = self.mock_popen.call_args[0][0] | |
| self.assertEqual(cmd, ["git", "push", "--no-progress"]) | |
| class GitCommandCaptureBytesTest(unittest.TestCase): | |
| """Tests the GitCommand class byte capture cases.""" | |
| def setUp(self) -> None: | |
| _ = git_command.user_agent.git | |
| self.mock_process = mock.MagicMock() | |
| self.mock_process.communicate.return_value = (None, None) | |
| self.mock_process.wait.return_value = 0 | |
| self.mock_popen = mock.MagicMock() | |
| self.mock_popen.return_value = self.mock_process | |
| mock.patch("subprocess.Popen", self.mock_popen).start() | |
| def tearDown(self) -> None: | |
| mock.patch.stopall() | |
| def test_captures_stdout_as_bytes(self) -> None: | |
| self.mock_process.communicate.return_value = (b"\xff\x00", b"error\r\n") | |
| cmd = git_command.GitCommand( | |
| None, | |
| ["status"], | |
| capture_stdout=True, | |
| capture_stdout_bytes=True, | |
| capture_stderr=True, | |
| ) | |
| self.mock_popen.assert_called_once_with( | |
| ["git", "status"], | |
| cwd=None, | |
| env=mock.ANY, | |
| stdin=None, | |
| stdout=subprocess.PIPE, | |
| stderr=subprocess.PIPE, | |
| ) | |
| self.assertEqual(cmd.stdout, b"\xff\x00") | |
| self.assertEqual(cmd.stderr, "error\n") | |
| def test_capture_stdout_bytes_auto_enables_capture_stdout(self) -> None: | |
| self.mock_process.communicate.return_value = (b"output", b"") | |
| cmd = git_command.GitCommand( | |
| None, | |
| ["status"], | |
| capture_stdout_bytes=True, | |
| ) | |
| self.mock_popen.assert_called_once_with( | |
| ["git", "status"], | |
| cwd=None, | |
| env=mock.ANY, | |
| stdin=None, | |
| stdout=subprocess.PIPE, | |
| stderr=None, | |
| ) | |
| self.assertEqual(cmd.stdout, b"output") | |
| def test_capture_stdout_bytes_with_merge_output_raises(self) -> None: | |
| with self.assertRaises(ValueError): | |
| git_command.GitCommand( | |
| None, | |
| ["status"], | |
| capture_stdout_bytes=True, | |
| merge_output=True, | |
| ) | |
| def test_captures_stdout_as_bytes_encodes_str_input(self) -> None: | |
| self.mock_process.communicate.return_value = (b"output", b"") | |
| git_command.GitCommand( | |
| None, | |
| ["status"], | |
| input="hello world", | |
| capture_stdout_bytes=True, | |
| ) | |
| self.mock_process.communicate.assert_called_once_with( | |
| input=b"hello world" | |
| ) | |
| def test_captures_stdout_as_bytes_encodes_surrogate_input(self) -> None: | |
| self.mock_process.communicate.return_value = (b"output", b"") | |
| git_command.GitCommand( | |
| None, | |
| ["status"], | |
| input="file_\udcff.txt", | |
| capture_stdout_bytes=True, | |
| ) | |
| self.mock_process.communicate.assert_called_once_with( | |
| input=b"file_\xff.txt" | |
| ) | |
| def test_captures_stdout_as_bytes_passes_bytes_input(self) -> None: | |
| self.mock_process.communicate.return_value = (b"output", b"") | |
| git_command.GitCommand( | |
| None, | |
| ["status"], | |
| input=b"raw_\xff.txt", | |
| capture_stdout_bytes=True, | |
| ) | |
| self.mock_process.communicate.assert_called_once_with( | |
| input=b"raw_\xff.txt" | |
| ) | |
| def test_verify_command_truncates_nul_delimited_stdout(self) -> None: | |
| cmd = git_command.GitCommand( | |
| None, | |
| ["status"], | |
| capture_stdout_bytes=True, | |
| ) | |
| cmd.rc = 1 | |
| cmd.stdout = b"first_file\0second_file\0third_file" | |
| cmd.stderr = "stderr" | |
| with self.assertRaises(git_command.GitCommandError) as cm: | |
| cmd.VerifyCommand() | |
| self.assertEqual(cm.exception.git_stdout, "first_file") | |
| def test_verify_command_decodes_bytes_stdout(self) -> None: | |
| cmd = git_command.GitCommand( | |
| None, | |
| ["status"], | |
| capture_stdout_bytes=True, | |
| ) | |
| cmd.rc = 1 | |
| cmd.stdout = b"error\xff\nline2" | |
| cmd.stderr = "stderr" | |
| with self.assertRaises(git_command.GitCommandError) as cm: | |
| cmd.VerifyCommand() | |
| self.assertEqual(cm.exception.git_stdout, "error\\xff") | |
| class GitCallUnitTest(unittest.TestCase): | |
| """Tests the _GitCall class (via git_command.git).""" | |
| def test_valid_branch_name_uses_branch_mode(self) -> None: | |
| """Branch validation applies Git's branch-specific restrictions.""" | |
| command = mock.MagicMock() | |
| command.Wait.return_value = 1 | |
| with mock.patch.object( | |
| git_command, "GitCommand", return_value=command | |
| ) as check: | |
| self.assertFalse(git_command.IsValidBranchName("-topic")) | |
| check.assert_called_once_with( | |
| None, | |
| ["check-ref-format", "--branch", "-topic"], | |
| capture_stdout=True, | |
| capture_stderr=True, | |
| add_event_log=False, | |
| log_as_error=False, | |
| ) | |
| def test_version_tuple(self): | |
| """Check git.version_tuple() handling.""" | |
| ver = git_command.git.version_tuple() | |
| self.assertIsNotNone(ver) | |
| # We don't dive too deep into the values here to avoid having to update | |
| # whenever git versions change. We do check relative to this min | |
| # version as this is what `repo` itself requires via MIN_GIT_VERSION. | |
| MIN_GIT_VERSION = (2, 10, 2) | |
| self.assertTrue(isinstance(ver.major, int)) | |
| self.assertTrue(isinstance(ver.minor, int)) | |
| self.assertTrue(isinstance(ver.micro, int)) | |
| self.assertGreater(ver.major, MIN_GIT_VERSION[0] - 1) | |
| self.assertGreaterEqual(ver.micro, 0) | |
| self.assertGreaterEqual(ver.major, 0) | |
| self.assertGreaterEqual(ver, MIN_GIT_VERSION) | |
| self.assertLess(ver, (9999, 9999, 9999)) | |
| self.assertNotEqual("", ver.full) | |
| class UserAgentUnitTest(unittest.TestCase): | |
| """Tests the UserAgent function.""" | |
| def test_smoke_os(self): | |
| """Make sure UA OS setting returns something useful.""" | |
| os_name = git_command.user_agent.os | |
| # We can't dive too deep because of OS/tool differences, but we can | |
| # check the general form. | |
| m = re.match(r"^[^ ]+$", os_name) | |
| self.assertIsNotNone(m) | |
| @pytest.mark.skip_cq("TODO(b/266734831): Find out why this fails in CQ") | |
| def test_smoke_repo(self): | |
| """Make sure repo UA returns something useful.""" | |
| ua = git_command.user_agent.repo | |
| # We can't dive too deep because of OS/tool differences, but we can | |
| # check the general form. | |
| m = re.match(r"^git-repo/[^ ]+ ([^ ]+) git/[^ ]+ Python/[0-9.]+", ua) | |
| self.assertIsNotNone(m) | |
| @pytest.mark.skip_cq("TODO(b/266734831): Find out why this fails in CQ") | |
| def test_smoke_git(self): | |
| """Make sure git UA returns something useful.""" | |
| ua = git_command.user_agent.git | |
| # We can't dive too deep because of OS/tool differences, but we can | |
| # check the general form. | |
| m = re.match(r"^git/[^ ]+ ([^ ]+) git-repo/[^ ]+", ua) | |
| self.assertIsNotNone(m) | |
| class GitRequireTests(unittest.TestCase): | |
| """Test the git_require helper.""" | |
| def setUp(self): | |
| self.wrapper = wrapper.Wrapper() | |
| ver = self.wrapper.GitVersion(1, 2, 3, 4) | |
| mock.patch.object( | |
| git_command.git, "version_tuple", return_value=ver | |
| ).start() | |
| def tearDown(self): | |
| mock.patch.stopall() | |
| def test_older_nonfatal(self): | |
| """Test non-fatal require calls with old versions.""" | |
| self.assertFalse(git_command.git_require((2,))) | |
| self.assertFalse(git_command.git_require((1, 3))) | |
| self.assertFalse(git_command.git_require((1, 2, 4))) | |
| self.assertFalse(git_command.git_require((1, 2, 3, 5))) | |
| def test_newer_nonfatal(self): | |
| """Test non-fatal require calls with newer versions.""" | |
| self.assertTrue(git_command.git_require((0,))) | |
| self.assertTrue(git_command.git_require((1, 0))) | |
| self.assertTrue(git_command.git_require((1, 2, 0))) | |
| self.assertTrue(git_command.git_require((1, 2, 3, 0))) | |
| def test_equal_nonfatal(self): | |
| """Test require calls with equal values.""" | |
| self.assertTrue(git_command.git_require((1, 2, 3, 4), fail=False)) | |
| self.assertTrue(git_command.git_require((1, 2, 3, 4), fail=True)) | |
| def test_older_fatal(self): | |
| """Test fatal require calls with old versions.""" | |
| with self.assertRaises(git_command.GitRequireError) as e: | |
| git_command.git_require((2,), fail=True) | |
| self.assertNotEqual(0, e.code) | |
| def test_older_fatal_msg(self): | |
| """Test fatal require calls with old versions and message.""" | |
| with self.assertRaises(git_command.GitRequireError) as e: | |
| git_command.git_require((2,), fail=True, msg="so sad") | |
| self.assertNotEqual(0, e.code) | |
| class GitCommandErrorTest(unittest.TestCase): | |
| """Test for the GitCommandError class.""" | |
| def test_augument_stderr(self): | |
| self.assertEqual( | |
| git_command.GitCommandError( | |
| git_stderr="couldn't find remote ref refs/heads/foo" | |
| ).suggestion, | |
| "Check if the provided ref exists in the remote.", | |
| ) | |
| self.assertEqual( | |
| git_command.GitCommandError( | |
| git_stderr="'foobar' does not appear to be a git repository" | |
| ).suggestion, | |
| "Are you running this repo command outside of a repo workspace?", | |
| ) |