summaryrefslogtreecommitdiff
path: root/tests/unit/test_network_session.py
blob: 044d1fb923b79a950dd6ff5bbb6b88e9cc91a7b8 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
import logging

import pytest

from pip import __version__
from pip._internal.network.session import CI_ENVIRONMENT_VARIABLES, PipSession


def get_user_agent():
    return PipSession().headers["User-Agent"]


def test_user_agent():
    user_agent = get_user_agent()

    assert user_agent.startswith(f"pip/{__version__}")


@pytest.mark.parametrize('name, expected_like_ci', [
    ('BUILD_BUILDID', True),
    ('BUILD_ID', True),
    ('CI', True),
    ('PIP_IS_CI', True),
    # Test a prefix substring of one of the variable names we use.
    ('BUILD', False),
])
def test_user_agent__ci(monkeypatch, name, expected_like_ci):
    # Delete the variable names we use to check for CI to prevent the
    # detection from always returning True in case the tests are being run
    # under actual CI.  It is okay to depend on CI_ENVIRONMENT_VARIABLES
    # here (part of the code under test) because this setup step can only
    # prevent false test failures.  It can't cause a false test passage.
    for ci_name in CI_ENVIRONMENT_VARIABLES:
        monkeypatch.delenv(ci_name, raising=False)

    # Confirm the baseline before setting the environment variable.
    user_agent = get_user_agent()
    assert '"ci":null' in user_agent
    assert '"ci":true' not in user_agent

    monkeypatch.setenv(name, 'true')
    user_agent = get_user_agent()
    assert ('"ci":true' in user_agent) == expected_like_ci
    assert ('"ci":null' in user_agent) == (not expected_like_ci)


def test_user_agent_user_data(monkeypatch):
    monkeypatch.setenv("PIP_USER_AGENT_USER_DATA", "some_string")
    assert "some_string" in PipSession().headers["User-Agent"]


class TestPipSession:

    def test_cache_defaults_off(self):
        session = PipSession()

        assert not hasattr(session.adapters["http://"], "cache")
        assert not hasattr(session.adapters["https://"], "cache")

    def test_cache_is_enabled(self, tmpdir):
        cache_directory = tmpdir.joinpath("test-cache")
        session = PipSession(cache=cache_directory)

        assert hasattr(session.adapters["https://"], "cache")

        assert (
            session.adapters["https://"].cache.directory == cache_directory
        )

    def test_http_cache_is_not_enabled(self, tmpdir):
        session = PipSession(cache=tmpdir.joinpath("test-cache"))

        assert not hasattr(session.adapters["http://"], "cache")

    def test_trusted_hosts_adapter(self, tmpdir):
        session = PipSession(
            cache=tmpdir.joinpath("test-cache"),
            trusted_hosts=["example.com"],
        )

        assert "https://example.com/" in session.adapters
        # Check that the "port wildcard" is present.
        assert "https://example.com:" in session.adapters
        # Check that the cache is enabled.
        assert hasattr(session.adapters["https://example.com/"], "cache")

    def test_add_trusted_host(self):
        # Leave a gap to test how the ordering is affected.
        trusted_hosts = ['host1', 'host3']
        session = PipSession(trusted_hosts=trusted_hosts)
        trusted_host_adapter = session._trusted_host_adapter
        prefix2 = 'https://host2/'
        prefix3 = 'https://host3/'
        prefix3_wildcard = 'https://host3:'

        # Confirm some initial conditions as a baseline.
        assert session.pip_trusted_origins == [
            ('host1', None), ('host3', None)
        ]
        assert session.adapters[prefix3] is trusted_host_adapter
        assert session.adapters[prefix3_wildcard] is trusted_host_adapter

        assert prefix2 not in session.adapters

        # Test adding a new host.
        session.add_trusted_host('host2')
        assert session.pip_trusted_origins == [
            ('host1', None), ('host3', None), ('host2', None)
        ]
        # Check that prefix3 is still present.
        assert session.adapters[prefix3] is trusted_host_adapter
        assert session.adapters[prefix2] is trusted_host_adapter

        # Test that adding the same host doesn't create a duplicate.
        session.add_trusted_host('host3')
        assert session.pip_trusted_origins == [
            ('host1', None), ('host3', None), ('host2', None)
        ], f'actual: {session.pip_trusted_origins}'

        session.add_trusted_host('host4:8080')
        prefix4 = 'https://host4:8080/'
        assert session.pip_trusted_origins == [
            ('host1', None), ('host3', None),
            ('host2', None), ('host4', 8080)
        ]
        assert session.adapters[prefix4] is trusted_host_adapter

    def test_add_trusted_host__logging(self, caplog):
        """
        Test logging when add_trusted_host() is called.
        """
        trusted_hosts = ['host0', 'host1']
        session = PipSession(trusted_hosts=trusted_hosts)
        with caplog.at_level(logging.INFO):
            # Test adding an existing host.
            session.add_trusted_host('host1', source='somewhere')
            session.add_trusted_host('host2')
            # Test calling add_trusted_host() on the same host twice.
            session.add_trusted_host('host2')

        actual = [(r.levelname, r.message) for r in caplog.records]
        # Observe that "host0" isn't included in the logs.
        expected = [
            ('INFO', "adding trusted host: 'host1' (from somewhere)"),
            ('INFO', "adding trusted host: 'host2'"),
            ('INFO', "adding trusted host: 'host2'"),
        ]
        assert actual == expected

    def test_iter_secure_origins(self):
        trusted_hosts = ['host1', 'host2', 'host3:8080']
        session = PipSession(trusted_hosts=trusted_hosts)

        actual = list(session.iter_secure_origins())
        assert len(actual) == 9
        # Spot-check that SECURE_ORIGINS is included.
        assert actual[0] == ('https', '*', '*')
        assert actual[-3:] == [
            ('*', 'host1', '*'),
            ('*', 'host2', '*'),
            ('*', 'host3', 8080)
        ]

    def test_iter_secure_origins__trusted_hosts_empty(self):
        """
        Test iter_secure_origins() after passing trusted_hosts=[].
        """
        session = PipSession(trusted_hosts=[])

        actual = list(session.iter_secure_origins())
        assert len(actual) == 6
        # Spot-check that SECURE_ORIGINS is included.
        assert actual[0] == ('https', '*', '*')

    @pytest.mark.parametrize(
        'location, trusted, expected',
        [
            ("http://pypi.org/something", [], False),
            ("https://pypi.org/something", [], True),
            ("git+http://pypi.org/something", [], False),
            ("git+https://pypi.org/something", [], True),
            ("git+ssh://git@pypi.org/something", [], True),
            ("http://localhost", [], True),
            ("http://127.0.0.1", [], True),
            ("http://example.com/something/", [], False),
            ("http://example.com/something/", ["example.com"], True),
            # Try changing the case.
            ("http://eXample.com/something/", ["example.cOm"], True),
            # Test hosts with port.
            ("http://example.com:8080/something/", ["example.com"], True),
            # Test a trusted_host with a port.
            ("http://example.com:8080/something/", ["example.com:8080"], True),
            ("http://example.com/something/", ["example.com:8080"], False),
            (
                "http://example.com:8888/something/",
                ["example.com:8080"],
                False
            ),
        ],
    )
    def test_is_secure_origin(self, caplog, location, trusted, expected):
        class MockLogger:
            def __init__(self):
                self.called = False

            def warning(self, *args, **kwargs):
                self.called = True

        session = PipSession(trusted_hosts=trusted)
        actual = session.is_secure_origin(location)
        assert actual == expected

        log_records = [(r.levelname, r.message) for r in caplog.records]
        if expected:
            assert not log_records
            return

        assert len(log_records) == 1
        actual_level, actual_message = log_records[0]
        assert actual_level == 'WARNING'
        assert 'is not a trusted or secure host' in actual_message