Allow installing runtimes from a script shebang.

src/manage/install_command.py

@@ -21,10 +21,6 @@
 )
 
 
-# In-process cache to save repeat downloads
-DOWNLOAD_CACHE = {}
-
-
 def _multihash(file, hashes):
     import hashlib
     LOGGER.debug("Calculating hashes: %s", ", ".join(hashes))
@@ -379,7 +375,8 @@ def _find_one(cmd, source, tag, *, installed=None, by_id=False):
     else:
         LOGGER.verbose("Searching for default Python version")
 
-    downloader = IndexDownloader(source, Index, {}, DOWNLOAD_CACHE)
+    download_cache = cmd.scratch.setdefault("install_command.download_cache", {})
+    downloader = IndexDownloader(source, Index, {}, download_cache)
     install = select_package(downloader, tag, cmd.default_platform, by_id=by_id)
 
     if by_id:
@@ -423,8 +420,9 @@ def _download_one(cmd, source, install, download_dir, *, must_copy=False):
     if install["url"].casefold().endswith(".nupkg".casefold()):
         package = package.with_suffix(".nupkg")
 
+    download_cache = cmd.scratch.setdefault("install_command.download_cache", {})
     with ProgressPrinter("Downloading", maxwidth=CONSOLE_MAX_WIDTH) as on_progress:
-        package = download_package(cmd, install, package, DOWNLOAD_CACHE, on_progress=on_progress)
+        package = download_package(cmd, install, package, download_cache, on_progress=on_progress)
     validate_package(install, package)
     if must_copy and package.parent != download_dir:
         import shutil
@@ -738,7 +736,7 @@ def execute(cmd):
                         break
                     except LookupError:
                         LOGGER.error("Failed to find a suitable install for '%s'.", tag)
-                        raise NoInstallFoundError()
+                        raise NoInstallFoundError(tag)
                     except Exception as ex:
                         LOGGER.debug("Capturing error in case fallbacks fail", exc_info=True)
                         first_exc = first_exc or ex
@@ -757,7 +755,11 @@ def execute(cmd):
             # Have already checked that we are not using --by-id
             from .scriptutils import find_install_from_script
             try:
-                spec = find_install_from_script(cmd, cmd.from_script)
+                spec = find_install_from_script(cmd, cmd.from_script)["tag"]
+            except NoInstallFoundError as ex:
+                # Usually expect this exception, since we should be installing
+                # a runtime that wasn't found.
+                spec = ex.tag
             except LookupError:
                 spec = None
             if spec:
@@ -796,7 +798,7 @@ def execute(cmd):
                                     break
                             except LookupError:
                                 LOGGER.error("Failed to find a suitable update for '%s'.", install['id'])
-                                raise NoInstallFoundError()
+                                raise NoInstallFoundError(install.get('tag'))
                             except Exception as ex:
                                 LOGGER.debug("Capturing error in case fallbacks fail", exc_info=True)
                                 first_exc = first_exc or ex
@@ -832,7 +834,7 @@ def execute(cmd):
                     break
                 except LookupError:
                     LOGGER.error("Failed to find a suitable install for '%s'.", tag)
-                    raise NoInstallFoundError()
+                    raise NoInstallFoundError(tag)
                 except (AssertionError, AttributeError, TypeError):
                     # These errors should never happen.
                     raise

tests/conftest.py

@@ -107,7 +107,8 @@ def __call__(self, *cmp):
                 else:
                     assert re.match(pat, x[0], flags=re.S)
                 if args is not None:
-                    assert tuple(x[1]) == tuple(args)
+                    values = tuple(type(v2)(v1) for v1, v2 in zip(x[1], args))
+                    assert values == tuple(args)
                 break
 
 

tests/test_install_command.py

@@ -1,11 +1,11 @@
 import json
 import os
 import pytest
-import secrets
 from pathlib import Path, PurePath
 
 from manage import install_command as IC
 from manage import installs
+from manage.exceptions import NoInstallFoundError
 
 
 def test_print_cli_shortcuts(patched_installs, assert_log, monkeypatch, tmp_path):
@@ -214,3 +214,116 @@ def cleanup_aliases(cmd, preserve):
         assert sorted(a.name for a in created) == ["python3.exe", "pythonw3.exe"]
     # Ensure we still only have the two targets
     assert set(a.target for a in created) == {"p.exe", "pw.exe"}
+
+
+class InstallCommandTestCmd:
+    def __init__(self, tmp_path, *args, **kwargs):
+        self.args = args
+        self.tags = None
+        self.download_cache = {}
+        self.scratch = {
+            "install_command.download_cache": self.download_cache,
+        }
+        self.automatic = kwargs.get("automatic", False)
+        self.by_id = kwargs.get("by_id", False)
+        self.default_install_tag = kwargs.get("default_install_tag", "1")
+        self.default_platform = kwargs.get("default_platform", "-32")
+        self.default_tag = kwargs.get("default_tag", "1")
+        self.download = kwargs.get("download")
+        if self.download:
+            self.download = tmp_path / self.download
+        self.dry_run = kwargs.get("dry_run", True)
+        self.fallback_source = kwargs.get("fallback_source")
+        self.force = kwargs.get("force", True)
+        self.from_script = kwargs.get("from_script")
+        self.log_file = kwargs.get("log_file")
+        self.refresh = kwargs.get("refresh", False)
+        self.repair = kwargs.get("repair", False)
+        self.shebang_can_run_anything = kwargs.get("shebang_can_run_anything", False)
+        self.shebang_can_run_anything_silently = kwargs.get("shebang_can_run_anything_silently", False)
+        self.source = kwargs.get("source", "http://example.com/index.json")
+        self.target = kwargs.get("target")
+        if self.target:
+            self.target = tmp_path / self.target
+        self.update = kwargs.get("update", False)
+        self.virtual_env = kwargs.get("virtual_env")
+
+        self.index_installs = [
+            {
+                "schema": 1,
+                "id": "test-1.1-32",
+                "sort-version": "1.1",
+                "company": "Test",
+                "tag": "1.1-32",
+                "install-for": ["1", "1.1", "1.1-32"],
+                "display-name": "Test 1.1 (32)",
+                "executable": "test.exe",
+                "url": "about:blank",
+            },
+            {
+                "schema": 1,
+                "id": "test-1.0-32",
+                "sort-version": "1.0",
+                "company": "Test",
+                "tag": "1.0-32",
+                "install-for": ["1", "1.0", "1.0-32"],
+                "display-name": "Test 1.0 (32)",
+                "executable": "test.exe",
+                "url": "about:blank",
+            },
+        ]
+        self.download_cache["http://example.com/index.json"] = json.dumps({
+            "versions": self.index_installs,
+        })
+        self.installs = [{
+            **self.index_installs[-1],
+            "source": self.source,
+            "prefix": tmp_path / "test-1.0-32",
+        }]
+
+    def get_log_file(self):
+        return self.log_file
+
+    def get_installs(self):
+        return self.installs
+
+    def get_install_to_run(self, tag):
+        for i in self.installs:
+            if i["tag"] == tag or f"{i['company']}/{i['tag']}" == tag:
+                return i
+        raise NoInstallFoundError(tag)
+
+
+def test_install_simple(tmp_path, assert_log):
+    cmd = InstallCommandTestCmd(tmp_path, "1.1", force=False)
+
+    IC.execute(cmd)
+    assert_log(
+        assert_log.skip_until("Searching for Python matching %s", ["1.1"]),
+        assert_log.skip_until("Installing %s", ["Test 1.1 (32)"]),
+        ("Tag: %s\\\\%s", ["Test", "1.1-32"]),
+    )
+
+
+def test_install_already_installed(tmp_path, assert_log):
+    cmd = InstallCommandTestCmd(tmp_path, "1.0", force=False)
+
+    IC.execute(cmd)
+    assert_log(
+        assert_log.skip_until("Searching for Python matching %s", ["1.0"]),
+        assert_log.skip_until("%s is already installed", ["Test 1.0 (32)"]),
+    )
+
+
+def test_install_from_script(tmp_path, assert_log):
+    cmd = InstallCommandTestCmd(tmp_path, from_script=tmp_path / "t.py")
+
+    cmd.from_script.parent.mkdir(parents=True, exist_ok=True)
+    cmd.from_script.write_text("#! python1.1.exe")
+
+    IC.execute(cmd)
+    assert_log(
+        assert_log.skip_until("Searching for Python matching"),
+        assert_log.skip_until("Installing %s", ["Test 1.1 (32)"]),
+        ("Tag: %s\\\\%s", ["Test", "1.1-32"]),
+    )