fix(self-update): pin installer ref to latest version

src/apm_cli/commands/self_update.py

@@ -40,7 +40,7 @@ def _is_windows_platform() -> bool:
     return sys.platform == "win32"
 
 
-def _get_update_installer_url() -> str:
+def _get_update_installer_url(script_ref: str | None = None) -> str:
     """Return the installer URL for the current platform, respecting mirror env vars.
 
     ``APM_INSTALLER_BASE_URL`` wins when configured, using ``install.sh`` on
@@ -65,7 +65,8 @@ def _get_update_installer_url() -> str:
     if github_url == _DEFAULT_GITHUB_URL:
         return "https://aka.ms/apm-windows" if _is_windows_platform() else "https://aka.ms/apm-unix"
 
-    return f"{github_url}/{apm_repo}/raw/{_INSTALL_SCRIPT_REF}/{script_name}"
+    resolved_ref = script_ref or _INSTALL_SCRIPT_REF
+    return f"{github_url}/{apm_repo}/raw/{resolved_ref}/{script_name}"
 
 
 def _get_update_installer_suffix() -> str:
@@ -155,7 +156,7 @@ def _build_self_update_installer_env(latest_version: str) -> dict[str, str]:
     channel = _get_effective_self_update_channel()
     if _ENV_SELF_UPDATE_CHANNEL not in env:
         env[_ENV_SELF_UPDATE_CHANNEL] = channel
-    if channel == "prerelease" and _ENV_VERSION not in env:
+    if _ENV_VERSION not in env:
         env[_ENV_VERSION] = f"v{latest_version}"
     return env
 
@@ -272,7 +273,8 @@ def self_update(check):
             import requests
 
             try:
-                install_script_url = _get_update_installer_url()
+                installer_ref = f"v{latest_version}"
+                install_script_url = _get_update_installer_url(installer_ref)
             except RuntimeError as e:
                 logger.error(str(e))
                 logger.info(

tests/unit/commands/test_self_update_air_gapped.py

@@ -122,6 +122,24 @@ def test_custom_github_url_unix_uses_sh(self) -> None:
         parsed = urlparse(url)
         assert parsed.path.endswith("install.sh")
 
+    def test_custom_github_url_uses_supplied_script_ref(self) -> None:
+        """Supplying a script ref points installer downloads to that tag path."""
+        from urllib.parse import urlparse
+
+        env = self._clean_env()
+        env["GITHUB_URL"] = "https://gh.corp.com"
+        env["APM_REPO"] = "corp/apm"
+        with patch.dict("os.environ", env, clear=True):
+            with patch(
+                "apm_cli.commands.self_update._is_windows_platform",
+                return_value=False,
+            ):
+                from apm_cli.commands.self_update import _get_update_installer_url
+
+                url = _get_update_installer_url("v1.2.3")
+        parsed = urlparse(url)
+        assert parsed.path == "/corp/apm/raw/v1.2.3/install.sh"
+
     def test_github_url_with_trailing_slash_is_normalised(self) -> None:
         """GITHUB_URL with a trailing slash must not produce double-slash in the URL."""
         env = self._clean_env()
@@ -357,3 +375,65 @@ def fake_get(url: str, **_kwargs: object) -> Mock:
             "/apm/installers/install.sh",
         ]
         mock_run.assert_called_once()
+
+    def test_self_update_pins_installer_ref_to_latest_version(self) -> None:
+        """Self-update downloads installer from the exact latest release ref."""
+        env = self._clean_env()
+        env.update(
+            {
+                "APM_NO_DIRECT_FALLBACK": "1",
+                "APM_RELEASE_METADATA_URL": "https://mirror.corp.example/apm/latest.json",
+                "GITHUB_URL": "https://gh.corp.com",
+                "APM_TEMP_DIR": str(self.scratch),
+                "APM_REPO": "corp/apm",
+                "APM_E2E_TESTS": "1",
+            }
+        )
+        requested_urls: list[str] = []
+        target_version = "1.1.0"
+        expected_installer_path = f"/corp/apm/raw/v{target_version}/install.sh"
+
+        def fake_get(url: str, **_kwargs: object) -> Mock:
+            requested_urls.append(url)
+            parsed = urlparse(url)
+            if parsed.hostname == "mirror.corp.example":
+                if parsed.path == "/apm/latest.json":
+                    response = Mock()
+                    response.status_code = 200
+                    response.raise_for_status.return_value = None
+                    response.json.return_value = {"tag_name": f"v{target_version}"}
+                    return response
+                raise AssertionError(f"unexpected mirror request: {url}")
+
+            if parsed.hostname == "gh.corp.com":
+                if parsed.path == expected_installer_path:
+                    response = Mock()
+                    response.status_code = 200
+                    response.raise_for_status.return_value = None
+                    response.text = "echo install"
+                    return response
+                raise AssertionError(f"unexpected github request: {url}")
+
+            raise AssertionError(f"unexpected request: {url}")
+
+        with (
+            patch.dict("os.environ", env, clear=True),
+            patch("requests.get", side_effect=fake_get),
+            patch("subprocess.run", return_value=Mock(returncode=0)) as mock_run,
+            patch("apm_cli.commands.self_update.get_version", return_value="1.0.0"),
+            patch("apm_cli.commands.self_update.os.chmod"),
+            patch.object(update_module.sys, "platform", "linux"),
+            patch("apm_cli.commands.self_update.os.path.exists", return_value=True),
+        ):
+            result = self.runner.invoke(cli, ["self-update"])
+
+        assert result.exit_code == 0
+        assert "Successfully updated to version 1.1.0" in result.output
+        parsed_paths = [urlparse(url).path for url in requested_urls]
+        assert parsed_paths == [
+            "/apm/latest.json",
+            expected_installer_path,
+        ]
+        assert "/main/" not in "".join(parsed_paths)
+        assert f"/raw/v{target_version}/" in "".join(parsed_paths)
+        mock_run.assert_called_once()