test_utils.py 56.7 KB
Newer Older
1 2
# -*- coding: utf-8 -*-

3 4 5 6
"""
util tests

"""
7
import codecs
8
import itertools
9
import locale
10
import os
P
Pradyun S. Gedam 已提交
11
import shutil
12
import stat
13
import sys
14
import tempfile
P
Pradyun S. Gedam 已提交
15
import time
16
import warnings
17
from io import BytesIO
18
from logging import DEBUG, ERROR, INFO, WARNING
19
from textwrap import dedent
20

P
Pradyun S. Gedam 已提交
21
import pytest
22
from mock import Mock, patch
23
from pip._vendor.six.moves.urllib import request as urllib_request
24

25
from pip._internal.exceptions import (
26 27 28
    HashMismatch,
    HashMissing,
    InstallationError,
P
Pradyun S. Gedam 已提交
29
)
30
from pip._internal.utils.deprecation import PipDeprecationWarning, deprecated
31
from pip._internal.utils.encoding import BOMS, auto_decode
32
from pip._internal.utils.glibc import (
33 34 35
    check_glibc_version,
    glibc_version_string,
    glibc_version_string_confstr,
36 37
    glibc_version_string_ctypes,
)
38 39
from pip._internal.utils.hashes import Hashes, MissingHashes
from pip._internal.utils.misc import (
40
    HiddenText,
41
    build_netloc,
42
    build_url_from_netloc,
43 44 45 46 47 48
    call_subprocess,
    egg_link_path,
    ensure_dir,
    format_command_args,
    get_installed_distributions,
    get_prog,
49 50 51
    hide_url,
    hide_value,
    make_command,
52 53 54
    make_subprocess_output_error,
    normalize_path,
    normalize_version_info,
55
    parse_netloc,
56 57
    path_to_display,
    path_to_url,
58
    redact_auth_from_url,
59 60 61 62 63 64 65
    redact_netloc,
    remove_auth_from_url,
    rmtree,
    split_auth_from_netloc,
    split_auth_netloc_from_url,
    untar_file,
    unzip_file,
P
Pradyun S. Gedam 已提交
66
)
67
from pip._internal.utils.setuptools_build import make_setuptools_shim_args
S
Steve Dower 已提交
68
from pip._internal.utils.temp_dir import AdjacentTempDirectory, TempDirectory
69
from pip._internal.utils.ui import SpinnerInterface
70 71 72 73 74 75 76 77 78 79 80 81


class Tests_EgglinkPath:
    "util.egg_link_path() tests"

    def setup(self):

        project = 'foo'

        self.mock_dist = Mock(project_name=project)
        self.site_packages = 'SITE_PACKAGES'
        self.user_site = 'USER_SITE'
82 83 84 85 86 87 88 89
        self.user_site_egglink = os.path.join(
            self.user_site,
            '%s.egg-link' % project
        )
        self.site_packages_egglink = os.path.join(
            self.site_packages,
            '%s.egg-link' % project,
        )
90

91
        # patches
92
        from pip._internal.utils import misc as utils
93 94 95 96
        self.old_site_packages = utils.site_packages
        self.mock_site_packages = utils.site_packages = 'SITE_PACKAGES'
        self.old_running_under_virtualenv = utils.running_under_virtualenv
        self.mock_running_under_virtualenv = utils.running_under_virtualenv = \
97
            Mock()
98 99 100 101
        self.old_virtualenv_no_global = utils.virtualenv_no_global
        self.mock_virtualenv_no_global = utils.virtualenv_no_global = Mock()
        self.old_user_site = utils.user_site
        self.mock_user_site = utils.user_site = self.user_site
102
        from os import path
103
        self.old_isfile = path.isfile
104 105
        self.mock_isfile = path.isfile = Mock()

106
    def teardown(self):
107
        from pip._internal.utils import misc as utils
108 109 110 111
        utils.site_packages = self.old_site_packages
        utils.running_under_virtualenv = self.old_running_under_virtualenv
        utils.virtualenv_no_global = self.old_virtualenv_no_global
        utils.user_site = self.old_user_site
112 113 114
        from os import path
        path.isfile = self.old_isfile

115 116
    def eggLinkInUserSite(self, egglink):
        return egglink == self.user_site_egglink
117

118 119
    def eggLinkInSitePackages(self, egglink):
        return egglink == self.site_packages_egglink
120

121 122 123
    # ####################### #
    # # egglink in usersite # #
    # ####################### #
124 125 126 127
    def test_egglink_in_usersite_notvenv(self):
        self.mock_virtualenv_no_global.return_value = False
        self.mock_running_under_virtualenv.return_value = False
        self.mock_isfile.side_effect = self.eggLinkInUserSite
128
        assert egg_link_path(self.mock_dist) == self.user_site_egglink
129 130 131 132 133

    def test_egglink_in_usersite_venv_noglobal(self):
        self.mock_virtualenv_no_global.return_value = True
        self.mock_running_under_virtualenv.return_value = True
        self.mock_isfile.side_effect = self.eggLinkInUserSite
134
        assert egg_link_path(self.mock_dist) is None
135 136 137 138 139

    def test_egglink_in_usersite_venv_global(self):
        self.mock_virtualenv_no_global.return_value = False
        self.mock_running_under_virtualenv.return_value = True
        self.mock_isfile.side_effect = self.eggLinkInUserSite
140
        assert egg_link_path(self.mock_dist) == self.user_site_egglink
141

142 143 144
    # ####################### #
    # # egglink in sitepkgs # #
    # ####################### #
145 146 147 148
    def test_egglink_in_sitepkgs_notvenv(self):
        self.mock_virtualenv_no_global.return_value = False
        self.mock_running_under_virtualenv.return_value = False
        self.mock_isfile.side_effect = self.eggLinkInSitePackages
149
        assert egg_link_path(self.mock_dist) == self.site_packages_egglink
150 151 152 153 154

    def test_egglink_in_sitepkgs_venv_noglobal(self):
        self.mock_virtualenv_no_global.return_value = True
        self.mock_running_under_virtualenv.return_value = True
        self.mock_isfile.side_effect = self.eggLinkInSitePackages
155
        assert egg_link_path(self.mock_dist) == self.site_packages_egglink
156 157 158 159 160

    def test_egglink_in_sitepkgs_venv_global(self):
        self.mock_virtualenv_no_global.return_value = False
        self.mock_running_under_virtualenv.return_value = True
        self.mock_isfile.side_effect = self.eggLinkInSitePackages
161
        assert egg_link_path(self.mock_dist) == self.site_packages_egglink
162

163 164 165
    # ################################## #
    # # egglink in usersite & sitepkgs # #
    # ################################## #
166 167 168 169
    def test_egglink_in_both_notvenv(self):
        self.mock_virtualenv_no_global.return_value = False
        self.mock_running_under_virtualenv.return_value = False
        self.mock_isfile.return_value = True
170
        assert egg_link_path(self.mock_dist) == self.user_site_egglink
171 172 173 174 175

    def test_egglink_in_both_venv_noglobal(self):
        self.mock_virtualenv_no_global.return_value = True
        self.mock_running_under_virtualenv.return_value = True
        self.mock_isfile.return_value = True
176
        assert egg_link_path(self.mock_dist) == self.site_packages_egglink
177 178 179 180 181

    def test_egglink_in_both_venv_global(self):
        self.mock_virtualenv_no_global.return_value = False
        self.mock_running_under_virtualenv.return_value = True
        self.mock_isfile.return_value = True
182
        assert egg_link_path(self.mock_dist) == self.site_packages_egglink
183

184 185 186
    # ############## #
    # # no egglink # #
    # ############## #
187 188 189 190
    def test_noegglink_in_sitepkgs_notvenv(self):
        self.mock_virtualenv_no_global.return_value = False
        self.mock_running_under_virtualenv.return_value = False
        self.mock_isfile.return_value = False
191
        assert egg_link_path(self.mock_dist) is None
192 193 194 195 196

    def test_noegglink_in_sitepkgs_venv_noglobal(self):
        self.mock_virtualenv_no_global.return_value = True
        self.mock_running_under_virtualenv.return_value = True
        self.mock_isfile.return_value = False
197
        assert egg_link_path(self.mock_dist) is None
198 199 200 201 202

    def test_noegglink_in_sitepkgs_venv_global(self):
        self.mock_virtualenv_no_global.return_value = False
        self.mock_running_under_virtualenv.return_value = True
        self.mock_isfile.return_value = False
203
        assert egg_link_path(self.mock_dist) is None
204

205

206 207 208
@patch('pip._internal.utils.misc.dist_in_usersite')
@patch('pip._internal.utils.misc.dist_is_local')
@patch('pip._internal.utils.misc.dist_is_editable')
209 210 211 212
class Tests_get_installed_distributions:
    """test util.get_installed_distributions"""

    workingset = [
213 214
        Mock(test_name="global"),
        Mock(test_name="editable"),
215
        Mock(test_name="normal"),
216
        Mock(test_name="user"),
217 218 219 220 221 222 223 224 225 226 227
    ]

    workingset_stdlib = [
        Mock(test_name='normal', key='argparse'),
        Mock(test_name='normal', key='wsgiref')
    ]

    workingset_freeze = [
        Mock(test_name='normal', key='pip'),
        Mock(test_name='normal', key='setuptools'),
        Mock(test_name='normal', key='distribute')
228
    ]
229 230 231 232 233

    def dist_is_editable(self, dist):
        return dist.test_name == "editable"

    def dist_is_local(self, dist):
234 235 236 237
        return dist.test_name != "global" and dist.test_name != 'user'

    def dist_in_usersite(self, dist):
        return dist.test_name == "user"
238

239
    @patch('pip._vendor.pkg_resources.working_set', workingset)
240 241 242
    def test_editables_only(self, mock_dist_is_editable,
                            mock_dist_is_local,
                            mock_dist_in_usersite):
243 244
        mock_dist_is_editable.side_effect = self.dist_is_editable
        mock_dist_is_local.side_effect = self.dist_is_local
245
        mock_dist_in_usersite.side_effect = self.dist_in_usersite
246 247 248 249
        dists = get_installed_distributions(editables_only=True)
        assert len(dists) == 1, dists
        assert dists[0].test_name == "editable"

250
    @patch('pip._vendor.pkg_resources.working_set', workingset)
251 252 253
    def test_exclude_editables(self, mock_dist_is_editable,
                               mock_dist_is_local,
                               mock_dist_in_usersite):
254 255
        mock_dist_is_editable.side_effect = self.dist_is_editable
        mock_dist_is_local.side_effect = self.dist_is_local
256
        mock_dist_in_usersite.side_effect = self.dist_in_usersite
257 258 259 260
        dists = get_installed_distributions(include_editables=False)
        assert len(dists) == 1
        assert dists[0].test_name == "normal"

261
    @patch('pip._vendor.pkg_resources.working_set', workingset)
262 263 264
    def test_include_globals(self, mock_dist_is_editable,
                             mock_dist_is_local,
                             mock_dist_in_usersite):
265 266
        mock_dist_is_editable.side_effect = self.dist_is_editable
        mock_dist_is_local.side_effect = self.dist_is_local
267
        mock_dist_in_usersite.side_effect = self.dist_in_usersite
268
        dists = get_installed_distributions(local_only=False)
269 270 271 272 273 274 275 276 277 278 279 280 281
        assert len(dists) == 4

    @patch('pip._vendor.pkg_resources.working_set', workingset)
    def test_user_only(self, mock_dist_is_editable,
                       mock_dist_is_local,
                       mock_dist_in_usersite):
        mock_dist_is_editable.side_effect = self.dist_is_editable
        mock_dist_is_local.side_effect = self.dist_is_local
        mock_dist_in_usersite.side_effect = self.dist_in_usersite
        dists = get_installed_distributions(local_only=False,
                                            user_only=True)
        assert len(dists) == 1
        assert dists[0].test_name == "user"
282

283 284
    @patch('pip._vendor.pkg_resources.working_set', workingset_stdlib)
    def test_gte_py27_excludes(self, mock_dist_is_editable,
285 286
                               mock_dist_is_local,
                               mock_dist_in_usersite):
287 288
        mock_dist_is_editable.side_effect = self.dist_is_editable
        mock_dist_is_local.side_effect = self.dist_is_local
289
        mock_dist_in_usersite.side_effect = self.dist_in_usersite
290 291 292 293
        dists = get_installed_distributions()
        assert len(dists) == 0

    @patch('pip._vendor.pkg_resources.working_set', workingset_freeze)
294 295 296
    def test_freeze_excludes(self, mock_dist_is_editable,
                             mock_dist_is_local,
                             mock_dist_in_usersite):
297 298
        mock_dist_is_editable.side_effect = self.dist_is_editable
        mock_dist_is_local.side_effect = self.dist_is_local
299
        mock_dist_in_usersite.side_effect = self.dist_in_usersite
300 301
        dists = get_installed_distributions(
            skip=('setuptools', 'pip', 'distribute'))
302 303
        assert len(dists) == 0

304

305 306
class TestUnpackArchives(object):
    """
307 308
    test_tar.tgz/test_tar.zip have content as follows engineered to confirm 3
    things:
309 310 311 312 313 314 315 316 317 318 319
     1) confirm that reg files, dirs, and symlinks get unpacked
     2) permissions are not preserved (and go by the 022 umask)
     3) reg files with *any* execute perms, get chmod +x

       file.txt         600 regular file
       symlink.txt      777 symlink to file.txt
       script_owner.sh  700 script where owner can execute
       script_group.sh  610 script where group can execute
       script_world.sh  601 script where world can execute
       dir              744 directory
       dir/dirfile      622 regular file
320 321
     4) the file contents are extracted correctly (though the content of
        each file isn't currently unique)
322

323
    """
324

325 326 327 328
    def setup(self):
        self.tempdir = tempfile.mkdtemp()
        self.old_mask = os.umask(0o022)
        self.symlink_expected_mode = None
329

330 331 332 333 334 335 336 337
    def teardown(self):
        os.umask(self.old_mask)
        shutil.rmtree(self.tempdir, ignore_errors=True)

    def mode(self, path):
        return stat.S_IMODE(os.stat(path).st_mode)

    def confirm_files(self):
J
Jakub Wilk 已提交
338
        # expectations based on 022 umask set above and the unpack logic that
339
        # sets execute permissions, not preservation
340 341 342 343 344 345 346 347 348 349
        for fname, expected_mode, test, expected_contents in [
                ('file.txt', 0o644, os.path.isfile, b'file\n'),
                # We don't test the "symlink.txt" contents for now.
                ('symlink.txt', 0o644, os.path.isfile, None),
                ('script_owner.sh', 0o755, os.path.isfile, b'file\n'),
                ('script_group.sh', 0o755, os.path.isfile, b'file\n'),
                ('script_world.sh', 0o755, os.path.isfile, b'file\n'),
                ('dir', 0o755, os.path.isdir, None),
                (os.path.join('dir', 'dirfile'), 0o644, os.path.isfile, b''),
        ]:
350 351 352 353 354
            path = os.path.join(self.tempdir, fname)
            if path.endswith('symlink.txt') and sys.platform == 'win32':
                # no symlinks created on windows
                continue
            assert test(path), path
355 356 357 358
            if expected_contents is not None:
                with open(path, mode='rb') as f:
                    contents = f.read()
                assert contents == expected_contents, 'fname: {}'.format(fname)
359 360 361 362 363
            if sys.platform == 'win32':
                # the permissions tests below don't apply in windows
                # due to os.chmod being a noop
                continue
            mode = self.mode(path)
364 365 366
            assert mode == expected_mode, (
                "mode: %s, expected mode: %s" % (mode, expected_mode)
            )
367

368
    def test_unpack_tgz(self, data):
369 370 371
        """
        Test unpacking a *.tgz, and setting execute permissions
        """
372
        test_file = data.packages.joinpath("test_tar.tgz")
373 374
        untar_file(test_file, self.tempdir)
        self.confirm_files()
375 376 377 378
        # Check the timestamp of an extracted file
        file_txt_path = os.path.join(self.tempdir, 'file.txt')
        mtime = time.gmtime(os.stat(file_txt_path).st_mtime)
        assert mtime[0:6] == (2013, 8, 16, 5, 13, 37), mtime
379

380
    def test_unpack_zip(self, data):
381 382 383
        """
        Test unpacking a *.zip, and setting execute permissions
        """
384
        test_file = data.packages.joinpath("test_zip.zip")
385 386
        unzip_file(test_file, self.tempdir)
        self.confirm_files()
387 388 389 390 391 392 393 394 395 396 397 398 399 400


class Failer:
    def __init__(self, duration=1):
        self.succeed_after = time.time() + duration

    def call(self, *args, **kw):
        """Fail with OSError self.max_fails times"""
        if time.time() < self.succeed_after:
            raise OSError("Failed")


def test_rmtree_retries(tmpdir, monkeypatch):
    """
401
    Test pip._internal.utils.rmtree will retry failures
402 403 404 405 406 407 408
    """
    monkeypatch.setattr(shutil, 'rmtree', Failer(duration=1).call)
    rmtree('foo')


def test_rmtree_retries_for_3sec(tmpdir, monkeypatch):
    """
409
    Test pip._internal.utils.rmtree will retry failures for no more than 3 sec
410 411 412 413
    """
    monkeypatch.setattr(shutil, 'rmtree', Failer(duration=5).call)
    with pytest.raises(OSError):
        rmtree('foo')
T
Thomas Kluyver 已提交
414

T
Thomas Kluyver 已提交
415

416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433
@pytest.mark.parametrize('path, fs_encoding, expected', [
    (None, None, None),
    # Test passing a text (unicode) string.
    (u'/path/déf', None, u'/path/déf'),
    # Test a bytes object with a non-ascii character.
    (u'/path/déf'.encode('utf-8'), 'utf-8', u'/path/déf'),
    # Test a bytes object with a character that can't be decoded.
    (u'/path/déf'.encode('utf-8'), 'ascii', u"b'/path/d\\xc3\\xa9f'"),
    (u'/path/déf'.encode('utf-16'), 'utf-8',
     u"b'\\xff\\xfe/\\x00p\\x00a\\x00t\\x00h\\x00/"
     "\\x00d\\x00\\xe9\\x00f\\x00'"),
])
def test_path_to_display(monkeypatch, path, fs_encoding, expected):
    monkeypatch.setattr(sys, 'getfilesystemencoding', lambda: fs_encoding)
    actual = path_to_display(path)
    assert actual == expected, 'actual: {!r}'.format(actual)


T
Thomas Kluyver 已提交
434 435 436 437 438
class Test_normalize_path(object):
    # Technically, symlinks are possible on Windows, but you need a special
    # permission bit to create them, and Python 2 doesn't support it anyway, so
    # it's easiest just to skip this test on Windows altogether.
    @pytest.mark.skipif("sys.platform == 'win32'")
439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468
    def test_resolve_symlinks(self, tmpdir):
        print(type(tmpdir))
        print(dir(tmpdir))
        orig_working_dir = os.getcwd()
        os.chdir(tmpdir)
        try:
            d = os.path.join('foo', 'bar')
            f = os.path.join(d, 'file1')
            os.makedirs(d)
            with open(f, 'w'):  # Create the file
                pass

            os.symlink(d, 'dir_link')
            os.symlink(f, 'file_link')

            assert normalize_path(
                'dir_link/file1', resolve_symlinks=True
            ) == os.path.join(tmpdir, f)
            assert normalize_path(
                'dir_link/file1', resolve_symlinks=False
            ) == os.path.join(tmpdir, 'dir_link', 'file1')

            assert normalize_path(
                'file_link', resolve_symlinks=True
            ) == os.path.join(tmpdir, f)
            assert normalize_path(
                'file_link', resolve_symlinks=False
            ) == os.path.join(tmpdir, 'file_link')
        finally:
            os.chdir(orig_working_dir)
469 470 471


class TestHashes(object):
472
    """Tests for pip._internal.utils.hashes"""
473

C
Chris Jerdonek 已提交
474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489
    @pytest.mark.parametrize('hash_name, hex_digest, expected', [
        # Test a value that matches but with the wrong hash_name.
        ('sha384', 128 * 'a', False),
        # Test matching values, including values other than the first.
        ('sha512', 128 * 'a', True),
        ('sha512', 128 * 'b', True),
        # Test a matching hash_name with a value that doesn't match.
        ('sha512', 128 * 'c', False),
    ])
    def test_is_hash_allowed(self, hash_name, hex_digest, expected):
        hashes_data = {
            'sha512': [128 * 'a', 128 * 'b'],
        }
        hashes = Hashes(hashes_data)
        assert hashes.is_hash_allowed(hash_name, hex_digest) == expected

490 491 492 493 494 495 496
    def test_success(self, tmpdir):
        """Make sure no error is raised when at least one hash matches.

        Test check_against_path because it calls everything else.

        """
        file = tmpdir / 'to_hash'
497
        file.write_text('hello')
498 499 500 501 502 503 504 505 506 507 508
        hashes = Hashes({
            'sha256': ['2cf24dba5fb0a30e26e83b2ac5b9e29e'
                       '1b161e5c1fa7425e73043362938b9824'],
            'sha224': ['wrongwrong'],
            'md5': ['5d41402abc4b2a76b9719d911017c592']})
        hashes.check_against_path(file)

    def test_failure(self):
        """Hashes should raise HashMismatch when no hashes match."""
        hashes = Hashes({'sha256': ['wrongwrong']})
        with pytest.raises(HashMismatch):
509
            hashes.check_against_file(BytesIO(b'hello'))
510 511 512 513

    def test_missing_hashes(self):
        """MissingHashes should raise HashMissing when any check is done."""
        with pytest.raises(HashMissing):
514
            MissingHashes().check_against_file(BytesIO(b'hello'))
515 516 517 518 519 520

    def test_unknown_hash(self):
        """Hashes should raise InstallationError when it encounters an unknown
        hash."""
        hashes = Hashes({'badbad': ['dummy']})
        with pytest.raises(InstallationError):
521
            hashes.check_against_file(BytesIO(b'hello'))
522 523 524 525 526 527 528

    def test_non_zero(self):
        """Test that truthiness tests tell whether any known-good hashes
        exist."""
        assert Hashes({'sha256': 'dummy'})
        assert not Hashes()
        assert not Hashes({})
529 530 531


class TestEncoding(object):
532
    """Tests for pip._internal.utils.encoding"""
533

534
    def test_auto_decode_utf_16_le(self):
535 536 537 538
        data = (
            b'\xff\xfeD\x00j\x00a\x00n\x00g\x00o\x00=\x00'
            b'=\x001\x00.\x004\x00.\x002\x00'
        )
539 540 541 542 543 544 545 546 547
        assert data.startswith(codecs.BOM_UTF16_LE)
        assert auto_decode(data) == "Django==1.4.2"

    def test_auto_decode_utf_16_be(self):
        data = (
            b'\xfe\xff\x00D\x00j\x00a\x00n\x00g\x00o\x00='
            b'\x00=\x001\x00.\x004\x00.\x002'
        )
        assert data.startswith(codecs.BOM_UTF16_BE)
548 549
        assert auto_decode(data) == "Django==1.4.2"

X
Xavier Fernandez 已提交
550 551
    def test_auto_decode_no_bom(self):
        assert auto_decode(b'foobar') == u'foobar'
552 553 554 555

    def test_auto_decode_pep263_headers(self):
        latin1_req = u'# coding=latin1\n# Pas trop de café'
        assert auto_decode(latin1_req.encode('latin1')) == latin1_req
556

557 558 559 560 561 562 563 564 565 566
    def test_auto_decode_no_preferred_encoding(self):
        om, em = Mock(), Mock()
        om.return_value = 'ascii'
        em.return_value = None
        data = u'data'
        with patch('sys.getdefaultencoding', om):
            with patch('locale.getpreferredencoding', em):
                ret = auto_decode(data.encode(sys.getdefaultencoding()))
        assert ret == data

567 568 569 570 571
    @pytest.mark.parametrize('encoding', [encoding for bom, encoding in BOMS])
    def test_all_encodings_are_valid(self, encoding):
        # we really only care that there is no LookupError
        assert ''.encode(encoding).decode(encoding) == ''

572

573 574
class TestTempDirectory(object):

575 576
    # No need to test symlinked directories on Windows
    @pytest.mark.skipif("sys.platform == 'win32'")
577 578 579 580 581
    def test_symlinked_path(self):
        with TempDirectory() as tmp_dir:
            assert os.path.exists(tmp_dir.path)

            alt_tmp_dir = tempfile.mkdtemp(prefix="pip-test-")
582
            assert (
583 584
                os.path.dirname(tmp_dir.path) ==
                os.path.dirname(os.path.realpath(alt_tmp_dir))
585 586
            )
            # are we on a system where /tmp is a symlink
587 588 589 590 591
            if os.path.realpath(alt_tmp_dir) != os.path.abspath(alt_tmp_dir):
                assert (
                    os.path.dirname(tmp_dir.path) !=
                    os.path.dirname(alt_tmp_dir)
                )
592
            else:
593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633
                assert (
                    os.path.dirname(tmp_dir.path) ==
                    os.path.dirname(alt_tmp_dir)
                )
            os.rmdir(tmp_dir.path)
            assert not os.path.exists(tmp_dir.path)

    def test_deletes_readonly_files(self):
        def create_file(*args):
            fpath = os.path.join(*args)
            ensure_dir(os.path.dirname(fpath))
            with open(fpath, "w") as f:
                f.write("Holla!")

        def readonly_file(*args):
            fpath = os.path.join(*args)
            os.chmod(fpath, stat.S_IREAD)

        with TempDirectory() as tmp_dir:
            create_file(tmp_dir.path, "normal-file")
            create_file(tmp_dir.path, "readonly-file")
            readonly_file(tmp_dir.path, "readonly-file")

            create_file(tmp_dir.path, "subfolder", "normal-file")
            create_file(tmp_dir.path, "subfolder", "readonly-file")
            readonly_file(tmp_dir.path, "subfolder", "readonly-file")

        assert tmp_dir.path is None

    def test_create_and_cleanup_work(self):
        tmp_dir = TempDirectory()
        assert tmp_dir.path is None

        tmp_dir.create()
        created_path = tmp_dir.path
        assert tmp_dir.path is not None
        assert os.path.exists(created_path)

        tmp_dir.cleanup()
        assert tmp_dir.path is None
        assert not os.path.exists(created_path)
634

S
Steve Dower 已提交
635 636 637 638 639
    @pytest.mark.parametrize("name", [
        "ABC",
        "ABC.dist-info",
        "_+-",
        "_package",
640 641 642 643
        "A......B",
        "AB",
        "A",
        "2",
S
Steve Dower 已提交
644
    ])
645
    def test_adjacent_directory_names(self, name):
S
Steve Dower 已提交
646 647 648
        def names():
            return AdjacentTempDirectory._generate_names(name)

649 650 651 652 653 654 655 656
        chars = AdjacentTempDirectory.LEADING_CHARS

        # Ensure many names are unique
        # (For long *name*, this sequence can be extremely long.
        # However, since we're only ever going to take the first
        # result that works, provided there are many of those
        # and that shorter names result in totally unique sets,
        # it's okay to skip part of the test.)
657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687
        some_names = list(itertools.islice(names(), 1000))
        # We should always get at least 1000 names
        assert len(some_names) == 1000

        # Ensure original name does not appear early in the set
        assert name not in some_names

        if len(name) > 2:
            # Names should be at least 90% unique (given the infinite
            # range of inputs, and the possibility that generated names
            # may already exist on disk anyway, this is a much cheaper
            # criteria to enforce than complete uniqueness).
            assert len(some_names) > 0.9 * len(set(some_names))

            # Ensure the first few names are the same length as the original
            same_len = list(itertools.takewhile(
                lambda x: len(x) == len(name),
                some_names
            ))
            assert len(same_len) > 10

            # Check the first group are correct
            expected_names = ['~' + name[1:]]
            expected_names.extend('~' + c + name[2:] for c in chars)
            for x, y in zip(some_names, expected_names):
                assert x == y

        else:
            # All names are going to be longer than our original
            assert min(len(x) for x in some_names) > 1

P
Pi Delport 已提交
688
            # All names are going to be unique
689 690 691 692 693 694 695 696
            assert len(some_names) == len(set(some_names))

            if len(name) == 2:
                # All but the first name are going to end with our original
                assert all(x.endswith(name) for x in some_names[1:])
            else:
                # All names are going to end with our original
                assert all(x.endswith(name) for x in some_names)
697

698 699 700 701 702 703 704 705
    @pytest.mark.parametrize("name", [
        "A",
        "ABC",
        "ABC.dist-info",
        "_+-",
        "_package",
    ])
    def test_adjacent_directory_exists(self, name, tmpdir):
706 707 708
        block_name, expect_name = itertools.islice(
            AdjacentTempDirectory._generate_names(name), 2)

709 710 711 712 713
        original = os.path.join(tmpdir, name)
        blocker = os.path.join(tmpdir, block_name)

        ensure_dir(original)
        ensure_dir(blocker)
714

715 716
        with AdjacentTempDirectory(original) as atmp_dir:
            assert expect_name == os.path.split(atmp_dir.path)[1]
717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733

    def test_adjacent_directory_permission_error(self, monkeypatch):
        name = "ABC"

        def raising_mkdir(*args, **kwargs):
            raise OSError("Unknown OSError")

        with TempDirectory() as tmp_dir:
            original = os.path.join(tmp_dir.path, name)

            ensure_dir(original)
            monkeypatch.setattr("os.mkdir", raising_mkdir)

            with pytest.raises(OSError):
                with AdjacentTempDirectory(original):
                    pass

734

735 736 737 738
def raises(error):
    raise error


739
class TestGlibc(object):
M
Mark Williams 已提交
740
    def test_manylinux_check_glibc_version(self):
741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770
        """
        Test that the check_glibc_version function is robust against weird
        glibc version strings.
        """
        for two_twenty in ["2.20",
                           # used by "linaro glibc", see gh-3588
                           "2.20-2014.11",
                           # weird possibilities that I just made up
                           "2.20+dev",
                           "2.20-custom",
                           "2.20.1",
                           ]:
            assert check_glibc_version(two_twenty, 2, 15)
            assert check_glibc_version(two_twenty, 2, 20)
            assert not check_glibc_version(two_twenty, 2, 21)
            assert not check_glibc_version(two_twenty, 3, 15)
            assert not check_glibc_version(two_twenty, 1, 15)

        # For strings that we just can't parse at all, we should warn and
        # return false
        for bad_string in ["asdf", "", "foo.bar"]:
            with warnings.catch_warnings(record=True) as ws:
                warnings.filterwarnings("always")
                assert not check_glibc_version(bad_string, 2, 5)
                for w in ws:
                    if "Expected glibc version with" in str(w.message):
                        break
                else:
                    # Didn't find the warning we were expecting
                    assert False
771

772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800
    def test_glibc_version_string(self, monkeypatch):
        monkeypatch.setattr(
            os, "confstr", lambda x: "glibc 2.20", raising=False,
        )
        assert glibc_version_string() == "2.20"

    def test_glibc_version_string_confstr(self, monkeypatch):
        monkeypatch.setattr(
            os, "confstr", lambda x: "glibc 2.20", raising=False,
        )
        assert glibc_version_string_confstr() == "2.20"

    @pytest.mark.parametrize("failure", [
        lambda x: raises(ValueError),
        lambda x: raises(OSError),
        lambda x: "XXX",
    ])
    def test_glibc_version_string_confstr_fail(self, monkeypatch, failure):
        monkeypatch.setattr(os, "confstr", failure, raising=False)
        assert glibc_version_string_confstr() is None

    def test_glibc_version_string_confstr_missing(self, monkeypatch):
        monkeypatch.delattr(os, "confstr", raising=False)
        assert glibc_version_string_confstr() is None

    def test_glibc_version_string_ctypes_missing(self, monkeypatch):
        monkeypatch.setitem(sys.modules, "ctypes", None)
        assert glibc_version_string_ctypes() is None

801

802 803 804 805 806 807 808 809 810 811 812 813
@pytest.mark.parametrize('version_info, expected', [
    ((), (0, 0, 0)),
    ((3, ), (3, 0, 0)),
    ((3, 6), (3, 6, 0)),
    ((3, 6, 2), (3, 6, 2)),
    ((3, 6, 2, 4), (3, 6, 2)),
])
def test_normalize_version_info(version_info, expected):
    actual = normalize_version_info(version_info)
    assert actual == expected


814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831
class TestGetProg(object):

    @pytest.mark.parametrize(
        ("argv", "executable", "expected"),
        [
            ('/usr/bin/pip', '', 'pip'),
            ('-c', '/usr/bin/python', '/usr/bin/python -m pip'),
            ('__main__.py', '/usr/bin/python', '/usr/bin/python -m pip'),
            ('/usr/bin/pip3', '', 'pip3'),
        ]
    )
    def test_get_prog(self, monkeypatch, argv, executable, expected):
        monkeypatch.setattr('pip._internal.utils.misc.sys.argv', [argv])
        monkeypatch.setattr(
            'pip._internal.utils.misc.sys.executable',
            executable
        )
        assert get_prog() == expected
P
Pradyun Gedam 已提交
832 833


834 835 836
@pytest.mark.parametrize('args, expected', [
    (['pip', 'list'], 'pip list'),
    (['foo', 'space space', 'new\nline', 'double"quote', "single'quote"],
837
     """foo 'space space' 'new\nline' 'double"quote' 'single'"'"'quote'"""),
838 839 840
    # Test HiddenText arguments.
    (make_command(hide_value('secret1'), 'foo', hide_value('secret2')),
        "'****' foo '****'"),
841 842 843 844 845 846
])
def test_format_command_args(args, expected):
    actual = format_command_args(args)
    assert actual == expected


847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868
def test_make_subprocess_output_error():
    cmd_args = ['test', 'has space']
    cwd = '/path/to/cwd'
    lines = ['line1\n', 'line2\n', 'line3\n']
    actual = make_subprocess_output_error(
        cmd_args=cmd_args,
        cwd=cwd,
        lines=lines,
        exit_status=3,
    )
    expected = dedent("""\
    Command errored out with exit status 3:
     command: test 'has space'
         cwd: /path/to/cwd
    Complete output (3 lines):
    line1
    line2
    line3
    ----------------------------------------""")
    assert actual == expected, 'actual: {}'.format(actual)


869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896
def test_make_subprocess_output_error__non_ascii_command_arg(monkeypatch):
    """
    Test a command argument with a non-ascii character.
    """
    cmd_args = ['foo', 'déf']
    if sys.version_info[0] == 2:
        # Check in Python 2 that the str (bytes object) with the non-ascii
        # character has the encoding we expect. (This comes from the source
        # code encoding at the top of the file.)
        assert cmd_args[1].decode('utf-8') == u'déf'

    # We need to monkeypatch so the encoding will be correct on Windows.
    monkeypatch.setattr(locale, 'getpreferredencoding', lambda: 'utf-8')
    actual = make_subprocess_output_error(
        cmd_args=cmd_args,
        cwd='/path/to/cwd',
        lines=[],
        exit_status=1,
    )
    expected = dedent(u"""\
    Command errored out with exit status 1:
     command: foo 'déf'
         cwd: /path/to/cwd
    Complete output (0 lines):
    ----------------------------------------""")
    assert actual == expected, u'actual: {}'.format(actual)


897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948
@pytest.mark.skipif("sys.version_info < (3,)")
def test_make_subprocess_output_error__non_ascii_cwd_python_3(monkeypatch):
    """
    Test a str (text) cwd with a non-ascii character in Python 3.
    """
    cmd_args = ['test']
    cwd = '/path/to/cwd/déf'
    actual = make_subprocess_output_error(
        cmd_args=cmd_args,
        cwd=cwd,
        lines=[],
        exit_status=1,
    )
    expected = dedent("""\
    Command errored out with exit status 1:
     command: test
         cwd: /path/to/cwd/déf
    Complete output (0 lines):
    ----------------------------------------""")
    assert actual == expected, 'actual: {}'.format(actual)


@pytest.mark.parametrize('encoding', [
    'utf-8',
    # Test a Windows encoding.
    'cp1252',
])
@pytest.mark.skipif("sys.version_info >= (3,)")
def test_make_subprocess_output_error__non_ascii_cwd_python_2(
    monkeypatch, encoding,
):
    """
    Test a str (bytes object) cwd with a non-ascii character in Python 2.
    """
    cmd_args = ['test']
    cwd = u'/path/to/cwd/déf'.encode(encoding)
    monkeypatch.setattr(sys, 'getfilesystemencoding', lambda: encoding)
    actual = make_subprocess_output_error(
        cmd_args=cmd_args,
        cwd=cwd,
        lines=[],
        exit_status=1,
    )
    expected = dedent(u"""\
    Command errored out with exit status 1:
     command: test
         cwd: /path/to/cwd/déf
    Complete output (0 lines):
    ----------------------------------------""")
    assert actual == expected, u'actual: {}'.format(actual)


949
# This test is mainly important for checking unicode in Python 2.
950
def test_make_subprocess_output_error__non_ascii_line():
951
    """
952
    Test a line with a non-ascii character.
953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970
    """
    lines = [u'curly-quote: \u2018\n']
    actual = make_subprocess_output_error(
        cmd_args=['test'],
        cwd='/path/to/cwd',
        lines=lines,
        exit_status=1,
    )
    expected = dedent(u"""\
    Command errored out with exit status 1:
     command: test
         cwd: /path/to/cwd
    Complete output (1 lines):
    curly-quote: \u2018
    ----------------------------------------""")
    assert actual == expected, u'actual: {}'.format(actual)


971
class FakeSpinner(SpinnerInterface):
972

973 974 975
    def __init__(self):
        self.spin_count = 0
        self.final_status = None
976

977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000
    def spin(self):
        self.spin_count += 1

    def finish(self, final_status):
        self.final_status = final_status


class TestCallSubprocess(object):

    """
    Test call_subprocess().
    """

    def check_result(
        self, capfd, caplog, log_level, spinner, result, expected,
        expected_spinner,
    ):
        """
        Check the result of calling call_subprocess().

        :param log_level: the logging level that caplog was set to.
        :param spinner: the FakeSpinner object passed to call_subprocess()
            to be checked.
        :param result: the call_subprocess() return value to be checked.
1001
        :param expected: a pair (expected_proc, expected_records), where
1002 1003 1004
            1) `expected_proc` is the expected return value of
              call_subprocess() as a list of lines, or None if the return
              value is expected to be None;
1005
            2) `expected_records` is the expected value of
1006 1007 1008 1009
              caplog.record_tuples.
        :param expected_spinner: a 2-tuple of the spinner's expected
            (spin_count, final_status).
        """
1010
        expected_proc, expected_records = expected
1011 1012

        if expected_proc is None:
1013
            assert result is None
1014 1015 1016
        else:
            assert result.splitlines() == expected_proc

1017
        # Confirm that stdout and stderr haven't been written to.
1018
        captured = capfd.readouterr()
1019
        assert (captured.out, captured.err) == ('', '')
1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051

        records = caplog.record_tuples
        if len(records) != len(expected_records):
            raise RuntimeError('{} != {}'.format(records, expected_records))

        for record, expected_record in zip(records, expected_records):
            # Check the logger_name and log level parts exactly.
            assert record[:2] == expected_record[:2]
            # For the message portion, check only a substring.  Also, we
            # can't use startswith() since the order of stdout and stderr
            # isn't guaranteed in cases where stderr is also present.
            # For example, we observed the stderr lines coming before stdout
            # in CI for PyPy 2.7 even though stdout happens first
            # chronologically.
            assert expected_record[2] in record[2]

        assert (spinner.spin_count, spinner.final_status) == expected_spinner

    def prepare_call(self, caplog, log_level, command=None):
        if command is None:
            command = 'print("Hello"); print("world")'

        caplog.set_level(log_level)
        spinner = FakeSpinner()
        args = [sys.executable, '-c', command]

        return (args, spinner)

    def test_debug_logging(self, capfd, caplog):
        """
        Test DEBUG logging (and without passing show_stdout=True).
        """
1052
        log_level = DEBUG
1053 1054 1055
        args, spinner = self.prepare_call(caplog, log_level)
        result = call_subprocess(args, spinner=spinner)

1056 1057 1058 1059
        expected = (['Hello', 'world'], [
            ('pip.subprocessor', DEBUG, 'Running command '),
            ('pip.subprocessor', DEBUG, 'Hello'),
            ('pip.subprocessor', DEBUG, 'world'),
1060 1061 1062 1063 1064
        ])
        # The spinner shouldn't spin in this case since the subprocess
        # output is already being logged to the console.
        self.check_result(
            capfd, caplog, log_level, spinner, result, expected,
1065
            expected_spinner=(0, None),
1066
        )
P
Pradyun Gedam 已提交
1067

1068 1069 1070 1071
    def test_info_logging(self, capfd, caplog):
        """
        Test INFO logging (and without passing show_stdout=True).
        """
1072
        log_level = INFO
1073 1074 1075
        args, spinner = self.prepare_call(caplog, log_level)
        result = call_subprocess(args, spinner=spinner)

1076
        expected = (['Hello', 'world'], [])
1077 1078 1079 1080 1081 1082
        # The spinner should spin twice in this case since the subprocess
        # output isn't being written to the console.
        self.check_result(
            capfd, caplog, log_level, spinner, result, expected,
            expected_spinner=(2, 'done'),
        )
P
Pradyun Gedam 已提交
1083

1084 1085 1086 1087 1088
    def test_info_logging__subprocess_error(self, capfd, caplog):
        """
        Test INFO logging of a subprocess with an error (and without passing
        show_stdout=True).
        """
1089
        log_level = INFO
1090 1091 1092
        command = 'print("Hello"); print("world"); exit("fail")'
        args, spinner = self.prepare_call(caplog, log_level, command=command)

1093
        with pytest.raises(InstallationError) as exc:
1094 1095
            call_subprocess(args, spinner=spinner)
        result = None
1096 1097 1098 1099 1100
        exc_message = str(exc.value)
        assert exc_message.startswith(
            'Command errored out with exit status 1: '
        )
        assert exc_message.endswith('Check the logs for full command output.')
1101

1102
        expected = (None, [
1103
            ('pip.subprocessor', ERROR, 'Complete output (3 lines):\n'),
1104 1105 1106 1107 1108 1109
        ])
        # The spinner should spin three times in this case since the
        # subprocess output isn't being written to the console.
        self.check_result(
            capfd, caplog, log_level, spinner, result, expected,
            expected_spinner=(3, 'error'),
1110
        )
1111

1112 1113 1114 1115 1116 1117 1118 1119 1120 1121
        # Do some further checking on the captured log records to confirm
        # that the subprocess output was logged.
        last_record = caplog.record_tuples[-1]
        last_message = last_record[2]
        lines = last_message.splitlines()

        # We have to sort before comparing the lines because we can't
        # guarantee the order in which stdout and stderr will appear.
        # For example, we observed the stderr lines coming before stdout
        # in CI for PyPy 2.7 even though stdout happens first chronologically.
1122 1123 1124 1125 1126 1127
        actual = sorted(lines)
        # Test the "command" line separately because we can't test an
        # exact match.
        command_line = actual.pop(1)
        assert actual == [
            '     cwd: None',
1128
            '----------------------------------------',
1129 1130
            'Command errored out with exit status 1:',
            'Complete output (3 lines):',
1131 1132 1133
            'Hello',
            'fail',
            'world',
1134 1135 1136 1137
        ], 'lines: {}'.format(actual)  # Show the full output on failure.

        assert command_line.startswith(' command: ')
        assert command_line.endswith('print("world"); exit("fail")\'')
1138 1139 1140 1141 1142

    def test_info_logging_with_show_stdout_true(self, capfd, caplog):
        """
        Test INFO logging with show_stdout=True.
        """
1143
        log_level = INFO
1144 1145 1146
        args, spinner = self.prepare_call(caplog, log_level)
        result = call_subprocess(args, spinner=spinner, show_stdout=True)

1147 1148 1149 1150 1151
        expected = (['Hello', 'world'], [
            ('pip.subprocessor', INFO, 'Running command '),
            ('pip.subprocessor', INFO, 'Hello'),
            ('pip.subprocessor', INFO, 'world'),
        ])
1152 1153 1154 1155
        # The spinner shouldn't spin in this case since the subprocess
        # output is already being written to the console.
        self.check_result(
            capfd, caplog, log_level, spinner, result, expected,
1156
            expected_spinner=(0, None),
1157 1158 1159 1160 1161 1162
        )

    @pytest.mark.parametrize((
        'exit_status', 'show_stdout', 'extra_ok_returncodes', 'log_level',
        'expected'),
        [
1163 1164 1165 1166 1167 1168
            # The spinner should show here because show_stdout=False means
            # the subprocess should get logged at DEBUG level, but the passed
            # log level is only INFO.
            (0, False, None, INFO, (None, 'done', 2)),
            # Test some cases where the spinner should not be shown.
            (0, False, None, DEBUG, (None, None, 0)),
1169
            # Test show_stdout=True.
1170 1171 1172 1173 1174 1175
            (0, True, None, DEBUG, (None, None, 0)),
            (0, True, None, INFO, (None, None, 0)),
            # The spinner should show here because show_stdout=True means
            # the subprocess should get logged at INFO level, but the passed
            # log level is only WARNING.
            (0, True, None, WARNING, (None, 'done', 2)),
1176
            # Test a non-zero exit status.
1177
            (3, False, None, INFO, (InstallationError, 'error', 2)),
1178
            # Test a non-zero exit status also in extra_ok_returncodes.
1179
            (3, False, (3, ), INFO, (None, 'done', 2)),
1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218
    ])
    def test_spinner_finish(
        self, exit_status, show_stdout, extra_ok_returncodes, log_level,
        caplog, expected,
    ):
        """
        Test that the spinner finishes correctly.
        """
        expected_exc_type = expected[0]
        expected_final_status = expected[1]
        expected_spin_count = expected[2]

        command = (
            'print("Hello"); print("world"); exit({})'.format(exit_status)
        )
        args, spinner = self.prepare_call(caplog, log_level, command=command)
        try:
            call_subprocess(
                args,
                show_stdout=show_stdout,
                extra_ok_returncodes=extra_ok_returncodes,
                spinner=spinner,
            )
        except Exception as exc:
            exc_type = type(exc)
        else:
            exc_type = None

        assert exc_type == expected_exc_type
        assert spinner.final_status == expected_final_status
        assert spinner.spin_count == expected_spin_count

    def test_closes_stdin(self):
        with pytest.raises(InstallationError):
            call_subprocess(
                [sys.executable, '-c', 'input()'],
                show_stdout=True,
            )

1219

1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235
@pytest.mark.skipif("sys.platform == 'win32'")
def test_path_to_url_unix():
    assert path_to_url('/tmp/file') == 'file:///tmp/file'
    path = os.path.join(os.getcwd(), 'file')
    assert path_to_url('file') == 'file://' + urllib_request.pathname2url(path)


@pytest.mark.skipif("sys.platform != 'win32'")
def test_path_to_url_win():
    assert path_to_url('c:/tmp/file') == 'file:///C:/tmp/file'
    assert path_to_url('c:\\tmp\\file') == 'file:///C:/tmp/file'
    assert path_to_url(r'\\unc\as\path') == 'file://unc/as/path'
    path = os.path.join(os.getcwd(), 'file')
    assert path_to_url('file') == 'file:' + urllib_request.pathname2url(path)


1236
@pytest.mark.parametrize('host_port, expected_netloc', [
1237
    # Test domain name.
1238 1239
    (('example.com', None), 'example.com'),
    (('example.com', 5000), 'example.com:5000'),
1240
    # Test IPv4 address.
1241 1242
    (('127.0.0.1', None), '127.0.0.1'),
    (('127.0.0.1', 5000), '127.0.0.1:5000'),
1243
    # Test bare IPv6 address.
1244
    (('2001:db6::1', None), '2001:db6::1'),
1245
    # Test IPv6 with port.
1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266
    (('2001:db6::1', 5000), '[2001:db6::1]:5000'),
])
def test_build_netloc(host_port, expected_netloc):
    assert build_netloc(*host_port) == expected_netloc


@pytest.mark.parametrize('netloc, expected_url, expected_host_port', [
    # Test domain name.
    ('example.com', 'https://example.com', ('example.com', None)),
    ('example.com:5000', 'https://example.com:5000', ('example.com', 5000)),
    # Test IPv4 address.
    ('127.0.0.1', 'https://127.0.0.1', ('127.0.0.1', None)),
    ('127.0.0.1:5000', 'https://127.0.0.1:5000', ('127.0.0.1', 5000)),
    # Test bare IPv6 address.
    ('2001:db6::1', 'https://[2001:db6::1]', ('2001:db6::1', None)),
    # Test IPv6 with port.
    (
        '[2001:db6::1]:5000',
        'https://[2001:db6::1]:5000',
        ('2001:db6::1', 5000)
    ),
1267 1268 1269 1270
    # Test netloc with auth.
    (
        'user:password@localhost:5000',
        'https://user:password@localhost:5000',
1271
        ('localhost', 5000)
1272 1273
    )
])
1274 1275
def test_build_url_from_netloc_and_parse_netloc(
    netloc, expected_url, expected_host_port,
1276 1277
):
    assert build_url_from_netloc(netloc) == expected_url
1278
    assert parse_netloc(netloc) == expected_host_port
1279 1280


1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293
@pytest.mark.parametrize('netloc, expected', [
    # Test a basic case.
    ('example.com', ('example.com', (None, None))),
    # Test with username and no password.
    ('user@example.com', ('example.com', ('user', None))),
    # Test with username and password.
    ('user:pass@example.com', ('example.com', ('user', 'pass'))),
    # Test with username and empty password.
    ('user:@example.com', ('example.com', ('user', ''))),
    # Test the password containing an @ symbol.
    ('user:pass@word@example.com', ('example.com', ('user', 'pass@word'))),
    # Test the password containing a : symbol.
    ('user:pass:word@example.com', ('example.com', ('user', 'pass:word'))),
C
Chris Jerdonek 已提交
1294 1295 1296
    # Test URL-encoded reserved characters.
    ('user%3Aname:%23%40%5E@example.com',
     ('example.com', ('user:name', '#@^'))),
1297 1298 1299 1300 1301 1302
])
def test_split_auth_from_netloc(netloc, expected):
    actual = split_auth_from_netloc(netloc)
    assert actual == expected


1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330
@pytest.mark.parametrize('url, expected', [
    # Test a basic case.
    ('http://example.com/path#anchor',
     ('http://example.com/path#anchor', 'example.com', (None, None))),
    # Test with username and no password.
    ('http://user@example.com/path#anchor',
     ('http://example.com/path#anchor', 'example.com', ('user', None))),
    # Test with username and password.
    ('http://user:pass@example.com/path#anchor',
     ('http://example.com/path#anchor', 'example.com', ('user', 'pass'))),
    # Test with username and empty password.
    ('http://user:@example.com/path#anchor',
     ('http://example.com/path#anchor', 'example.com', ('user', ''))),
    # Test the password containing an @ symbol.
    ('http://user:pass@word@example.com/path#anchor',
     ('http://example.com/path#anchor', 'example.com', ('user', 'pass@word'))),
    # Test the password containing a : symbol.
    ('http://user:pass:word@example.com/path#anchor',
     ('http://example.com/path#anchor', 'example.com', ('user', 'pass:word'))),
    # Test URL-encoded reserved characters.
    ('http://user%3Aname:%23%40%5E@example.com/path#anchor',
     ('http://example.com/path#anchor', 'example.com', ('user:name', '#@^'))),
])
def test_split_auth_netloc_from_url(url, expected):
    actual = split_auth_netloc_from_url(url)
    assert actual == expected


1331 1332 1333 1334
@pytest.mark.parametrize('netloc, expected', [
    # Test a basic case.
    ('example.com', 'example.com'),
    # Test with username and no password.
1335
    ('accesstoken@example.com', '****@example.com'),
1336 1337 1338 1339 1340 1341 1342 1343
    # Test with username and password.
    ('user:pass@example.com', 'user:****@example.com'),
    # Test with username and empty password.
    ('user:@example.com', 'user:****@example.com'),
    # Test the password containing an @ symbol.
    ('user:pass@word@example.com', 'user:****@example.com'),
    # Test the password containing a : symbol.
    ('user:pass:word@example.com', 'user:****@example.com'),
1344 1345
    # Test URL-encoded reserved characters.
    ('user%3Aname:%23%40%5E@example.com', 'user%3Aname:****@example.com'),
1346 1347 1348 1349 1350 1351
])
def test_redact_netloc(netloc, expected):
    actual = redact_netloc(netloc)
    assert actual == expected


1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369
@pytest.mark.parametrize('auth_url, expected_url', [
    ('https://user:pass@domain.tld/project/tags/v0.2',
     'https://domain.tld/project/tags/v0.2'),
    ('https://domain.tld/project/tags/v0.2',
     'https://domain.tld/project/tags/v0.2',),
    ('https://user:pass@domain.tld/svn/project/trunk@8181',
     'https://domain.tld/svn/project/trunk@8181'),
    ('https://domain.tld/project/trunk@8181',
     'https://domain.tld/project/trunk@8181',),
    ('git+https://pypi.org/something',
     'git+https://pypi.org/something'),
    ('git+https://user:pass@pypi.org/something',
     'git+https://pypi.org/something'),
    ('git+ssh://git@pypi.org/something',
     'git+ssh://pypi.org/something'),
])
def test_remove_auth_from_url(auth_url, expected_url):
    url = remove_auth_from_url(auth_url)
1370
    assert url == expected_url
1371 1372 1373


@pytest.mark.parametrize('auth_url, expected_url', [
1374
    ('https://accesstoken@example.com/abc', 'https://****@example.com/abc'),
1375 1376
    ('https://user:password@example.com', 'https://user:****@example.com'),
    ('https://user:@example.com', 'https://user:****@example.com'),
1377 1378 1379 1380
    ('https://example.com', 'https://example.com'),
    # Test URL-encoded reserved characters.
    ('https://user%3Aname:%23%40%5E@example.com',
     'https://user%3Aname:****@example.com'),
1381
])
1382 1383
def test_redact_auth_from_url(auth_url, expected_url):
    url = redact_auth_from_url(auth_url)
1384
    assert url == expected_url
1385 1386


1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453
class TestHiddenText:

    def test_basic(self):
        """
        Test str(), repr(), and attribute access.
        """
        hidden = HiddenText('my-secret', redacted='######')
        assert repr(hidden) == "<HiddenText '######'>"
        assert str(hidden) == '######'
        assert hidden.redacted == '######'
        assert hidden.secret == 'my-secret'

    def test_equality_with_str(self):
        """
        Test equality (and inequality) with str objects.
        """
        hidden = HiddenText('secret', redacted='****')

        # Test that the object doesn't compare equal to either its original
        # or redacted forms.
        assert hidden != hidden.secret
        assert hidden.secret != hidden

        assert hidden != hidden.redacted
        assert hidden.redacted != hidden

    def test_equality_same_secret(self):
        """
        Test equality with an object having the same secret.
        """
        # Choose different redactions for the two objects.
        hidden1 = HiddenText('secret', redacted='****')
        hidden2 = HiddenText('secret', redacted='####')

        assert hidden1 == hidden2
        # Also test __ne__.  This assertion fails in Python 2 without
        # defining HiddenText.__ne__.
        assert not hidden1 != hidden2

    def test_equality_different_secret(self):
        """
        Test equality with an object having a different secret.
        """
        hidden1 = HiddenText('secret-1', redacted='****')
        hidden2 = HiddenText('secret-2', redacted='****')

        assert hidden1 != hidden2
        # Also test __eq__.
        assert not hidden1 == hidden2


def test_hide_value():
    hidden = hide_value('my-secret')
    assert repr(hidden) == "<HiddenText '****'>"
    assert str(hidden) == '****'
    assert hidden.redacted == '****'
    assert hidden.secret == 'my-secret'


def test_hide_url():
    hidden_url = hide_url('https://user:password@example.com')
    assert repr(hidden_url) == "<HiddenText 'https://user:****@example.com'>"
    assert str(hidden_url) == 'https://user:****@example.com'
    assert hidden_url.redacted == 'https://user:****@example.com'
    assert hidden_url.secret == 'https://user:password@example.com'


1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524 1525 1526 1527
@pytest.fixture()
def patch_deprecation_check_version():
    # We do this, so that the deprecation tests are easier to write.
    import pip._internal.utils.deprecation as d
    old_version = d.current_version
    d.current_version = "1.0"
    yield
    d.current_version = old_version


@pytest.mark.usefixtures("patch_deprecation_check_version")
@pytest.mark.parametrize("replacement", [None, "a magic 8 ball"])
@pytest.mark.parametrize("gone_in", [None, "2.0"])
@pytest.mark.parametrize("issue", [None, 988])
def test_deprecated_message_contains_information(gone_in, replacement, issue):
    with pytest.warns(PipDeprecationWarning) as record:
        deprecated(
            "Stop doing this!",
            replacement=replacement,
            gone_in=gone_in,
            issue=issue,
        )

    assert len(record) == 1
    message = record[0].message.args[0]

    assert "DEPRECATION: Stop doing this!" in message
    # Ensure non-None values are mentioned.
    for item in [gone_in, replacement, issue]:
        if item is not None:
            assert str(item) in message


@pytest.mark.usefixtures("patch_deprecation_check_version")
@pytest.mark.parametrize("replacement", [None, "a magic 8 ball"])
@pytest.mark.parametrize("issue", [None, 988])
def test_deprecated_raises_error_if_too_old(replacement, issue):
    with pytest.raises(PipDeprecationWarning) as exception:
        deprecated(
            "Stop doing this!",
            gone_in="1.0",  # this matches the patched version.
            replacement=replacement,
            issue=issue,
        )

    message = exception.value.args[0]

    assert "DEPRECATION: Stop doing this!" in message
    assert "1.0" in message
    # Ensure non-None values are mentioned.
    for item in [replacement, issue]:
        if item is not None:
            assert str(item) in message


@pytest.mark.usefixtures("patch_deprecation_check_version")
def test_deprecated_message_reads_well():
    with pytest.raises(PipDeprecationWarning) as exception:
        deprecated(
            "Stop doing this!",
            gone_in="1.0",  # this matches the patched version.
            replacement="to be nicer",
            issue="100000",  # I hope we never reach this number.
        )

    message = exception.value.args[0]

    assert message == (
        "DEPRECATION: Stop doing this! "
        "pip 1.0 will remove support for this functionality. "
        "A possible replacement is to be nicer. "
        "You can find discussion regarding this at "
        "https://github.com/pypa/pip/issues/100000."
    )
1528 1529


1530 1531
def test_make_setuptools_shim_args():
    # Test all arguments at once, including the overall ordering.
1532
    args = make_setuptools_shim_args(
1533 1534 1535 1536 1537 1538 1539 1540 1541 1542 1543 1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554
        '/dir/path/setup.py',
        global_options=['--some', '--option'],
        no_user_config=True,
        unbuffered_output=True,
    )

    assert args[1:3] == ['-u', '-c']
    # Spot-check key aspects of the command string.
    assert "sys.argv[0] = '/dir/path/setup.py'" in args[3]
    assert "__file__='/dir/path/setup.py'" in args[3]
    assert args[4:] == ['--some', '--option', '--no-user-cfg']


@pytest.mark.parametrize('global_options', [
    None,
    [],
    ['--some', '--option']
])
def test_make_setuptools_shim_args__global_options(global_options):
    args = make_setuptools_shim_args(
        '/dir/path/setup.py',
        global_options=global_options,
1555 1556
    )

1557 1558 1559 1560 1561 1562
    if global_options:
        assert len(args) == 5
        for option in global_options:
            assert option in args
    else:
        assert len(args) == 3
1563 1564


1565 1566 1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 1579 1580
@pytest.mark.parametrize('no_user_config', [False, True])
def test_make_setuptools_shim_args__no_user_config(no_user_config):
    args = make_setuptools_shim_args(
        '/dir/path/setup.py',
        no_user_config=no_user_config,
    )
    assert ('--no-user-cfg' in args) == no_user_config


@pytest.mark.parametrize('unbuffered_output', [False, True])
def test_make_setuptools_shim_args__unbuffered_output(unbuffered_output):
    args = make_setuptools_shim_args(
        '/dir/path/setup.py',
        unbuffered_output=unbuffered_output
    )
    assert ('-u' in args) == unbuffered_output