Repository: pypa/pip
File: tests/unit/test_network_download.py (Python)

Code before:
    link = Link("http://example.com/foo.tgz")
    downloader = Downloader(session, "on", resume_retries=5)

    # Mock an incomplete download followed by a successful resume
    incomplete_resp = MockResponse(b"0cfa7e9d-1868-4dd7-9fb3-")
    incomplete_resp.headers = {"content-length": "36"}
    incomplete_resp.status_code = 200

    resume_resp = MockResponse(b"f2561d5dfd89")
    resume_resp.headers = {"content-length": "12"}
    resume_resp.status_code = 206

    responses = [incomplete_resp, resume_resp]
    _http_get_mock = MagicMock(side_effect=responses)

    # Mock the session's adapters to have a cache controller
    mock_adapter = MagicMock(spec=CacheControlAdapter)
    mock_controller = MagicMock()
    mock_adapter.controller = mock_controller
    mock_controller.cache_url = MagicMock(return_value="cache_key")
    mock_controller.serializer = MagicMock()
    mock_controller.serializer.dumps = MagicMock(return_value=b"serialized_data")

    # Mock the cache to be a SafeFileCache
    mock_cache = MagicMock(spec=SafeFileCache)
    mock_adapter.cache = mock_cache

    # Create a mock for the session adapters
    adapters_mock = MagicMock()
    adapters_mock.__getitem__ = MagicMock(return_value=mock_adapter)

    with (
        patch.object(Downloader, "_http_get", _http_get_mock),
        patch.object(session, "adapters", adapters_mock),
    ):

        filepath, _ = downloader(link, str(tmpdir))

        # Verify the file was downloaded correctly
        with open(filepath, "rb") as downloaded_file:
            downloaded_bytes = downloaded_file.read()
            expected_bytes = b"0cfa7e9d-1868-4dd7-9fb3-f2561d5dfd89"
            assert downloaded_bytes == expected_bytes

        # Verify that cache.set was called for metadata
        mock_cache.set.assert_called_once()

        # Verify that set_body_from_io was called for streaming the body
        mock_cache.set_body_from_io.assert_called_once()

        # Verify the call arguments

Review comment (performance):
Could you just not create a pip session with caching already enabled? `PipSession(cache=str(tmpdir))` I get this is an unit test, but practically, I'd prefer if the we tested the actual implementation to avoid mismatches that hide issues down the line.

Code after:
    link = Link("https://example.com/foo.tgz")
    downloader = Downloader(session, "on", resume_retries=5)

    # Mock an incomplete download followed by a successful resume
    incomplete_resp = MockResponse(b"0cfa7e9d-1868-4dd7-9fb3-")
    incomplete_resp.headers = {"content-length": "36"}
    incomplete_resp.status_code = 200

    resume_resp = MockResponse(b"f2561d5dfd89")
    resume_resp.headers = {"content-length": "12"}
    resume_resp.status_code = 206

    responses = [incomplete_resp, resume_resp]
    _http_get_mock = MagicMock(side_effect=responses)

    with patch.object(Downloader, "_http_get", _http_get_mock):
        # Perform the download (incomplete then resumed)
        filepath, _ = downloader(link, str(tmpdir))

        # Verify the file was downloaded correctly
        with open(filepath, "rb") as downloaded_file:
            downloaded_bytes = downloaded_file.read()
            expected_bytes = b"0cfa7e9d-1868-4dd7-9fb3-f2561d5dfd89"
            assert downloaded_bytes == expected_bytes

        # Verify that the cache directory was created and contains cache files
        # The resumed download should have been cached for future use
        assert cache_dir.exists()
        cache_files = list(cache_dir.rglob("*"))
        # Should have cache files (both metadata and body files)
        assert len([f for f in cache_files if f.is_file()]) == 2