tests/unit/test_base_command.py - main - niceSOFT/pip

tests/unit/test_base_command.py
278 строк9 KB
Islam Asabaev
Prevent sub-commands from accepting --version (#14231)
09 авг 2026, 00:26
09 авг 2026, 00:26f399c37
from __future__ import annotations

import importlib
import logging
import os
import time
from collections.abc import Callable, Iterator
from optparse import Values
from pathlib import Path
from typing import NoReturn
from unittest.mock import Mock, patch

import pytest

from pip._internal.cli import base_command
from pip._internal.cli.base_command import Command
from pip._internal.cli.status_codes import (
    BROKEN_STDOUT,
    SUCCESS,
    UNKNOWN_ERROR,
    VIRTUALENV_NOT_FOUND,
)
from pip._internal.commands import commands_dict, create_command
from pip._internal.utils import temp_dir
from pip._internal.utils.logging import BrokenStdoutLoggingError
from pip._internal.utils.temp_dir import TempDirectory


@pytest.fixture
def fixed_time() -> Iterator[None]:
    # Patch time so logs contain a constant timestamp. time.time_ns is used by
    # logging starting with Python 3.13.
    year2019 = 1547704837.040001
    with (
        patch("time.time", lambda: year2019),
        patch("time.time_ns", lambda: int(year2019 * 1e9)),
        patch.object(logging.Formatter, "converter", time.gmtime),
    ):
        yield


class FakeCommand(Command):
    _name = "fake"

    def __init__(
        self, run_func: Callable[[], int] | None = None, error: bool = False
    ) -> None:
        if error:

            def run_func() -> int:
                raise SystemExit(1)

        self.run_func = run_func
        super().__init__(self._name, self._name)

    def main(self, args: list[str]) -> int:
        args.append("--disable-pip-version-check")
        return super().main(args)

    def run(self, options: Values, args: list[str]) -> int:
        logging.getLogger("pip.tests").info("fake")
        # Return SUCCESS from run if run_func is not provided
        if self.run_func:
            return self.run_func()
        else:
            return SUCCESS


class FakeCommandWithUnicode(FakeCommand):
    _name = "fake_unicode"

    def run(self, options: Values, args: list[str]) -> int:
        logging.getLogger("pip.tests").info(b"bytes here \xe9")
        logging.getLogger("pip.tests").info(b"unicode here \xc3\xa9".decode("utf-8"))
        return SUCCESS


class TestCommand:
    def call_main(self, capfd: pytest.CaptureFixture[str], args: list[str]) -> str:
        """
        Call command.main(), and return the command's stderr.
        """

        def raise_broken_stdout() -> NoReturn:
            raise BrokenStdoutLoggingError()

        cmd = FakeCommand(run_func=raise_broken_stdout)
        status = cmd.main(args)
        assert status == BROKEN_STDOUT
        stderr = capfd.readouterr().err

        return stderr

    def test_raise_broken_stdout(self, capfd: pytest.CaptureFixture[str]) -> None:
        """
        Test raising BrokenStdoutLoggingError.
        """
        stderr = self.call_main(capfd, [])

        assert stderr.rstrip() == "ERROR: Pipe to stdout was broken"
        assert "Exception ignored on flushing sys.stdout:" not in stderr

    def test_raise_broken_stdout__debug_logging(
        self, capfd: pytest.CaptureFixture[str]
    ) -> None:
        """
        Test raising BrokenStdoutLoggingError with debug logging enabled.
        """
        stderr = self.call_main(capfd, ["-vv"])

        assert "ERROR: Pipe to stdout was broken" in stderr
        assert "Traceback (most recent call last):" in stderr
        assert "Exception ignored on flushing sys.stdout:" not in stderr


@patch("pip._internal.cli.index_command.Command.pip_version_check")
def test_pip_version_check_called(mock_version_check: Mock) -> None:
    """
    Check that ``Command.pip_version_check()`` wraps the command body.
    """
    cmd = FakeCommand()
    cmd.main([])
    mock_version_check.assert_called_once()


def test_debug_enables_verbose_logs() -> None:
    cmd = FakeCommand()
    cmd.main(["fake", "--debug"])
    assert cmd.verbosity >= 2


def test_log_command_success(fixed_time: None, tmpdir: Path) -> None:
    """Test the --log option logs when command succeeds."""
    cmd = FakeCommand()
    log_path = os.path.join(tmpdir, "log")
    cmd.main(["fake", "--log", log_path])
    with open(log_path) as f:
        assert f.read().rstrip() == "2019-01-17T06:00:37,040 fake"


def test_log_command_error(fixed_time: None, tmpdir: Path) -> None:
    """Test the --log option logs when command fails."""
    cmd = FakeCommand(error=True)
    log_path = os.path.join(tmpdir, "log")
    cmd.main(["fake", "--log", log_path])
    with open(log_path) as f:
        assert f.read().startswith("2019-01-17T06:00:37,040 fake")


def test_log_file_command_error(fixed_time: None, tmpdir: Path) -> None:
    """Test the --log-file option logs (when there's an error)."""
    cmd = FakeCommand(error=True)
    log_file_path = os.path.join(tmpdir, "log_file")
    cmd.main(["fake", "--log-file", log_file_path])
    with open(log_file_path) as f:
        assert f.read().startswith("2019-01-17T06:00:37,040 fake")


def test_log_unicode_messages(fixed_time: None, tmpdir: Path) -> None:
    """Tests that logging bytestrings and unicode objects
    don't break logging.
    """
    cmd = FakeCommandWithUnicode()
    log_path = os.path.join(tmpdir, "log")
    cmd.main(["fake_unicode", "--log", log_path])


@pytest.mark.no_auto_tempdir_manager
def test_base_command_provides_tempdir_helpers() -> None:
    assert temp_dir._tempdir_manager is None
    assert temp_dir._tempdir_registry is None

    def assert_helpers_set(options: Values, args: list[str]) -> int:
        assert temp_dir._tempdir_manager is not None
        assert temp_dir._tempdir_registry is not None
        return SUCCESS

    c = Command("fake", "fake")
    # https://github.com/python/mypy/issues/2427
    c.run = Mock(side_effect=assert_helpers_set)  # type: ignore[method-assign]
    assert c.main(["fake"]) == SUCCESS
    c.run.assert_called_once()


not_deleted = "not_deleted"


@pytest.mark.parametrize("kind,exists", [(not_deleted, True), ("deleted", False)])
@pytest.mark.no_auto_tempdir_manager
def test_base_command_global_tempdir_cleanup(kind: str, exists: bool) -> None:
    assert temp_dir._tempdir_manager is None
    assert temp_dir._tempdir_registry is None

    class Holder:
        value: str

    def create_temp_dirs(options: Values, args: list[str]) -> int:
        assert c.tempdir_registry is not None
        c.tempdir_registry.set_delete(not_deleted, False)
        Holder.value = TempDirectory(kind=kind, globally_managed=True).path
        return SUCCESS

    c = Command("fake", "fake")
    # https://github.com/python/mypy/issues/2427
    c.run = Mock(side_effect=create_temp_dirs)  # type: ignore[method-assign]
    assert c.main(["fake"]) == SUCCESS
    c.run.assert_called_once()
    assert os.path.exists(Holder.value) == exists


@pytest.mark.parametrize("kind,exists", [(not_deleted, True), ("deleted", False)])
@pytest.mark.no_auto_tempdir_manager
def test_base_command_local_tempdir_cleanup(kind: str, exists: bool) -> None:
    assert temp_dir._tempdir_manager is None
    assert temp_dir._tempdir_registry is None

    def create_temp_dirs(options: Values, args: list[str]) -> int:
        assert c.tempdir_registry is not None
        c.tempdir_registry.set_delete(not_deleted, False)

        with TempDirectory(kind=kind) as d:
            path = d.path
            assert os.path.exists(path)
        assert os.path.exists(path) == exists
        return SUCCESS

    c = Command("fake", "fake")
    # https://github.com/python/mypy/issues/2427
    c.run = Mock(side_effect=create_temp_dirs)  # type: ignore[method-assign]
    assert c.main(["fake"]) == SUCCESS
    c.run.assert_called_once()


def test_require_virtualenv_exits_when_not_in_venv(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(base_command, "running_under_virtualenv", lambda: False)

    cmd = create_command("install")
    with pytest.raises(SystemExit) as excinfo:
        cmd.main(["--require-virtualenv", "pip"])

    assert excinfo.value.code == VIRTUALENV_NOT_FOUND


def test_require_virtualenv_is_ignored_by_opt_out_commands(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(base_command, "running_under_virtualenv", lambda: False)

    cmd = create_command("help")
    assert cmd.main(["--require-virtualenv"]) == SUCCESS


def test_commands_ignore_require_virtualenv_is_explicit() -> None:
    commands_that_require_venv = ["download", "install", "lock", "uninstall", "wheel"]

    for name, info in commands_dict.items():
        module = importlib.import_module(info.module_path)
        command_class = getattr(module, info.class_name)
        assert not command_class.ignore_require_venv == (
            name in commands_that_require_venv
        )


@pytest.mark.parametrize("command_name", commands_dict)
def test_commands_reject_version_option(
    command_name: str,
    capfd: pytest.CaptureFixture[str],
) -> None:
    cmd = create_command(command_name)

    with pytest.raises(SystemExit) as excinfo:
        cmd.main(["--version"])
    stderr = capfd.readouterr().err

    assert excinfo.value.code == UNKNOWN_ERROR
    assert "no such option: --version" in stderr