use std::env::consts::EXE_EXTENSION;
use std::io;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::sync::LazyLock;
use anyhow::{Context, Result, bail};
use http::header::ACCEPT;
use semver::{Version, VersionReq};
use target_lexicon::{Architecture, ArmArchitecture, Environment, HOST, OperatingSystem, Triple};
use tracing::{debug, trace, warn};
use prek_consts::env_vars::{EnvVars, EnvVarsRead};
use crate::archive;
use crate::checksum::{digest_from_sha256sums, fetch_checksum};
use crate::fs::LockedFile;
use crate::http::{REQWEST_CLIENT, TempDownload, download_artifact};
use crate::process::Cmd;
use crate::store::{CacheBucket, Store};
use crate::warn_user;
const CUR_UV_VERSION: &str = "0.12.19";
static UV_VERSION_RANGE: LazyLock<VersionReq> =
LazyLock::new(|| VersionReq::parse(">=0.7.0").unwrap());
const ASTRAL_UV_RELEASE_BASE: &str = "https://releases.astral.sh/github/uv/releases/download";
const GITHUB_UV_RELEASE_BASE: &str = "https://github.com/astral-sh/uv/releases/download";
fn release_archive_url(base: &str, version: &str, archive_name: &str) -> String {
format!("{base}/{version}/{archive_name}")
}
fn static_musl_release_target_for_host(
operating_system: OperatingSystem,
architecture: Architecture,
environment: Environment,
) -> Option<&'static str> {
match (operating_system, architecture, environment) {
(OperatingSystem::Linux, Architecture::X86_64, Environment::Gnu) => {
Some("x86_64-unknown-linux-musl")
}
(
OperatingSystem::Linux,
Architecture::Aarch64(target_lexicon::Aarch64Architecture::Aarch64),
Environment::Gnu,
) => Some("aarch64-unknown-linux-musl"),
_ => None,
}
}
fn wheel_platform_tag_for_host(
operating_system: OperatingSystem,
architecture: Architecture,
environment: Environment,
) -> Result<&'static str> {
let platform_tag = match (operating_system, architecture, environment) {
(OperatingSystem::Linux, Architecture::X86_64, Environment::Musl) => "musllinux_1_1_x86_64",
(OperatingSystem::Linux, Architecture::X86_64, _) => {
"manylinux_2_17_x86_64.manylinux2014_x86_64"
}
(OperatingSystem::Linux, Architecture::Aarch64(_), _) => {
"manylinux_2_17_aarch64.manylinux2014_aarch64.musllinux_1_1_aarch64"
}
(OperatingSystem::Linux, Architecture::Arm(ArmArchitecture::Armv7), Environment::Musl) => {
"manylinux_2_17_armv7l.manylinux2014_armv7l.musllinux_1_1_armv7l"
}
(OperatingSystem::Linux, Architecture::Arm(ArmArchitecture::Armv7), _) => {
"manylinux_2_17_armv7l.manylinux2014_armv7l"
}
(OperatingSystem::Linux, Architecture::Arm(ArmArchitecture::Armv6), _) => "linux_armv6l", (OperatingSystem::Linux, Architecture::X86_32(_), Environment::Musl) => {
"musllinux_1_1_i686"
}
(OperatingSystem::Linux, Architecture::X86_32(_), _) => {
"manylinux_2_17_i686.manylinux2014_i686"
}
(OperatingSystem::Linux, Architecture::Powerpc64, _) => {
"manylinux_2_17_ppc64.manylinux2014_ppc64"
}
(OperatingSystem::Linux, Architecture::Powerpc64le, _) => {
"manylinux_2_17_ppc64le.manylinux2014_ppc64le"
}
(OperatingSystem::Linux, Architecture::S390x, _) => {
"manylinux_2_17_s390x.manylinux2014_s390x"
}
(OperatingSystem::Linux, Architecture::Riscv64(_), _) => "manylinux_2_31_riscv64",
(OperatingSystem::Darwin(_), Architecture::X86_64, _) => "macosx_10_12_x86_64",
(OperatingSystem::Darwin(_), Architecture::Aarch64(_), _) => "macosx_11_0_arm64",
(OperatingSystem::Windows, Architecture::X86_64, _) => "win_amd64",
(OperatingSystem::Windows, Architecture::X86_32(_), _) => "win32",
(OperatingSystem::Windows, Architecture::Aarch64(_), _) => "win_arm64",
_ => bail!(
"Unsupported platform: operating_system={operating_system:?}, architecture={architecture:?}, environment={environment:?}"
),
};
Ok(platform_tag)
}
fn get_wheel_platform_tag() -> Result<String> {
wheel_platform_tag_for_host(HOST.operating_system, HOST.architecture, HOST.environment)
.map(ToString::to_string)
}
fn get_uv_version(uv_path: &Path) -> Result<Version> {
let output = Command::new(uv_path)
.arg("--version")
.output()
.context("Failed to execute uv")?;
if !output.status.success() {
bail!("Failed to get uv version");
}
let version_output = str::from_utf8(&output.stdout)?;
let version_str = version_output
.split_whitespace()
.nth(1)
.context("Invalid version output format")?;
Version::parse(version_str).map_err(Into::into)
}
fn validate_uv_binary(uv_path: &Path) -> Result<Version> {
let version = get_uv_version(uv_path)?;
if !UV_VERSION_RANGE.matches(&version) {
bail!(
"uv version `{version}` does not satisfy required range `{}`",
*UV_VERSION_RANGE
);
}
Ok(version)
}
fn should_use_static_musl_fallback(error: &anyhow::Error, uv_path: &Path) -> bool {
if !uv_path.is_file() {
return false;
}
error.chain().any(|cause| {
cause
.downcast_ref::<io::Error>()
.is_some_and(|error| error.kind() == io::ErrorKind::NotFound)
})
}
fn invalid_uv_binary_error(uv_path: &Path, error: anyhow::Error) -> anyhow::Error {
error.context(format!(
"uv binary at `{}` failed validation. This usually means the downloaded uv binary is \
incompatible with the current runtime environment, for example due to a libc mismatch \
or a missing dynamic loader path. If this keeps happening, please report it with details \
about your environment and the full error output",
uv_path.display()
))
}
struct DownloadedUv {
path: PathBuf,
_download: TempDownload,
}
impl DownloadedUv {
fn path(&self) -> &Path {
&self.path
}
}
async fn replace_uv_binary(source: &Path, target_path: &Path) -> Result<()> {
if let Some(parent) = target_path.parent() {
fs_err::tokio::create_dir_all(parent).await?;
}
if target_path.exists() {
debug!(target = %target_path.display(), "Removing existing uv binary");
fs_err::tokio::remove_file(target_path).await?;
}
crate::fs::rename_with_retry(source, target_path).await?;
Ok(())
}
static UV_EXE: LazyLock<Option<(PathBuf, Version)>> = LazyLock::new(|| {
for uv_path in which::which_all("uv").ok()? {
debug!("Found uv in PATH: {}", uv_path.display());
match validate_uv_binary(&uv_path) {
Ok(version) => return Some((uv_path, version)),
Err(err) => warn!(uv = %uv_path.display(), error = %err, "Skipping incompatible uv"),
}
}
None
});
#[derive(Debug, PartialEq, Eq)]
enum PyPiMirror {
Pypi,
Tuna,
Aliyun,
Tencent,
Custom(String),
}
impl PyPiMirror {
fn url(&self) -> &str {
match self {
Self::Pypi => "https://pypi.org/simple/",
Self::Tuna => "https://pypi.tuna.tsinghua.edu.cn/simple/",
Self::Aliyun => "https://mirrors.aliyun.com/pypi/simple/",
Self::Tencent => "https://mirrors.cloud.tencent.com/pypi/simple/",
Self::Custom(url) => url,
}
}
}
#[derive(Debug, PartialEq, Eq)]
enum UvSource {
Auto,
None,
Explicit(InstallSource),
}
#[derive(Debug, PartialEq, Eq)]
enum InstallSource {
Astral,
GitHub,
PyPi(PyPiMirror),
Pip,
}
impl InstallSource {
async fn install(&self, store: &Store, target: &Path) -> Result<Uv> {
match self {
Self::Astral => {
self.install_from_release_archive(store, target, ASTRAL_UV_RELEASE_BASE, &HOST)
.await?;
}
Self::GitHub => {
self.install_from_release_archive(store, target, GITHUB_UV_RELEASE_BASE, &HOST)
.await?;
}
Self::PyPi(source) => self.install_from_pypi(store, target, source).await?,
Self::Pip => self.install_from_pip(target).await?,
}
let uv_path = target.join("uv").with_extension(EXE_EXTENSION);
let version = validate_uv_binary(&uv_path)
.map_err(|error| invalid_uv_binary_error(&uv_path, error))?;
trace!(version = %version, "Successfully installed uv");
Ok(Uv::new(uv_path))
}
async fn install_from_release_archive(
&self,
store: &Store,
target: &Path,
base_url: &str,
host: &Triple,
) -> Result<()> {
let host_target = host.to_string();
let downloaded = self
.download_release_archive(store, base_url, &host_target)
.await?;
let downloaded = match validate_uv_binary(downloaded.path()) {
Ok(_) => downloaded,
Err(error) => {
let Some(static_target) = static_musl_release_target_for_host(
host.operating_system,
host.architecture,
host.environment,
) else {
return Err(invalid_uv_binary_error(downloaded.path(), error));
};
if !should_use_static_musl_fallback(&error, downloaded.path()) {
return Err(invalid_uv_binary_error(downloaded.path(), error));
}
warn!(
target = %host_target,
fallback_target = static_target,
%error,
"uv release binary cannot start; retrying with the static musl binary"
);
drop(downloaded);
let fallback = self
.download_release_archive(store, base_url, static_target)
.await
.with_context(|| {
format!("Failed to download static uv fallback `{static_target}`")
})?;
validate_uv_binary(fallback.path())
.map_err(|error| invalid_uv_binary_error(fallback.path(), error))?;
fallback
}
};
let target_path = target.join("uv").with_extension(EXE_EXTENSION);
debug!(source = ?downloaded.path(), target = %target_path.display(), "Moving uv to target");
replace_uv_binary(downloaded.path(), &target_path).await?;
Ok(())
}
async fn download_release_archive(
&self,
store: &Store,
base_url: &str,
release_target: &str,
) -> Result<DownloadedUv> {
let ext = if cfg!(windows) { "zip" } else { "tar.gz" };
let archive_name = format!("uv-{release_target}.{ext}");
let download_url = release_archive_url(base_url, CUR_UV_VERSION, &archive_name);
let checksum_url = format!("{download_url}.sha256");
let download = download_artifact(&download_url, &archive_name, store, async || {
let Some(checksum) = fetch_checksum(&checksum_url).await? else {
return Ok(None);
};
digest_from_sha256sums(&checksum, &archive_name)
})
.await
.context("Failed to download uv")?;
let extracted = archive::extract_archive(download.path())
.await
.context("Failed to extract uv")?;
let path = extracted.join("uv").with_extension(EXE_EXTENSION);
Ok(DownloadedUv {
path,
_download: download,
})
}
async fn install_from_pypi(
&self,
store: &Store,
target: &Path,
source: &PyPiMirror,
) -> Result<()> {
let platform_tag = get_wheel_platform_tag()?;
let wheel_name = format!("uv-{CUR_UV_VERSION}-py3-none-{platform_tag}.whl");
let api_url = match source {
PyPiMirror::Pypi => format!("https://pypi.org/pypi/uv/{CUR_UV_VERSION}/json"),
_ => return self.install_from_simple_api(store, target, source).await,
};
debug!("Fetching uv metadata from: {}", api_url);
let response = REQWEST_CLIENT
.get(&api_url)
.header("Accept", "*/*")
.send()
.await
.and_then(reqwest::Response::error_for_status)
.with_context(|| format!("Failed to fetch uv metadata from PyPI at {api_url}"))?;
let metadata: serde_json::Value = response.json().await?;
let files = metadata["urls"]
.as_array()
.context("Invalid PyPI response: missing urls")?;
let wheel_file = files
.iter()
.find(|file| {
file["filename"].as_str() == Some(&wheel_name)
&& file["packagetype"].as_str() == Some("bdist_wheel")
&& file["yanked"].as_bool() != Some(true)
})
.with_context(|| format!("Could not find wheel for {wheel_name} in PyPI response"))?;
let download_url = wheel_file["url"]
.as_str()
.context("Missing download URL in PyPI response")?;
let digest = wheel_file["digests"]["sha256"].as_str();
self.install_from_wheel_url(store, target, &wheel_name, download_url, digest)
.await
}
async fn install_from_simple_api(
&self,
store: &Store,
target: &Path,
source: &PyPiMirror,
) -> Result<()> {
let platform_tag = get_wheel_platform_tag()?;
let wheel_name = format!("uv-{CUR_UV_VERSION}-py3-none-{platform_tag}.whl");
let simple_url = format!("{}uv/", source.url());
debug!("Fetching from simple API: {}", simple_url);
let response = REQWEST_CLIENT
.get(&simple_url)
.header(ACCEPT, "*/*")
.send()
.await?;
let html = response.text().await?;
let search_pattern = r#"href=""#.to_string();
let download_path = html
.lines()
.find(|line| line.contains(&wheel_name))
.and_then(|line| {
if let Some(start) = line.find(&search_pattern) {
let start = start + search_pattern.len();
if let Some(end) = line[start..].find('"') {
return Some(&line[start..start + end]);
}
}
None
})
.with_context(|| {
format!(
"Could not find wheel download link for {wheel_name} in simple API response"
)
})?;
let (download_path, digest) = match download_path.split_once('#') {
Some((path, fragment)) => (path, fragment.strip_prefix("sha256=")),
None => (download_path, None),
};
let download_url = if download_path.starts_with("http") {
download_path.to_string()
} else {
format!("{simple_url}{download_path}")
};
self.install_from_wheel_url(store, target, &wheel_name, &download_url, digest)
.await
}
async fn install_from_wheel_url(
&self,
store: &Store,
target: &Path,
filename: &str,
download_url: &str,
expected_digest: Option<&str>,
) -> Result<()> {
let download = download_artifact(download_url, filename, store, async || {
expected_digest.map(str::parse).transpose()
})
.await
.context("Failed to download uv wheel")?;
let extracted = archive::extract_archive(download.path())
.await
.context("Failed to extract uv wheel")?;
let data_dir = format!("uv-{CUR_UV_VERSION}.data");
let extracted_uv = extracted
.join(data_dir)
.join("scripts")
.join("uv")
.with_extension(EXE_EXTENSION);
let target_path = target.join("uv").with_extension(EXE_EXTENSION);
debug!(?extracted_uv, target = %target_path.display(), "Moving uv to target");
replace_uv_binary(&extracted_uv, &target_path).await?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let metadata = fs_err::tokio::metadata(&target_path).await?;
let mut perms = metadata.permissions();
perms.set_mode(0o755);
fs_err::tokio::set_permissions(&target_path, perms).await?;
}
Ok(())
}
async fn install_from_pip(&self, target: &Path) -> Result<()> {
Cmd::new("python3")
.arg("-m")
.arg("pip")
.arg("install")
.arg("--prefix")
.arg(target)
.arg("--only-binary=:all:")
.arg("--progress-bar=off")
.arg("--disable-pip-version-check")
.arg(format!("uv=={CUR_UV_VERSION}"))
.check(true)
.output()
.await?;
let local_dir = target.join("local");
let uv_src = if local_dir.is_dir() {
&local_dir
} else {
target
};
let bin_dir = uv_src.join(if cfg!(windows) { "Scripts" } else { "bin" });
let lib_dir = uv_src.join(if cfg!(windows) { "Lib" } else { "lib" });
let uv = bin_dir.join("uv").with_extension(EXE_EXTENSION);
let target_path = target.join("uv").with_extension(EXE_EXTENSION);
replace_uv_binary(&uv, &target_path).await?;
fs_err::tokio::remove_dir_all(bin_dir).await?;
fs_err::tokio::remove_dir_all(lib_dir).await?;
Ok(())
}
}
pub(crate) struct Uv {
path: PathBuf,
}
impl Uv {
pub(crate) fn new(path: PathBuf) -> Self {
Self { path }
}
pub(crate) fn cmd(&self, store: &Store) -> Cmd {
let mut cmd = Cmd::new(&self.path);
cmd.env(EnvVars::UV_CACHE_DIR, store.cache_path(CacheBucket::Uv));
cmd
}
async fn install_with_fallbacks(store: &Store, uv_dir: &Path) -> Result<Self> {
for source in [
InstallSource::Astral,
InstallSource::PyPi(PyPiMirror::Pypi),
InstallSource::PyPi(PyPiMirror::Tuna),
InstallSource::PyPi(PyPiMirror::Aliyun),
InstallSource::PyPi(PyPiMirror::Tencent),
] {
match source.install(store, uv_dir).await {
Ok(uv) => return Ok(uv),
Err(err) => {
warn!(?source, %err, "Failed to install uv, trying next source");
}
}
}
InstallSource::Pip
.install(store, uv_dir)
.await
.context("Failed to install uv from every source")
}
pub(crate) async fn find_or_install(store: &Store, uv_dir: &Path) -> Result<Self> {
let prek_exe = std::env::current_exe()?.canonicalize()?;
if let Some(prek_dir) = prek_exe.parent() {
let uv_path = prek_dir.join("uv").with_extension(EXE_EXTENSION);
if uv_path.is_file() {
match validate_uv_binary(&uv_path) {
Ok(_) => {
trace!(uv = %uv_path.display(), "Found compatible uv alongside prek binary");
return Ok(Self::new(uv_path));
}
Err(err) => {
warn!(uv = %uv_path.display(), error = %err, "Skipping incompatible uv");
}
}
}
}
if let Some((uv_path, version)) = UV_EXE.as_ref() {
trace!(
"Using system uv version {} at {}",
version,
uv_path.display()
);
return Ok(Self::new(uv_path.clone()));
}
let uv_path = uv_dir.join("uv").with_extension(EXE_EXTENSION);
if uv_path.is_file() {
match validate_uv_binary(&uv_path) {
Ok(_) => {
trace!(uv = %uv_path.display(), "Found compatible managed uv");
return Ok(Self::new(uv_path));
}
Err(err) => {
warn!(uv = %uv_path.display(), error = %err, "Skipping incompatible managed uv");
}
}
}
let source = match uv_source_from_env(&EnvVars) {
UvSource::Auto => None,
UvSource::None => bail!(
"No compatible uv found and automatic installation is disabled by {}=none. \
Install uv ({}) and add it to PATH",
EnvVars::PREK_UV_SOURCE,
*UV_VERSION_RANGE,
),
UvSource::Explicit(source) => Some(source),
};
fs_err::tokio::create_dir_all(&uv_dir).await?;
let _lock = LockedFile::acquire(uv_dir.join(".lock"), "uv").await?;
if uv_path.is_file() {
match validate_uv_binary(&uv_path) {
Ok(_) => {
trace!(uv = %uv_path.display(), "Found compatible managed uv");
return Ok(Self::new(uv_path));
}
Err(err) => {
warn!(uv = %uv_path.display(), error = %err, "Skipping incompatible managed uv");
}
}
}
if let Some(source) = source {
source.install(store, uv_dir).await
} else {
Self::install_with_fallbacks(store, uv_dir).await
}
}
}
fn uv_source_from_env(env_vars: &impl EnvVarsRead) -> UvSource {
let Ok(var) = env_vars.var(EnvVars::PREK_UV_SOURCE) else {
return UvSource::Auto;
};
match var.as_str() {
"none" => UvSource::None,
"astral" => UvSource::Explicit(InstallSource::Astral),
"github" => UvSource::Explicit(InstallSource::GitHub),
"pypi" => UvSource::Explicit(InstallSource::PyPi(PyPiMirror::Pypi)),
"tuna" => UvSource::Explicit(InstallSource::PyPi(PyPiMirror::Tuna)),
"aliyun" => UvSource::Explicit(InstallSource::PyPi(PyPiMirror::Aliyun)),
"tencent" => UvSource::Explicit(InstallSource::PyPi(PyPiMirror::Tencent)),
"pip" => UvSource::Explicit(InstallSource::Pip),
custom if custom.starts_with("http") => {
UvSource::Explicit(InstallSource::PyPi(PyPiMirror::Custom(var)))
}
_ => {
warn_user!(
"Invalid value for {}: {:?}. Expected none, astral, github, pypi, tuna, aliyun, tencent, pip, or an http(s) URL; using default ({:?})",
EnvVars::PREK_UV_SOURCE,
var,
"auto",
);
UvSource::Auto
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn ensure_cur_uv_version_in_range() {
let version = Version::parse(CUR_UV_VERSION).expect("Invalid CUR_UV_VERSION");
assert!(
UV_VERSION_RANGE.matches(&version),
"CUR_UV_VERSION {CUR_UV_VERSION} does not satisfy the version requirement {}",
*UV_VERSION_RANGE
);
}
#[test]
fn release_archive_url_joins_base_version_and_name() {
assert_eq!(
release_archive_url(
ASTRAL_UV_RELEASE_BASE,
"0.11.29",
"uv-x86_64-unknown-linux-gnu.tar.gz",
),
"https://releases.astral.sh/github/uv/releases/download/0.11.29/uv-x86_64-unknown-linux-gnu.tar.gz"
);
assert_eq!(
release_archive_url(
GITHUB_UV_RELEASE_BASE,
"0.11.29",
"uv-x86_64-pc-windows-msvc.zip"
),
"https://github.com/astral-sh/uv/releases/download/0.11.29/uv-x86_64-pc-windows-msvc.zip"
);
}
#[test]
fn uv_source_from_env_reads_source_override() {
assert_eq!(uv_source_from_env(&EnvVars::from_map(&[])), UvSource::Auto);
assert_eq!(
uv_source_from_env(&EnvVars::from_map(&[(EnvVars::PREK_UV_SOURCE, "none")])),
UvSource::None
);
assert_eq!(
uv_source_from_env(&EnvVars::from_map(&[(EnvVars::PREK_UV_SOURCE, "astral")])),
UvSource::Explicit(InstallSource::Astral)
);
assert_eq!(
uv_source_from_env(&EnvVars::from_map(&[(EnvVars::PREK_UV_SOURCE, "github")])),
UvSource::Explicit(InstallSource::GitHub)
);
assert_eq!(
uv_source_from_env(&EnvVars::from_map(&[(EnvVars::PREK_UV_SOURCE, "pypi")])),
UvSource::Explicit(InstallSource::PyPi(PyPiMirror::Pypi))
);
assert_eq!(
uv_source_from_env(&EnvVars::from_map(&[(EnvVars::PREK_UV_SOURCE, "pip")])),
UvSource::Explicit(InstallSource::Pip)
);
assert_eq!(
uv_source_from_env(&EnvVars::from_map(&[(
EnvVars::PREK_UV_SOURCE,
"https://example.com/simple",
)])),
UvSource::Explicit(InstallSource::PyPi(PyPiMirror::Custom(
"https://example.com/simple".to_string()
)))
);
assert_eq!(
uv_source_from_env(&EnvVars::from_map(&[(EnvVars::PREK_UV_SOURCE, "unknown")])),
UvSource::Auto
);
}
#[test]
fn wheel_platform_tag_x86_64_linux_gnu() -> Result<()> {
let tag = wheel_platform_tag_for_host(
OperatingSystem::Linux,
Architecture::X86_64,
Environment::Gnu,
)?;
assert_eq!(tag, "manylinux_2_17_x86_64.manylinux2014_x86_64");
Ok(())
}
#[test]
fn wheel_platform_tag_x86_64_linux_musl() -> Result<()> {
let tag = wheel_platform_tag_for_host(
OperatingSystem::Linux,
Architecture::X86_64,
Environment::Musl,
)?;
assert_eq!(tag, "musllinux_1_1_x86_64");
Ok(())
}
#[test]
fn wheel_platform_tag_i686_linux_gnu() -> Result<()> {
let tag = wheel_platform_tag_for_host(
OperatingSystem::Linux,
Architecture::X86_32(target_lexicon::X86_32Architecture::I686),
Environment::Gnu,
)?;
assert_eq!(tag, "manylinux_2_17_i686.manylinux2014_i686");
Ok(())
}
#[test]
fn wheel_platform_tag_i686_linux_musl() -> Result<()> {
let tag = wheel_platform_tag_for_host(
OperatingSystem::Linux,
Architecture::X86_32(target_lexicon::X86_32Architecture::I686),
Environment::Musl,
)?;
assert_eq!(tag, "musllinux_1_1_i686");
Ok(())
}
#[test]
fn wheel_platform_tag_aarch64_linux_gnu() -> Result<()> {
let tag = wheel_platform_tag_for_host(
OperatingSystem::Linux,
Architecture::Aarch64(target_lexicon::Aarch64Architecture::Aarch64),
Environment::Gnu,
)?;
assert_eq!(
tag,
"manylinux_2_17_aarch64.manylinux2014_aarch64.musllinux_1_1_aarch64"
);
Ok(())
}
#[test]
fn wheel_platform_tag_aarch64_linux_musl() -> Result<()> {
let tag = wheel_platform_tag_for_host(
OperatingSystem::Linux,
Architecture::Aarch64(target_lexicon::Aarch64Architecture::Aarch64),
Environment::Musl,
)?;
assert_eq!(
tag,
"manylinux_2_17_aarch64.manylinux2014_aarch64.musllinux_1_1_aarch64"
);
Ok(())
}
#[test]
fn wheel_platform_tag_armv7_linux_gnu() -> Result<()> {
let tag = wheel_platform_tag_for_host(
OperatingSystem::Linux,
Architecture::Arm(ArmArchitecture::Armv7),
Environment::Gnu,
)?;
assert_eq!(tag, "manylinux_2_17_armv7l.manylinux2014_armv7l");
Ok(())
}
#[test]
fn wheel_platform_tag_armv7_linux_musl() -> Result<()> {
let tag = wheel_platform_tag_for_host(
OperatingSystem::Linux,
Architecture::Arm(ArmArchitecture::Armv7),
Environment::Musl,
)?;
assert_eq!(
tag,
"manylinux_2_17_armv7l.manylinux2014_armv7l.musllinux_1_1_armv7l"
);
Ok(())
}
#[cfg(unix)]
mod install_from_release_archive {
use std::collections::HashMap;
use anyhow::{Context, Result};
use async_compression::tokio::write::GzipEncoder;
use aws_lc_rs::digest::{SHA256, digest};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
use tokio::task::JoinHandle;
use tokio_tar::{Builder, Header};
use super::*;
async fn uv_archive(contents: &[u8]) -> Result<Vec<u8>> {
let mut header = Header::new_gnu();
header.set_mode(0o755);
header.set_size(contents.len().try_into()?);
let mut archive = Builder::new(Vec::new());
archive
.append_data(&mut header, "uv-release/uv", contents)
.await?;
let archive = archive.into_inner().await?;
let mut encoder = GzipEncoder::new(Vec::new());
encoder.write_all(&archive).await?;
encoder.shutdown().await?;
Ok(encoder.into_inner())
}
fn add_release(
files: &mut HashMap<String, Vec<u8>>,
release_target: &str,
archive: Vec<u8>,
) {
let archive_name = format!("uv-{release_target}.tar.gz");
let archive_path = format!("/{CUR_UV_VERSION}/{archive_name}");
let checksum = hex::encode(digest(&SHA256, &archive).as_ref());
files.insert(
format!("{archive_path}.sha256"),
format!("{checksum} {archive_name}\n").into_bytes(),
);
files.insert(archive_path, archive);
}
async fn serve_files(
files: HashMap<String, Vec<u8>>,
) -> Result<(String, JoinHandle<Result<()>>)> {
let listener = TcpListener::bind(("127.0.0.1", 0)).await?;
let url = format!("http://{}", listener.local_addr()?);
let server = tokio::spawn(async move {
loop {
let (mut stream, _) = listener.accept().await?;
let mut request = Vec::new();
loop {
let mut buffer = [0_u8; 1024];
let read = stream.read(&mut buffer).await?;
if read == 0 {
break;
}
request.extend_from_slice(&buffer[..read]);
if request.windows(4).any(|window| window == b"\r\n\r\n") {
break;
}
}
let request = std::str::from_utf8(&request)?;
let path = request
.split_ascii_whitespace()
.nth(1)
.context("Missing request path")?;
let body = files
.get(path)
.cloned()
.with_context(|| format!("Unexpected request path `{path}`"))?;
stream
.write_all(
format!(
"HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
)
.as_bytes(),
)
.await?;
stream.write_all(&body).await?;
}
});
Ok((url, server))
}
#[tokio::test]
async fn retries_with_static_musl_when_native_interpreter_is_missing() -> Result<()> {
let host: Triple = "x86_64-unknown-linux-gnu"
.parse()
.map_err(|error| anyhow::anyhow!("Failed to parse test target: {error}"))?;
let static_target = static_musl_release_target_for_host(
host.operating_system,
host.architecture,
host.environment,
)
.context("Missing static musl target")?;
let working_uv = format!("#!/bin/sh\nprintf 'uv {CUR_UV_VERSION}\\n'\n").into_bytes();
let mut files = HashMap::new();
add_release(
&mut files,
&host.to_string(),
uv_archive(b"#!/definitely/missing/prek-uv-interpreter\n").await?,
);
add_release(&mut files, static_target, uv_archive(&working_uv).await?);
let (base_url, server) = serve_files(files).await?;
let temp = tempfile::tempdir()?;
let store = Store::from_path(temp.path().join("store"))?.init()?;
let target = temp.path().join("target");
let source = InstallSource::Astral;
let result = source
.install_from_release_archive(&store, &target, &base_url, &host)
.await;
server.abort();
result?;
let installed = target.join("uv").with_extension(EXE_EXTENSION);
assert_eq!(fs_err::read(installed)?, working_uv);
Ok(())
}
}
#[test]
fn missing_download_does_not_use_static_musl_fallback() -> Result<()> {
let temp = tempfile::tempdir()?;
let uv = temp.path().join("missing-uv");
let error = validate_uv_binary(&uv).unwrap_err();
assert!(!should_use_static_musl_fallback(&error, &uv));
Ok(())
}
#[tokio::test]
async fn replace_uv_binary_overwrites_existing_file() -> Result<()> {
let temp = tempfile::tempdir()?;
let source = temp.path().join("source-uv");
let target_dir = temp.path().join("tools").join("uv");
let target_path = target_dir.join("uv").with_extension(EXE_EXTENSION);
fs_err::create_dir_all(&target_dir)?;
fs_err::write(&source, b"new")?;
fs_err::write(&target_path, b"old")?;
replace_uv_binary(&source, &target_path).await?;
assert!(!source.exists());
assert_eq!(fs_err::read(&target_path)?, b"new");
Ok(())
}
#[tokio::test]
async fn replace_uv_binary_recreates_missing_parent_dir() -> Result<()> {
let temp = tempfile::tempdir()?;
let source = temp.path().join("source-uv");
let target_dir = temp.path().join("tools").join("uv");
let target_path = target_dir.join("uv").with_extension(EXE_EXTENSION);
fs_err::create_dir_all(&target_dir)?;
fs_err::write(&target_path, b"old")?;
fs_err::remove_dir_all(&target_dir)?;
fs_err::write(&source, b"new")?;
replace_uv_binary(&source, &target_path).await?;
assert!(target_dir.exists());
assert_eq!(fs_err::read(&target_path)?, b"new");
Ok(())
}
}