fix: pinned version are not working once periodic / manual updates kick-in

Fixes pypa#2203

docs/changelog/2203.bugfix.rst

@@ -0,0 +1 @@
+Fix installation of pinned versions of ``pip``, ``setuptools`` & ``wheel`` - by :user:`mayeut`.

src/virtualenv/seed/wheels/bundle.py

@@ -15,7 +15,9 @@ def from_bundle(distribution, version, for_py_version, search_dirs, app_data, do
     if version != Version.embed:
         # 2. check if we have upgraded embed
         if app_data.can_update:
-            wheel = periodic_update(distribution, for_py_version, wheel, search_dirs, app_data, do_periodic_update, env)
+            wheel = periodic_update(
+                distribution, of_version, for_py_version, wheel, search_dirs, app_data, do_periodic_update, env
+            )
 
         # 3. acquire from extra search dir
         found_wheel = from_dir(distribution, of_version, for_py_version, search_dirs)

src/virtualenv/seed/wheels/periodic_update.py

@@ -36,7 +36,7 @@ if PY2:
         pass  # pragma: no cov
 
 
-def periodic_update(distribution, for_py_version, wheel, search_dirs, app_data, do_periodic_update, env):
+def periodic_update(distribution, of_version, for_py_version, wheel, search_dirs, app_data, do_periodic_update, env):
     if do_periodic_update:
         handle_auto_update(distribution, for_py_version, wheel, search_dirs, app_data, env)
 
@@ -44,15 +44,23 @@ def periodic_update(distribution, for_py_version, wheel, search_dirs, app_data,
 
     u_log = UpdateLog.from_app_data(app_data, distribution, for_py_version)
     u_log_older_than_hour = now - u_log.completed > timedelta(hours=1) if u_log.completed is not None else False
-    for _, group in groupby(u_log.versions, key=lambda v: v.wheel.version_tuple[0:2]):
-        version = next(group)  # use only latest patch version per minor, earlier assumed to be buggy
-        if wheel is not None and Path(version.filename).name == wheel.name:
-            break
-        if u_log.periodic is False or (u_log_older_than_hour and version.use(now)):
-            updated_wheel = Wheel(app_data.house / version.filename)
-            logging.debug("using %supdated wheel %s", "periodically " if updated_wheel else "", updated_wheel)
-            wheel = updated_wheel
-            break
+    if of_version is None:
+        for _, group in groupby(u_log.versions, key=lambda v: v.wheel.version_tuple[0:2]):
+            version = next(group)  # use only latest patch version per minor, earlier assumed to be buggy
+            if wheel is not None and Path(version.filename).name == wheel.name:
+                break
+            if u_log.periodic is False or (u_log_older_than_hour and version.use(now)):
+                updated_wheel = Wheel(app_data.house / version.filename)
+                logging.debug("using %supdated wheel %s", "periodically " if updated_wheel else "", updated_wheel)
+                wheel = updated_wheel
+                break
+    elif u_log.periodic is False or u_log_older_than_hour:
+        for version in u_log.versions:
+            if version.wheel.version == of_version:
+                updated_wheel = Wheel(app_data.house / version.filename)
+                logging.debug("using %supdated wheel %s", "periodically " if updated_wheel else "", updated_wheel)
+                wheel = updated_wheel
+                break
 
     return wheel
 

tests/unit/seed/wheels/test_bundle.py

@@ -0,0 +1,71 @@
+from __future__ import absolute_import, unicode_literals
+
+import os
+
+import pytest
+
+from virtualenv.app_data import AppDataDiskFolder
+from virtualenv.seed.wheels.bundle import from_bundle
+from virtualenv.seed.wheels.embed import get_embed_wheel
+from virtualenv.seed.wheels.util import Version, Wheel
+from virtualenv.util.path import Path
+
+
+@pytest.fixture(scope="module")
+def next_pip_wheel(for_py_version):
+    wheel = get_embed_wheel("pip", for_py_version)
+    new_version = list(wheel.version_tuple)
+    new_version[-1] += 1
+    new_name = wheel.name.replace(wheel.version, ".".join(str(i) for i in new_version))
+    return Wheel.from_path(Path(new_name))
+
+
+@pytest.fixture(scope="module")
+def app_data(tmp_path_factory, for_py_version, next_pip_wheel):
+    temp_folder = tmp_path_factory.mktemp("module-app-data")
+    app_data_ = AppDataDiskFolder(str(temp_folder))
+    app_data_.embed_update_log("pip", for_py_version).write(
+        {
+            "completed": "2000-01-01T00:00:00.000000Z",
+            "periodic": True,
+            "started": "2000-01-01T00:00:00.000000Z",
+            "versions": [
+                {
+                    "filename": next_pip_wheel.name,
+                    "found_date": "2000-01-01T00:00:00.000000Z",
+                    "release_date": "2000-01-01T00:00:00.000000Z",
+                }
+            ],
+        }
+    )
+    yield app_data_
+
+
+def test_version_embed(app_data, for_py_version):
+    wheel = from_bundle("pip", Version.embed, for_py_version, [], app_data, False, os.environ)
+    assert wheel is not None
+    assert wheel.name == get_embed_wheel("pip", for_py_version).name
+
+
+def test_version_bundle(app_data, for_py_version, next_pip_wheel):
+    wheel = from_bundle("pip", Version.bundle, for_py_version, [], app_data, False, os.environ)
+    assert wheel is not None
+    assert wheel.name == next_pip_wheel.name
+
+
+def test_version_pinned_not_found(app_data, for_py_version):
+    wheel = from_bundle("pip", "0.0.0", for_py_version, [], app_data, False, os.environ)
+    assert wheel is None
+
+
+def test_version_pinned_is_embed(app_data, for_py_version):
+    expected_wheel = get_embed_wheel("pip", for_py_version)
+    wheel = from_bundle("pip", expected_wheel.version, for_py_version, [], app_data, False, os.environ)
+    assert wheel is not None
+    assert wheel.name == expected_wheel.name
+
+
+def test_version_pinned_in_app_data(app_data, for_py_version, next_pip_wheel):
+    wheel = from_bundle("pip", next_pip_wheel.version, for_py_version, [], app_data, False, os.environ)
+    assert wheel is not None
+    assert wheel.name == next_pip_wheel.name

tests/unit/seed/wheels/test_periodic_update.py

@@ -101,7 +101,7 @@ def test_periodic_update_stops_at_current(mocker, session_app_data, for_py_versi
     )
     mocker.patch("virtualenv.app_data.via_disk_folder.JSONStoreDisk.read", return_value=u_log.to_dict())
 
-    result = periodic_update("setuptools", for_py_version, current, [], session_app_data, False, os.environ)
+    result = periodic_update("setuptools", None, for_py_version, current, [], session_app_data, False, os.environ)
     assert result.path == current.path
 
 
@@ -120,7 +120,7 @@ def test_periodic_update_latest_per_patch(mocker, session_app_data, for_py_versi
     )
     mocker.patch("virtualenv.app_data.via_disk_folder.JSONStoreDisk.read", return_value=u_log.to_dict())
 
-    result = periodic_update("setuptools", for_py_version, current, [], session_app_data, False, os.environ)
+    result = periodic_update("setuptools", None, for_py_version, current, [], session_app_data, False, os.environ)
     assert result.path == current.path
 
 
@@ -166,7 +166,7 @@ def test_periodic_update_skip(u_log, mocker, for_py_version, session_app_data, f
     mocker.patch("virtualenv.app_data.via_disk_folder.JSONStoreDisk.read", return_value=u_log.to_dict())
     mocker.patch("virtualenv.seed.wheels.periodic_update.trigger_update", side_effect=RuntimeError)
 
-    result = periodic_update("setuptools", for_py_version, None, [], session_app_data, os.environ, True)
+    result = periodic_update("setuptools", None, for_py_version, None, [], session_app_data, os.environ, True)
     assert result is None
 
 
@@ -194,7 +194,7 @@ def test_periodic_update_trigger(u_log, mocker, for_py_version, session_app_data
     write = mocker.patch("virtualenv.app_data.via_disk_folder.JSONStoreDisk.write")
     trigger_update_ = mocker.patch("virtualenv.seed.wheels.periodic_update.trigger_update")
 
-    result = periodic_update("setuptools", for_py_version, None, [], session_app_data, os.environ, True)
+    result = periodic_update("setuptools", None, for_py_version, None, [], session_app_data, os.environ, True)
 
     assert result is None
     assert trigger_update_.call_count