Test trigger

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

setup.cfg

@@ -97,6 +97,7 @@ testing =
     pytest-env >= 0.6.2
     pytest-randomly >= 1
     pytest-timeout >= 1
+    pytest-freezegun >= 0.4.1
     flaky >= 3
     xonsh >= 0.9.16; python_version > '3.4' and python_version != '3.9'
 

src/virtualenv/seed/wheels/periodic_update.py

@@ -63,7 +63,7 @@ def handle_auto_update(distribution, for_py_version, wheel, search_dirs, app_dat
         trigger_update(distribution, for_py_version, wheel, search_dirs, app_data, periodic=True)
 
 
-DATETIME_FMT = "%Y-%m-%dT%H:%M:%SZ"
+DATETIME_FMT = "%Y-%m-%dT%H:%M:%S.%fZ"
 
 
 def dump_datetime(value):
@@ -151,11 +151,10 @@ def needs_update(self):
             return self._check_start(now)
 
     def _check_start(self, now):
-        return self.started is None or now - self.started >= timedelta(hours=1)
+        return self.started is None or now - self.started > timedelta(hours=1)
 
 
 def trigger_update(distribution, for_py_version, wheel, search_dirs, app_data, periodic):
-
     wheel_path = None if wheel is None else str(wheel.path)
     cmd = [
         sys.executable,
@@ -165,9 +164,10 @@ def trigger_update(distribution, for_py_version, wheel, search_dirs, app_data, p
             distribution, for_py_version, wheel_path, str(app_data), [str(p) for p in search_dirs], periodic,
         ),
     ]
-    pipe = subprocess.PIPE if False else None
+    debug = os.environ.get(str("_VIRTUALENV_PERIODIC_UPDATE_INLINE")) == str("1")
+    pipe = None if debug else subprocess.PIPE
     kwargs = {"stdout": pipe, "stderr": pipe}
-    if sys.platform == "win32":
+    if not debug and sys.platform == "win32":
         kwargs["creationflags"] = DETACHED_PROCESS
     process = Popen(cmd, **kwargs)
     logging.info(
@@ -177,7 +177,7 @@ def trigger_update(distribution, for_py_version, wheel, search_dirs, app_data, p
         for_py_version,
         process.pid,
     )
-    if os.environ.get(str("_VIRTUALENV_PERIODIC_UPDATE_INLINE")) == str("1"):
+    if debug:
         process.communicate()  # on purpose not called to make it a background process
 
 
@@ -291,4 +291,8 @@ def _run_manual_upgrade(app_data, distribution, for_py_version):
     "do_update",
     "manual_upgrade",
     "NewVersion",
+    "UpdateLog",
+    "load_datetime",
+    "dump_datetime",
+    "trigger_update",
 )

tests/unit/seed/wheels/test_periodic_update.py

@@ -1,12 +1,23 @@
 from __future__ import absolute_import, unicode_literals
 
+import subprocess
+import sys
 from datetime import datetime, timedelta
 
+import pytest
 from six.moves 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, UpdateLog, manual_upgrade, periodic_update
+from virtualenv.seed.wheels.periodic_update import (
+    NewVersion,
+    UpdateLog,
+    load_datetime,
+    manual_upgrade,
+    periodic_update,
+    trigger_update,
+)
+from virtualenv.util.subprocess import DETACHED_PROCESS
 
 
 def test_manual_upgrade(session_app_data, caplog, mocker, for_py_version):
@@ -90,3 +101,122 @@ 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 str(wheel.path.parent / new_name)
+
+
+_UP_NOW = datetime.now()
+_UPDATE_SKIP = {
+    "started_just_now_no_complete": UpdateLog(started=_UP_NOW, completed=None, versions=[], periodic=True),
+    "started_1_hour_no_complete": UpdateLog(
+        started=_UP_NOW - timedelta(hours=1), completed=None, versions=[], periodic=True,
+    ),
+    "completed_under_two_weeks": UpdateLog(
+        started=None, completed=_UP_NOW - timedelta(days=14), versions=[], periodic=True,
+    ),
+    "started_just_now_completed_two_weeks": UpdateLog(
+        started=_UP_NOW, completed=_UP_NOW - timedelta(days=14, seconds=1), versions=[], periodic=True,
+    ),
+    "started_1_hour_completed_two_weeks": UpdateLog(
+        started=_UP_NOW - timedelta(hours=1),
+        completed=_UP_NOW - timedelta(days=14, seconds=1),
+        versions=[],
+        periodic=True,
+    ),
+}
+
+
+@pytest.mark.parametrize("u_log", _UPDATE_SKIP.values(), ids=_UPDATE_SKIP.keys())
+def test_periodic_update_skip(u_log, mocker, for_py_version, session_app_data, freezer):
+    freezer.move_to(_UP_NOW)
+    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, True)
+    assert result is None
+
+
+_UPDATE_YES = {
+    "never_started": UpdateLog(started=None, completed=None, versions=[], periodic=False),
+    "started_1_hour": UpdateLog(
+        started=_UP_NOW - timedelta(hours=1, microseconds=1), completed=None, versions=[], periodic=False,
+    ),
+    "completed_two_week": UpdateLog(
+        started=_UP_NOW - timedelta(days=14, microseconds=2),
+        completed=_UP_NOW - timedelta(days=14, microseconds=1),
+        versions=[],
+        periodic=False,
+    ),
+}
+
+
+@pytest.mark.parametrize("u_log", _UPDATE_YES.values(), ids=_UPDATE_YES.keys())
+def test_periodic_update_trigger(u_log, mocker, for_py_version, session_app_data, freezer):
+    freezer.move_to(_UP_NOW)
+    mocker.patch("virtualenv.app_data.via_disk_folder.JSONStoreDisk.read", return_value=u_log.to_dict())
+    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, True)
+
+    assert result is None
+    assert trigger_update_.call_count
+    assert write.call_count == 1
+    wrote_json = write.call_args[0][0]
+    assert wrote_json["periodic"] is True
+    assert load_datetime(wrote_json["started"]) == _UP_NOW
+
+
+def test_trigger_update_no_debug(for_py_version, session_app_data, tmp_path, mocker, monkeypatch):
+    monkeypatch.delenv(str("_VIRTUALENV_PERIODIC_UPDATE_INLINE"), raising=False)
+    current = get_embed_wheel("setuptools", for_py_version)
+    process = mocker.MagicMock()
+    process.communicate.return_value = None, None
+    Popen = mocker.patch("virtualenv.seed.wheels.periodic_update.Popen", return_value=process)
+
+    trigger_update("setuptools", for_py_version, current, [tmp_path / "a", tmp_path / "b"], session_app_data, True)
+
+    assert Popen.call_count == 1
+    args, kwargs = Popen.call_args
+    assert args == (
+        [
+            sys.executable,
+            "-c",
+            "from virtualenv.seed.wheels.periodic_update import do_update;"
+            "do_update('setuptools', '{}', '{}', '{}', ['{}', '{}'], True)".format(
+                for_py_version, current.path, session_app_data, tmp_path / "a", tmp_path / "b",
+            ),
+        ],
+    )
+    expected = {"stdout": subprocess.PIPE, "stderr": subprocess.PIPE}
+    if sys.platform == "win32":
+        expected["creationflags"] = DETACHED_PROCESS
+    assert kwargs == expected
+    assert process.communicate.call_count == 0
+
+
+def test_trigger_update_debug(for_py_version, session_app_data, tmp_path, mocker, monkeypatch):
+    monkeypatch.setenv(str("_VIRTUALENV_PERIODIC_UPDATE_INLINE"), str("1"))
+    current = get_embed_wheel("pip", for_py_version)
+
+    process = mocker.MagicMock()
+    process.communicate.return_value = None, None
+    Popen = mocker.patch("virtualenv.seed.wheels.periodic_update.Popen", return_value=process)
+
+    trigger_update("pip", for_py_version, current, [tmp_path / "a", tmp_path / "b"], session_app_data, False)
+
+    assert Popen.call_count == 1
+    args, kwargs = Popen.call_args
+    assert args == (
+        [
+            sys.executable,
+            "-c",
+            "from virtualenv.seed.wheels.periodic_update import do_update;"
+            "do_update('pip', '{}', '{}', '{}', ['{}', '{}'], False)".format(
+                for_py_version, current.path, session_app_data, tmp_path / "a", tmp_path / "b",
+            ),
+        ],
+    )
+    expected = {"stdout": None, "stderr": None}
+    if sys.platform == "win32":
+        expected["creationflags"] = DETACHED_PROCESS
+    assert kwargs == expected
+    assert process.communicate.call_count == 1