test_collector.py

import itertools
import logging
import os.path
import re
import urllib.request
import uuid
from textwrap import dedent
from typing import List, Optional, Tuple
from unittest import mock

import pytest
from pip._vendor import html5lib, requests

from pip._internal.exceptions import NetworkConnectionError
from pip._internal.index.collector import (
    HTMLPage,
    LinkCollector,
    _clean_link,
    _clean_url_path,
    _determine_base_url,
    _get_html_page,
    _get_html_response,
    _make_html_page,
    _NotHTML,
    _NotHTTP,
    parse_links,
)
from pip._internal.index.sources import _FlatDirectorySource, _IndexDirectorySource
from pip._internal.models.candidate import InstallationCandidate
from pip._internal.models.index import PyPI
from pip._internal.models.link import Link
from pip._internal.network.session import PipSession
from tests.lib import TestData, make_test_link_collector
from tests.lib.path import Path


@pytest.mark.parametrize(
    "url",
    [
        "ftp://python.org/python-3.7.1.zip",
        "file:///opt/data/pip-18.0.tar.gz",
    ],
)
def test_get_html_response_archive_to_naive_scheme(url: str) -> None:
    """
    `_get_html_response()` should error on an archive-like URL if the scheme
    does not allow "poking" without getting data.
    """
    with pytest.raises(_NotHTTP):
        _get_html_response(url, session=mock.Mock(PipSession))


@pytest.mark.parametrize(
    "url, content_type",
    [
        ("http://python.org/python-3.7.1.zip", "application/zip"),
        ("https://pypi.org/pip-18.0.tar.gz", "application/gzip"),
    ],
)
@mock.patch("pip._internal.index.collector.raise_for_status")
def test_get_html_response_archive_to_http_scheme(
    mock_raise_for_status: mock.Mock, url: str, content_type: str
) -> None:
    """
    `_get_html_response()` should send a HEAD request on an archive-like URL
    if the scheme supports it, and raise `_NotHTML` if the response isn't HTML.
    """
    session = mock.Mock(PipSession)
    session.head.return_value = mock.Mock(
        **{
            "request.method": "HEAD",
            "headers": {"Content-Type": content_type},
        }
    )

    with pytest.raises(_NotHTML) as ctx:
        _get_html_response(url, session=session)

    session.assert_has_calls(
        [
            mock.call.head(url, allow_redirects=True),
        ]
    )
    mock_raise_for_status.assert_called_once_with(session.head.return_value)
    assert ctx.value.args == (content_type, "HEAD")


@pytest.mark.parametrize(
    "url",
    [
        ("ftp://python.org/python-3.7.1.zip"),
        ("file:///opt/data/pip-18.0.tar.gz"),
    ],
)
def test_get_html_page_invalid_content_type_archive(
    caplog: pytest.LogCaptureFixture, url: str
) -> None:
    """`_get_html_page()` should warn if an archive URL is not HTML
    and therefore cannot be used for a HEAD request.
    """
    caplog.set_level(logging.WARNING)
    link = Link(url)

    session = mock.Mock(PipSession)

    assert _get_html_page(link, session=session) is None
    assert (
        "pip._internal.index.collector",
        logging.WARNING,
        "Skipping page {} because it looks like an archive, and cannot "
        "be checked by a HTTP HEAD request.".format(url),
    ) in caplog.record_tuples


@pytest.mark.parametrize(
    "url",
    [
        "http://python.org/python-3.7.1.zip",
        "https://pypi.org/pip-18.0.tar.gz",
    ],
)
@mock.patch("pip._internal.index.collector.raise_for_status")
def test_get_html_response_archive_to_http_scheme_is_html(
    mock_raise_for_status: mock.Mock, url: str
) -> None:
    """
    `_get_html_response()` should work with archive-like URLs if the HEAD
    request is responded with text/html.
    """
    session = mock.Mock(PipSession)
    session.head.return_value = mock.Mock(
        **{
            "request.method": "HEAD",
            "headers": {"Content-Type": "text/html"},
        }
    )
    session.get.return_value = mock.Mock(headers={"Content-Type": "text/html"})

    resp = _get_html_response(url, session=session)

    assert resp is not None
    assert session.mock_calls == [
        mock.call.head(url, allow_redirects=True),
        mock.call.get(
            url,
            headers={
                "Accept": "text/html",
                "Cache-Control": "max-age=0",
            },
        ),
    ]
    assert mock_raise_for_status.mock_calls == [
        mock.call(session.head.return_value),
        mock.call(resp),
    ]


@pytest.mark.parametrize(
    "url",
    [
        "https://pypi.org/simple/pip",
        "https://pypi.org/simple/pip/",
        "https://python.org/sitemap.xml",
    ],
)
@mock.patch("pip._internal.index.collector.raise_for_status")
def test_get_html_response_no_head(mock_raise_for_status: mock.Mock, url: str) -> None:
    """
    `_get_html_response()` shouldn't send a HEAD request if the URL does not
    look like an archive, only the GET request that retrieves data.
    """
    session = mock.Mock(PipSession)

    # Mock the headers dict to ensure it is accessed.
    session.get.return_value = mock.Mock(
        headers=mock.Mock(
            **{
                "get.return_value": "text/html",
            }
        )
    )

    resp = _get_html_response(url, session=session)

    assert resp is not None
    assert session.head.call_count == 0
    assert session.get.mock_calls == [
        mock.call(
            url,
            headers={
                "Accept": "text/html",
                "Cache-Control": "max-age=0",
            },
        ),
        mock.call().headers.get("Content-Type", ""),
    ]
    mock_raise_for_status.assert_called_once_with(resp)


@mock.patch("pip._internal.index.collector.raise_for_status")
def test_get_html_response_dont_log_clear_text_password(
    mock_raise_for_status: mock.Mock, caplog: pytest.LogCaptureFixture
) -> None:
    """
    `_get_html_response()` should redact the password from the index URL
    in its DEBUG log message.
    """
    session = mock.Mock(PipSession)

    # Mock the headers dict to ensure it is accessed.
    session.get.return_value = mock.Mock(
        headers=mock.Mock(
            **{
                "get.return_value": "text/html",
            }
        )
    )

    caplog.set_level(logging.DEBUG)

    resp = _get_html_response(
        "https://user:my_password@example.com/simple/", session=session
    )

    assert resp is not None
    mock_raise_for_status.assert_called_once_with(resp)

    assert len(caplog.records) == 1
    record = caplog.records[0]
    assert record.levelname == "DEBUG"
    assert record.message.splitlines() == [
        "Getting page https://user:****@example.com/simple/",
    ]


@pytest.mark.parametrize(
    ("html", "url", "expected"),
    [
        (b"<html></html>", "https://example.com/", "https://example.com/"),
        (
            b'<html><head><base href="https://foo.example.com/"></head></html>',
            "https://example.com/",
            "https://foo.example.com/",
        ),
        (
            b"<html><head>"
            b'<base><base href="https://foo.example.com/">'
            b"</head></html>",
            "https://example.com/",
            "https://foo.example.com/",
        ),
    ],
)
def test_determine_base_url(html: bytes, url: str, expected: str) -> None:
    document = html5lib.parse(
        html,
        transport_encoding=None,
        namespaceHTMLElements=False,
    )
    assert _determine_base_url(document, url) == expected


@pytest.mark.parametrize(
    ("path", "expected"),
    [
        # Test a character that needs quoting.
        ("a b", "a%20b"),
        # Test an unquoted "@".
        ("a @ b", "a%20@%20b"),
        # Test multiple unquoted "@".
        ("a @ @ b", "a%20@%20@%20b"),
        # Test a quoted "@".
        ("a %40 b", "a%20%40%20b"),
        # Test a quoted "@" before an unquoted "@".
        ("a %40b@ c", "a%20%40b@%20c"),
        # Test a quoted "@" after an unquoted "@".
        ("a @b%40 c", "a%20@b%40%20c"),
        # Test alternating quoted and unquoted "@".
        ("a %40@b %40@c %40", "a%20%40@b%20%40@c%20%40"),
        # Test an unquoted "/".
        ("a / b", "a%20/%20b"),
        # Test multiple unquoted "/".
        ("a / / b", "a%20/%20/%20b"),
        # Test a quoted "/".
        ("a %2F b", "a%20%2F%20b"),
        # Test a quoted "/" before an unquoted "/".
        ("a %2Fb/ c", "a%20%2Fb/%20c"),
        # Test a quoted "/" after an unquoted "/".
        ("a /b%2F c", "a%20/b%2F%20c"),
        # Test alternating quoted and unquoted "/".
        ("a %2F/b %2F/c %2F", "a%20%2F/b%20%2F/c%20%2F"),
        # Test normalizing non-reserved quoted characters "[" and "]"
        ("a %5b %5d b", "a%20%5B%20%5D%20b"),
        # Test normalizing a reserved quoted "/"
        ("a %2f b", "a%20%2F%20b"),
    ],
)
@pytest.mark.parametrize("is_local_path", [True, False])
def test_clean_url_path(path: str, expected: str, is_local_path: bool) -> None:
    assert _clean_url_path(path, is_local_path=is_local_path) == expected


@pytest.mark.parametrize(
    ("path", "expected"),
    [
        # Test a VCS path with a Windows drive letter and revision.
        pytest.param(
            "/T:/with space/repo.git@1.0",
            "///T:/with%20space/repo.git@1.0",
            marks=pytest.mark.skipif("sys.platform != 'win32'"),
        ),
        # Test a VCS path with a Windows drive letter and revision,
        # running on non-windows platform.
        pytest.param(
            "/T:/with space/repo.git@1.0",
            "/T%3A/with%20space/repo.git@1.0",
            marks=pytest.mark.skipif("sys.platform == 'win32'"),
        ),
    ],
)
def test_clean_url_path_with_local_path(path: str, expected: str) -> None:
    actual = _clean_url_path(path, is_local_path=True)
    assert actual == expected


@pytest.mark.parametrize(
    ("url", "clean_url"),
    [
        # URL with hostname and port. Port separator should not be quoted.
        (
            "https://localhost.localdomain:8181/path/with space/",
            "https://localhost.localdomain:8181/path/with%20space/",
        ),
        # URL that is already properly quoted. The quoting `%`
        # characters should not be quoted again.
        (
            "https://localhost.localdomain:8181/path/with%20quoted%20space/",
            "https://localhost.localdomain:8181/path/with%20quoted%20space/",
        ),
        # URL with IPv4 address and port.
        (
            "https://127.0.0.1:8181/path/with space/",
            "https://127.0.0.1:8181/path/with%20space/",
        ),
        # URL with IPv6 address and port. The `[]` brackets around the
        # IPv6 address should not be quoted.
        (
            "https://[fd00:0:0:236::100]:8181/path/with space/",
            "https://[fd00:0:0:236::100]:8181/path/with%20space/",
        ),
        # URL with query. The leading `?` should not be quoted.
        (
            "https://localhost.localdomain:8181/path/with/query?request=test",
            "https://localhost.localdomain:8181/path/with/query?request=test",
        ),
        # URL with colon in the path portion.
        (
            "https://localhost.localdomain:8181/path:/with:/colon",
            "https://localhost.localdomain:8181/path%3A/with%3A/colon",
        ),
        # URL with something that looks like a drive letter, but is
        # not. The `:` should be quoted.
        (
            "https://localhost.localdomain/T:/path/",
            "https://localhost.localdomain/T%3A/path/",
        ),
        # URL with a quoted "/" in the path portion.
        (
            "https://example.com/access%2Ftoken/path/",
            "https://example.com/access%2Ftoken/path/",
        ),
        # VCS URL containing revision string.
        (
            "git+ssh://example.com/path to/repo.git@1.0#egg=my-package-1.0",
            "git+ssh://example.com/path%20to/repo.git@1.0#egg=my-package-1.0",
        ),
        # VCS URL with a quoted "#" in the revision string.
        (
            "git+https://example.com/repo.git@hash%23symbol#egg=my-package-1.0",
            "git+https://example.com/repo.git@hash%23symbol#egg=my-package-1.0",
        ),
        # VCS URL with a quoted "@" in the revision string.
        (
            "git+https://example.com/repo.git@at%40 space#egg=my-package-1.0",
            "git+https://example.com/repo.git@at%40%20space#egg=my-package-1.0",
        ),
        # URL with Windows drive letter. The `:` after the drive
        # letter should not be quoted. The trailing `/` should be
        # removed.
        pytest.param(
            "file:///T:/path/with spaces/",
            "file:///T:/path/with%20spaces",
            marks=pytest.mark.skipif("sys.platform != 'win32'"),
        ),
        # URL with Windows drive letter, running on non-windows
        # platform. The `:` after the drive should be quoted.
        pytest.param(
            "file:///T:/path/with spaces/",
            "file:///T%3A/path/with%20spaces/",
            marks=pytest.mark.skipif("sys.platform == 'win32'"),
        ),
        # Test a VCS URL with a Windows drive letter and revision.
        pytest.param(
            "git+file:///T:/with space/repo.git@1.0#egg=my-package-1.0",
            "git+file:///T:/with%20space/repo.git@1.0#egg=my-package-1.0",
            marks=pytest.mark.skipif("sys.platform != 'win32'"),
        ),
        # Test a VCS URL with a Windows drive letter and revision,
        # running on non-windows platform.
        pytest.param(
            "git+file:///T:/with space/repo.git@1.0#egg=my-package-1.0",
            "git+file:/T%3A/with%20space/repo.git@1.0#egg=my-package-1.0",
            marks=pytest.mark.skipif("sys.platform == 'win32'"),
        ),
    ],
)
def test_clean_link(url: str, clean_url: str) -> None:
    assert _clean_link(url) == clean_url


def _test_parse_links_data_attribute(
    anchor_html: str, attr: str, expected: Optional[str]
) -> None:
    html = (
        "<!DOCTYPE html>"
        '<html><head><meta charset="utf-8"><head>'
        "<body>{}</body></html>"
    ).format(anchor_html)
    html_bytes = html.encode("utf-8")
    page = HTMLPage(
        html_bytes,
        encoding=None,
        # parse_links() is cached by url, so we inject a random uuid to ensure
        # the page content isn't cached.
        url=f"https://example.com/simple-{uuid.uuid4()}/",
    )
    links = list(parse_links(page, use_deprecated_html5lib=False))
    (link,) = links
    actual = getattr(link, attr)
    assert actual == expected


@pytest.mark.parametrize(
    "anchor_html, expected",
    [
        # Test not present.
        ('<a href="/pkg-1.0.tar.gz"></a>', None),
        # Test present with no value.
        ('<a href="/pkg-1.0.tar.gz" data-requires-python></a>', None),
        # Test a value with an escaped character.
        (
            '<a href="/pkg-1.0.tar.gz" data-requires-python="&gt;=3.6"></a>',
            ">=3.6",
        ),
        # Test requires python is unescaped once.
        (
            '<a href="/pkg-1.0.tar.gz" data-requires-python="&amp;gt;=3.6"></a>',
            "&gt;=3.6",
        ),
    ],
)
def test_parse_links__requires_python(
    anchor_html: str, expected: Optional[str]
) -> None:
    _test_parse_links_data_attribute(anchor_html, "requires_python", expected)


@pytest.mark.parametrize(
    "anchor_html, expected",
    [
        # Test not present.
        ('<a href="/pkg1-1.0.tar.gz"></a>', None),
        # Test present with no value.
        ('<a href="/pkg2-1.0.tar.gz" data-yanked></a>', None),
        # Test the empty string.
        ('<a href="/pkg3-1.0.tar.gz" data-yanked=""></a>', ""),
        # Test a non-empty string.
        ('<a href="/pkg4-1.0.tar.gz" data-yanked="error"></a>', "error"),
        # Test a value with an escaped character.
        ('<a href="/pkg4-1.0.tar.gz" data-yanked="version &lt 1"></a>', "version < 1"),
        # Test a yanked reason with a non-ascii character.
        (
            '<a href="/pkg-1.0.tar.gz" data-yanked="curlyquote \u2018"></a>',
            "curlyquote \u2018",
        ),
        # Test yanked reason is unescaped once.
        (
            '<a href="/pkg-1.0.tar.gz" data-yanked="version &amp;lt; 1"></a>',
            "version &lt; 1",
        ),
    ],
)
def test_parse_links__yanked_reason(anchor_html: str, expected: Optional[str]) -> None:
    _test_parse_links_data_attribute(anchor_html, "yanked_reason", expected)


def test_parse_links_caches_same_page_by_url() -> None:
    html = (
        "<!DOCTYPE html>"
        '<html><head><meta charset="utf-8"><head>'
        '<body><a href="/pkg1-1.0.tar.gz"></a></body></html>'
    )
    html_bytes = html.encode("utf-8")

    url = "https://example.com/simple/"

    page_1 = HTMLPage(
        html_bytes,
        encoding=None,
        url=url,
    )
    # Make a second page with zero content, to ensure that it's not accessed,
    # because the page was cached by url.
    page_2 = HTMLPage(
        b"",
        encoding=None,
        url=url,
    )
    # Make a third page which represents an index url, which should not be
    # cached, even for the same url. We modify the page content slightly to
    # verify that the result is not cached.
    page_3 = HTMLPage(
        re.sub(b"pkg1", b"pkg2", html_bytes),
        encoding=None,
        url=url,
        cache_link_parsing=False,
    )

    parsed_links_1 = list(parse_links(page_1, use_deprecated_html5lib=False))
    assert len(parsed_links_1) == 1
    assert "pkg1" in parsed_links_1[0].url

    parsed_links_2 = list(parse_links(page_2, use_deprecated_html5lib=False))
    assert parsed_links_2 == parsed_links_1

    parsed_links_3 = list(parse_links(page_3, use_deprecated_html5lib=False))
    assert len(parsed_links_3) == 1
    assert parsed_links_3 != parsed_links_1
    assert "pkg2" in parsed_links_3[0].url


def test_parse_link_handles_deprecated_usage_properly() -> None:
    html = b'<a href="/pkg1-1.0.tar.gz"></a><a href="/pkg1-2.0.tar.gz"></a>'
    url = "https://example.com/simple/"
    page = HTMLPage(html, encoding=None, url=url, cache_link_parsing=False)

    parsed_links = list(parse_links(page, use_deprecated_html5lib=True))

    assert len(parsed_links) == 2
    assert "pkg1-1.0" in parsed_links[0].url
    assert "pkg1-2.0" in parsed_links[1].url


@mock.patch("pip._internal.index.collector.deprecated")
def test_parse_links_presents_deprecation_warning_on_non_html5_page(
    mock_deprecated: mock.Mock,
) -> None:
    html = b'<a href="/pkg1-1.0.tar.gz"></a><a href="/pkg1-2.0.tar.gz"></a>'
    url = "https://example.com/simple/"
    page = HTMLPage(html, encoding=None, url=url, cache_link_parsing=False)

    parsed_links = list(parse_links(page, use_deprecated_html5lib=False))

    assert len(parsed_links) == 2, parsed_links
    assert "pkg1-1.0" in parsed_links[0].url
    assert "pkg1-2.0" in parsed_links[1].url

    mock_deprecated.assert_called_once()


@mock.patch("pip._internal.index.collector.raise_for_status")
def test_request_http_error(
    mock_raise_for_status: mock.Mock, caplog: pytest.LogCaptureFixture
) -> None:
    caplog.set_level(logging.DEBUG)
    link = Link("http://localhost")
    session = mock.Mock(PipSession)
    session.get.return_value = mock.Mock()
    mock_raise_for_status.side_effect = NetworkConnectionError("Http error")
    assert _get_html_page(link, session=session) is None
    assert "Could not fetch URL http://localhost: Http error - skipping" in caplog.text


def test_request_retries(caplog: pytest.LogCaptureFixture) -> None:
    caplog.set_level(logging.DEBUG)
    link = Link("http://localhost")
    session = mock.Mock(PipSession)
    session.get.side_effect = requests.exceptions.RetryError("Retry error")
    assert _get_html_page(link, session=session) is None
    assert "Could not fetch URL http://localhost: Retry error - skipping" in caplog.text


def test_make_html_page() -> None:
    headers = {"Content-Type": "text/html; charset=UTF-8"}
    response = mock.Mock(
        content=b"<content>",
        url="https://example.com/index.html",
        headers=headers,
    )

    actual = _make_html_page(response)
    assert actual.content == b"<content>"
    assert actual.encoding == "UTF-8"
    assert actual.url == "https://example.com/index.html"


@pytest.mark.parametrize(
    "url, vcs_scheme",
    [
        ("svn+http://pypi.org/something", "svn"),
        ("git+https://github.com/pypa/pip.git", "git"),
    ],
)
def test_get_html_page_invalid_scheme(
    caplog: pytest.LogCaptureFixture, url: str, vcs_scheme: str
) -> None:
    """`_get_html_page()` should error if an invalid scheme is given.

    Only file:, http:, https:, and ftp: are allowed.
    """
    with caplog.at_level(logging.WARNING):
        page = _get_html_page(Link(url), session=mock.Mock(PipSession))

    assert page is None
    assert caplog.record_tuples == [
        (
            "pip._internal.index.collector",
            logging.WARNING,
            "Cannot look at {} URL {} because it does not support "
            "lookup as web pages.".format(vcs_scheme, url),
        ),
    ]


@pytest.mark.parametrize(
    "content_type",
    [
        "application/xhtml+xml",
        "application/json",
    ],
)
@mock.patch("pip._internal.index.collector.raise_for_status")
def test_get_html_page_invalid_content_type(
    mock_raise_for_status: mock.Mock,
    caplog: pytest.LogCaptureFixture,
    content_type: str,
) -> None:
    """`_get_html_page()` should warn if an invalid content-type is given.
    Only text/html is allowed.
    """
    caplog.set_level(logging.DEBUG)
    url = "https://pypi.org/simple/pip"
    link = Link(url)

    session = mock.Mock(PipSession)
    session.get.return_value = mock.Mock(
        **{
            "request.method": "GET",
            "headers": {"Content-Type": content_type},
        }
    )
    assert _get_html_page(link, session=session) is None
    mock_raise_for_status.assert_called_once_with(session.get.return_value)
    assert (
        "pip._internal.index.collector",
        logging.WARNING,
        "Skipping page {} because the GET request got Content-Type: {}."
        "The only supported Content-Type is text/html".format(url, content_type),
    ) in caplog.record_tuples


def make_fake_html_response(url: str) -> mock.Mock:
    """
    Create a fake requests.Response object.
    """
    html = dedent(
        """\
    <html><head><meta name="api-version" value="2" /></head>
    <body>
    <a href="/abc-1.0.tar.gz#md5=000000000">abc-1.0.tar.gz</a>
    </body></html>
    """
    )
    content = html.encode("utf-8")
    return mock.Mock(content=content, url=url, headers={})


def test_get_html_page_directory_append_index(tmpdir: Path) -> None:
    """`_get_html_page()` should append "index.html" to a directory URL."""
    dirpath = tmpdir / "something"
    dirpath.mkdir()
    dir_url = "file:///{}".format(
        urllib.request.pathname2url(dirpath).lstrip("/"),
    )
    expected_url = "{}/index.html".format(dir_url.rstrip("/"))

    session = mock.Mock(PipSession)
    fake_response = make_fake_html_response(expected_url)
    mock_func = mock.patch("pip._internal.index.collector._get_html_response")
    with mock_func as mock_func:
        mock_func.return_value = fake_response
        actual = _get_html_page(Link(dir_url), session=session)
        assert mock_func.mock_calls == [
            mock.call(expected_url, session=session),
        ], f"actual calls: {mock_func.mock_calls}"

        assert actual is not None
        assert actual.content == fake_response.content
        assert actual.encoding is None
        assert actual.url == expected_url


def test_collect_sources__file_expand_dir(data: TestData) -> None:
    """
    Test that a file:// dir from --find-links becomes _FlatDirectorySource
    """
    collector = LinkCollector.create(
        session=mock.Mock(is_secure_origin=None),  # Shouldn't be used.
        options=mock.Mock(
            index_url="ignored-by-no-index",
            extra_index_urls=[],
            no_index=True,
            find_links=[data.find_links],
        ),
    )
    sources = collector.collect_sources(
        # Shouldn't be used.
        project_name=None,  # type: ignore[arg-type]
        candidates_from_page=None,  # type: ignore[arg-type]
    )
    assert (
        not sources.index_urls
        and len(sources.find_links) == 1
        and isinstance(sources.find_links[0], _FlatDirectorySource)
    ), (
        "Directory source should have been found "
        f"at find-links url: {data.find_links}"
    )


def test_collect_sources__file_not_find_link(data: TestData) -> None:
    """
    Test that a file:// dir from --index-url doesn't become _FlatDirectorySource
    run
    """
    collector = LinkCollector.create(
        session=mock.Mock(is_secure_origin=None),  # Shouldn't be used.
        options=mock.Mock(
            index_url=data.index_url("empty_with_pkg"),
            extra_index_urls=[],
            no_index=False,
            find_links=[],
        ),
    )
    sources = collector.collect_sources(
        project_name="",
        # Shouldn't be used.
        candidates_from_page=None,  # type: ignore[arg-type]
    )
    assert (
        not sources.find_links
        and len(sources.index_urls) == 1
        and isinstance(sources.index_urls[0], _IndexDirectorySource)
    ), "Directory specified as index should be treated as a page"


def test_collect_sources__non_existing_path() -> None:
    """
    Test that a non-existing path is ignored.
    """
    collector = LinkCollector.create(
        session=mock.Mock(is_secure_origin=None),  # Shouldn't be used.
        options=mock.Mock(
            index_url="ignored-by-no-index",
            extra_index_urls=[],
            no_index=True,
            find_links=[os.path.join("this", "doesnt", "exist")],
        ),
    )
    sources = collector.collect_sources(
        # Shouldn't be used.
        project_name=None,  # type: ignore[arg-type]
        candidates_from_page=None,  # type: ignore[arg-type]
    )
    assert not sources.index_urls and sources.find_links == [
        None
    ], "Nothing should have been found"


def check_links_include(links: List[Link], names: List[str]) -> None:
    """
    Assert that the given list of Link objects includes, for each of the
    given names, a link whose URL has a base name matching that name.
    """
    for name in names:
        assert any(
            link.url.endswith(name) for link in links
        ), f"name {name!r} not among links: {links}"


class TestLinkCollector:
    @mock.patch("pip._internal.index.collector._get_html_response")
    def test_fetch_page(self, mock_get_html_response: mock.Mock) -> None:
        url = "https://pypi.org/simple/twine/"

        fake_response = make_fake_html_response(url)
        mock_get_html_response.return_value = fake_response

        location = Link(url, cache_link_parsing=False)
        link_collector = make_test_link_collector()
        actual = link_collector.fetch_page(location)

        assert actual is not None
        assert actual.content == fake_response.content
        assert actual.encoding is None
        assert actual.url == url
        assert actual.cache_link_parsing == location.cache_link_parsing

        # Also check that the right session object was passed to
        # _get_html_response().
        mock_get_html_response.assert_called_once_with(
            url,
            session=link_collector.session,
        )

    def test_collect_sources(
        self, caplog: pytest.LogCaptureFixture, data: TestData
    ) -> None:
        caplog.set_level(logging.DEBUG)

        link_collector = make_test_link_collector(
            find_links=[data.find_links],
            # Include two copies of the URL to check that the second one
            # is skipped.
            index_urls=[PyPI.simple_url, PyPI.simple_url],
        )
        collected_sources = link_collector.collect_sources(
            "twine",
            candidates_from_page=lambda link: [
                InstallationCandidate("twine", "1.0", link)
            ],
        )

        files_it = itertools.chain.from_iterable(
            source.file_links()
            for sources in collected_sources
            for source in sources
            if source is not None
        )
        pages_it = itertools.chain.from_iterable(
            source.page_candidates()
            for sources in collected_sources
            for source in sources
            if source is not None
        )
        files = list(files_it)
        pages = list(pages_it)

        # Spot-check the returned sources.
        assert len(files) > 20
        check_links_include(files, names=["simple-1.0.tar.gz"])

        assert [page.link for page in pages] == [Link("https://pypi.org/simple/twine/")]
        # Check that index URLs are marked as *un*cacheable.
        assert not pages[0].link.cache_link_parsing

        expected_message = dedent(
            """\
        1 location(s) to search for versions of twine:
        * https://pypi.org/simple/twine/"""
        )
        assert caplog.record_tuples == [
            ("pip._internal.index.collector", logging.DEBUG, expected_message),
        ]


@pytest.mark.parametrize(
    "find_links, no_index, suppress_no_index, expected",
    [
        (["link1"], False, False, (["link1"], ["default_url", "url1", "url2"])),
        (["link1"], False, True, (["link1"], ["default_url", "url1", "url2"])),
        (["link1"], True, False, (["link1"], [])),
        # Passing suppress_no_index=True suppresses no_index=True.
        (["link1"], True, True, (["link1"], ["default_url", "url1", "url2"])),
        # Test options.find_links=False.
        (False, False, False, ([], ["default_url", "url1", "url2"])),
    ],
)
def test_link_collector_create(
    find_links: List[str],
    no_index: bool,
    suppress_no_index: bool,
    expected: Tuple[List[str], List[str]],
) -> None:
    """
    :param expected: the expected (find_links, index_urls) values.
    """
    expected_find_links, expected_index_urls = expected
    session = PipSession()
    options = mock.Mock(
        find_links=find_links,
        index_url="default_url",
        extra_index_urls=["url1", "url2"],
        no_index=no_index,
    )
    link_collector = LinkCollector.create(
        session,
        options=options,
        suppress_no_index=suppress_no_index,
    )

    assert link_collector.session is session

    search_scope = link_collector.search_scope
    assert search_scope.find_links == expected_find_links
    assert search_scope.index_urls == expected_index_urls


@mock.patch("os.path.expanduser")
def test_link_collector_create_find_links_expansion(
    mock_expanduser: mock.Mock, tmpdir: Path
) -> None:
    """
    Test "~" expansion in --find-links paths.
    """
    # This is a mock version of expanduser() that expands "~" to the tmpdir.
    def expand_path(path: str) -> str:
        if path.startswith("~/"):
            path = os.path.join(tmpdir, path[2:])
        return path

    mock_expanduser.side_effect = expand_path

    session = PipSession()
    options = mock.Mock(
        find_links=["~/temp1", "~/temp2"],
        index_url="default_url",
        extra_index_urls=[],
        no_index=False,
    )
    # Only create temp2 and not temp1 to test that "~" expansion only occurs
    # when the directory exists.
    temp2_dir = os.path.join(tmpdir, "temp2")
    os.mkdir(temp2_dir)

    link_collector = LinkCollector.create(session, options=options)

    search_scope = link_collector.search_scope
    # Only ~/temp2 gets expanded. Also, the path is normalized when expanded.
    expected_temp2_dir = os.path.normcase(temp2_dir)
    assert search_scope.find_links == ["~/temp1", expected_temp2_dir]
    assert search_scope.index_urls == ["default_url"]