add tests

Signed-off-by: Bernat Gabor <[email protected]>

src/virtualenv/seed/wheels/periodic_update.py

@@ -6,11 +6,13 @@
 
 import json
 import logging
+import os
 import subprocess
 import sys
 from datetime import datetime, timedelta
 from itertools import groupby
 from shutil import copy2
+from threading import Thread
 
 from six.moves.urllib.request import urlopen
 
@@ -40,7 +42,7 @@ def periodic_update(distribution, for_py_version, wheel, search_dirs, app_data,
     u_log = UpdateLog.from_app_data(app_data, distribution, for_py_version)
     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 version.filename == wheel.name:
+        if wheel is not None and version.filename.name == wheel.name:
             break
         if u_log.periodic is False or version.use(now):
             updated_wheel = Wheel(app_data.house / version.filename)
@@ -127,7 +129,8 @@ def from_dict(cls, dictionary):
 
     @classmethod
     def from_app_data(cls, app_data, distribution, for_py_version):
-        return cls.from_dict(app_data.embed_update_log(distribution, for_py_version).read())
+        raw_json = app_data.embed_update_log(distribution, for_py_version).read()
+        return cls.from_dict(raw_json)
 
     def to_dict(self):
         return {
@@ -174,7 +177,8 @@ def trigger_update(distribution, for_py_version, wheel, search_dirs, app_data, p
         for_py_version,
         process.pid,
     )
-    # process.communicate()  # on purpose not called to make it detached
+    if os.environ.get(str("_VIRTUALENV_PERIODIC_UPDATE_INLINE")) == str("1"):
+        process.communicate()  # on purpose not called to make it a background process
 
 
 def do_update(distribution, for_py_version, embed_filename, app_data, search_dirs, periodic):
@@ -230,45 +234,56 @@ def _get_release_date(dest):
 
 
 def manual_upgrade(app_data):
-    from .bundle import from_bundle
+    threads = []
 
     for for_py_version, distribution_to_package in BUNDLE_SUPPORT.items():
         # load extra search dir for the given for_py
         for distribution in distribution_to_package.keys():
-            start = datetime.now()
-            current = from_bundle(
-                distribution=distribution,
-                version=None,
-                for_py_version=for_py_version,
-                search_dirs=[],
-                app_data=app_data,
-                do_periodic_update=False,
-            )
-            logging.warning(
-                "upgrade %s for python %s with current %s",
-                distribution,
-                for_py_version,
-                "" if current is None else current.name,
-            )
-            versions = do_update(
-                distribution=distribution,
-                for_py_version=for_py_version,
-                embed_filename=current.path,
-                app_data=app_data,
-                search_dirs=[],
-                periodic=False,
-            )
-            msg = "upgraded %s for python %s in %s {}".format(
-                "new entries found:\n%s" if versions else "no new versions found",
-            )
-            args = [
-                distribution,
-                for_py_version,
-                datetime.now() - start,
-            ]
-            if versions:
-                args.append("\n".join("\t{}".format(v) for v in versions))
-            logging.warning(msg, *args)
+            thread = Thread(target=_run_manual_upgrade, args=(app_data, distribution, for_py_version))
+            thread.start()
+            threads.append(thread)
+
+    for thread in threads:
+        thread.join()
+
+
+def _run_manual_upgrade(app_data, distribution, for_py_version):
+    start = datetime.now()
+    from .bundle import from_bundle
+
+    current = from_bundle(
+        distribution=distribution,
+        version=None,
+        for_py_version=for_py_version,
+        search_dirs=[],
+        app_data=app_data,
+        do_periodic_update=False,
+    )
+    logging.warning(
+        "upgrade %s for python %s with current %s",
+        distribution,
+        for_py_version,
+        "" if current is None else current.name,
+    )
+    versions = do_update(
+        distribution=distribution,
+        for_py_version=for_py_version,
+        embed_filename=current.path,
+        app_data=app_data,
+        search_dirs=[],
+        periodic=False,
+    )
+    msg = "upgraded %s for python %s in %s {}".format(
+        "new entries found:\n%s" if versions else "no new versions found",
+    )
+    args = [
+        distribution,
+        for_py_version,
+        datetime.now() - start,
+    ]
+    if versions:
+        args.append("\n".join("\t{}".format(v) for v in versions))
+    logging.warning(msg, *args)
 
 
 __all__ = (

tests/unit/seed/wheels/test_periodic_update.py

@@ -1,7 +1,11 @@
+from __future__ import absolute_import, unicode_literals
+
 from datetime import datetime, timedelta
+from itertools import zip_longest
 
+from virtualenv import cli_run
 from virtualenv.seed.wheels.embed import BUNDLE_SUPPORT, get_embed_wheel
-from virtualenv.seed.wheels.periodic_update import NewVersion, manual_upgrade
+from virtualenv.seed.wheels.periodic_update import NewVersion, manual_upgrade, UpdateLog, periodic_update
 
 
 def test_manual_upgrade(session_app_data, caplog, mocker, for_py_version):
@@ -21,3 +25,67 @@ def _do_update(distribution, for_py_version, embed_filename, app_data, search_di
     assert " new entries found:\n\tNewVersion" in caplog.text
     assert " no new versions found" in caplog.text
     assert do_update.call_count == 3 * len(BUNDLE_SUPPORT)
+
+
+def test_pick_periodic_update(tmp_path, session_app_data, mocker, for_py_version):
+    embed, current = get_embed_wheel("setuptools", "3.4"), get_embed_wheel("setuptools", for_py_version)
+    mocker.patch("virtualenv.seed.wheels.bundle.load_embed_wheel", return_value=embed)
+    completed = datetime.now() - timedelta(days=29)
+    u_log = UpdateLog(
+        started=datetime.now() - timedelta(days=30),
+        completed=completed,
+        versions=[NewVersion(filename=current.path, found_date=completed, release_date=completed)],
+        periodic=True,
+    )
+    read_dict = mocker.patch("virtualenv.app_data.via_disk_folder.JSONStoreDisk.read", return_value=u_log.to_dict())
+
+    result = cli_run([str(tmp_path), "--activators", "", "--no-periodic-update", "--no-wheel", "--no-pip"])
+
+    assert read_dict.call_count == 1
+    installed = list(i.name for i in result.creator.purelib.iterdir() if i.suffix == ".dist-info")
+    assert "setuptools-{}.dist-info".format(current.version) in installed
+
+
+def test_periodic_update_stops_at_current(mocker, session_app_data, for_py_version):
+    current = get_embed_wheel("setuptools", for_py_version)
+
+    now, completed = datetime.now(), datetime.now() - timedelta(days=29)
+    u_log = UpdateLog(
+        started=completed,
+        completed=completed,
+        versions=[
+            NewVersion(wheel_path(current, (1,)), completed, now - timedelta(days=1)),
+            NewVersion(filename=current.path, found_date=completed, release_date=now - timedelta(days=2)),
+            NewVersion(wheel_path(current, (-1,)), completed, now - timedelta(days=30)),
+        ],
+        periodic=True,
+    )
+    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)
+    assert result.path == current.path
+
+
+def test_periodic_update_latest_per_patch(mocker, session_app_data, for_py_version):
+    current = get_embed_wheel("setuptools", for_py_version)
+    now, completed = datetime.now(), datetime.now() - timedelta(days=29)
+    u_log = UpdateLog(
+        started=completed,
+        completed=completed,
+        versions=[
+            NewVersion(wheel_path(current, (0, 1, 2)), completed, now - timedelta(days=1)),
+            NewVersion(wheel_path(current, (0, 1, 1)), completed, now - timedelta(days=30)),
+            NewVersion(filename=current.path, found_date=completed, release_date=now - timedelta(days=2)),
+        ],
+        periodic=True,
+    )
+    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)
+    assert result.path == current.path
+
+
+def wheel_path(wheel, of):
+    new_version = ".".join(str(i) for i in (tuple(sum(x) for x in zip_longest(wheel.version_tuple, of, fillvalue=0))))
+    new_name = wheel.name.replace(wheel.version, new_version)
+    return wheel.path.parent / new_name