use std::path::PathBuf;
use quilt_rs::lineage::SyncScope;
use quilt_uri::Namespace;
use crate::cli::Error;
use crate::cli::keep;
use crate::cli::model::Commands;
use crate::cli::output::Render;
use crate::cli::output::Std;
#[derive(Debug)]
pub struct Input {
pub namespace: Option<Namespace>,
pub paths: Option<Vec<PathBuf>>,
pub uri: String,
pub keep: Option<SyncScope>,
pub download: bool,
}
#[derive(Debug)]
pub struct Output {
installed_package: quilt_rs::InstalledPackage,
paths: Vec<std::path::PathBuf>,
skipped: Vec<std::path::PathBuf>,
hash_mismatch: Vec<std::path::PathBuf>,
already_installed: Vec<std::path::PathBuf>,
keep: Option<keep::Output>,
}
#[cfg(test)]
impl Output {
pub fn get_installed_package(self) -> quilt_rs::InstalledPackage {
self.installed_package
}
}
impl Output {
fn installed(&self) -> impl Iterator<Item = &PathBuf> {
self.paths.iter().filter(|path| {
!self.skipped.contains(path)
&& !self.hash_mismatch.contains(path)
&& !self.already_installed.contains(path)
})
}
}
impl std::fmt::Display for Output {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let mut output = vec![format!("{}", self.installed_package)];
let installed: Vec<_> = self.installed().collect();
if installed.is_empty() && self.keep.is_none() {
output.push("No paths installed".to_string());
} else {
for path in installed {
output.push(format!("Path: \"{}\"", path.display()));
}
}
for path in &self.already_installed {
output.push(format!("Already installed: \"{}\"", path.display()));
}
for path in &self.skipped {
output.push(format!(
"Skipped: \"{}\" (no longer on the remote)",
path.display()
));
}
for path in &self.hash_mismatch {
output.push(format!(
"Hash mismatch: \"{}\" (the package may be corrupt)",
path.display()
));
}
if let Some(keep) = &self.keep {
output.push(keep.to_string());
}
write!(f, "{}", output.join("\n"))
}
}
impl Render for Output {
fn to_json(&self) -> serde_json::Value {
serde_json::json!({
"namespace": self.installed_package.namespace.to_string(),
"paths": self
.installed()
.map(|path| path.display().to_string())
.collect::<Vec<_>>(),
"skipped": self
.skipped
.iter()
.map(|path| path.display().to_string())
.collect::<Vec<_>>(),
"hash_mismatch": self
.hash_mismatch
.iter()
.map(|path| path.display().to_string())
.collect::<Vec<_>>(),
"already_installed": self
.already_installed
.iter()
.map(|path| path.display().to_string())
.collect::<Vec<_>>(),
"keep": self.keep.as_ref().map(Render::to_json),
})
}
}
pub async fn command(m: impl Commands, args: Input) -> Std {
Std::from_result(m.install(args).await)
}
async fn install_package(
local_domain: &quilt_rs::LocalDomain,
uri: &quilt_uri::S3PackageUri,
namespace: Namespace,
) -> Result<quilt_rs::InstalledPackage, Error> {
let remote = local_domain.get_remote();
if let Some(installed_package) = local_domain.get_installed_package(&namespace).await? {
return Ok(installed_package);
}
let manifest_uri =
quilt_rs::io::manifest::resolve_manifest_uri(remote, uri.catalog.as_ref(), uri).await?;
Ok(local_domain.install_package(&manifest_uri).await?)
}
fn get_entries(
uri_path: Option<PathBuf>,
arg_paths: Option<Vec<PathBuf>>,
) -> Vec<std::path::PathBuf> {
let mut paths = Vec::new();
if let Some(logical_key) = uri_path {
paths.push(logical_key);
}
if let Some(logical_keys) = arg_paths {
for logical_key in logical_keys {
paths.push(logical_key);
}
}
paths
}
pub async fn model(
local_domain: &quilt_rs::LocalDomain,
Input {
namespace,
paths,
uri,
keep,
download,
}: Input,
) -> Result<Output, Error> {
let uri: quilt_uri::S3PackageUri = uri.parse()?;
let path = uri.path.clone();
let namespace = namespace.unwrap_or(uri.namespace.clone());
let paths = get_entries(path, paths);
if keep.is_some() && !download && !paths.is_empty() {
return Err(Error::NoDownloadWithPath);
}
let installed_package = install_package(local_domain, &uri, namespace).await?;
let report = installed_package.install_paths(&paths).await?;
let keep = match keep {
Some(scope) => Some(keep::keep(&installed_package, Some(scope), download).await?),
None => None,
};
Ok(Output {
installed_package,
paths,
skipped: report.skipped,
hash_mismatch: report.hash_mismatch,
already_installed: report.already_installed,
keep,
})
}
#[cfg(test)]
mod tests {
use super::*;
use std::path::PathBuf;
use test_log::test;
use quilt_rs::io::storage::LocalStorage;
use quilt_rs::io::storage::Storage;
use quilt_rs::paths::DomainPaths;
use crate::cli::fixtures::packages::default as pkg;
use crate::cli::model::create_model_in_temp_dir;
#[allow(
clippy::too_many_lines,
reason = "large integration test; allowed per-test so new large tests stay flagged"
)]
#[test(tokio::test)]
async fn live_model() -> Result<(), Error> {
let uri = format!("{}&path={}", pkg::URI, pkg::README_LK_ESCAPED);
let readme_logical_key = PathBuf::from(pkg::README_LK);
let timestamp_logical_key = PathBuf::from(pkg::TIMESTAMP_LK);
let (m, temp_dir) = create_model_in_temp_dir().await?;
let working_dir = temp_dir.path().join(pkg::NAMESPACE_STR);
{
let local_domain = m.get_local_domain();
let output = model(
local_domain,
Input {
namespace: None,
paths: Some(vec![timestamp_logical_key.clone()]),
uri,
keep: None,
download: true,
},
)
.await?;
assert_eq!(
format!("{output}"),
format!(
"Installed package \"{}\"\nPath: \"{}\"\nPath: \"{}\"",
pkg::NAMESPACE_STR,
pkg::README_LK,
pkg::TIMESTAMP_LK,
)
);
let installed_package = output.installed_package;
assert_eq!(installed_package.namespace, (pkg::NAMESPACE).into());
assert!(
installed_package
.lineage()
.await?
.paths
.contains_key(&readme_logical_key)
);
assert_eq!(
installed_package.package_home().await?,
PathBuf::from(temp_dir.as_ref()).join(pkg::NAMESPACE_STR)
);
assert_eq!(
output.paths,
vec![readme_logical_key.clone(), timestamp_logical_key.clone()]
);
}
let storage = LocalStorage::new();
assert!(storage.exists(working_dir.join(&readme_logical_key)).await);
assert!(
storage
.exists(working_dir.join(×tamp_logical_key))
.await
);
let domain_paths = DomainPaths::new(temp_dir.path().to_path_buf());
assert!(
storage
.exists(domain_paths.installed_manifest(&pkg::NAMESPACE.into(), pkg::TOP_HASH))
.await
);
assert!(
storage
.exists(
domain_paths
.cached_manifests_dir("data-yaml-spec-tests")
.join(pkg::TOP_HASH)
)
.await
);
assert!(
storage
.exists(
domain_paths
.objects_dir()
.join("3e5e75033079a0b5bfaeff79c8f10dbc3f461e283ad8126c333cd74792e62ea7")
)
.await
);
assert!(
storage
.exists(
domain_paths
.objects_dir()
.join("dc3ea61d9a4aaf7d822eed1de089db83d46aa29f3fbdd99466f7e5e216c91c8a")
)
.await
);
{
let local_domain = m.get_local_domain();
let install_once_more = model(
local_domain,
Input {
namespace: None,
paths: None,
uri: pkg::URI.to_string(),
keep: None,
download: true,
},
)
.await?;
assert_eq!(
format!("{install_once_more}"),
format!(
"Installed package \"{}\"\nNo paths installed",
pkg::NAMESPACE_STR,
)
);
assert_eq!(
install_once_more.installed_package.namespace,
pkg::NAMESPACE.into(),
);
let paths = install_once_more.installed_package.lineage().await?.paths;
assert!(paths.contains_key(&readme_logical_key));
assert!(paths.contains_key(×tamp_logical_key));
}
Ok(())
}
#[test(tokio::test)]
async fn live_unversioned_install_skips_a_replaced_file_and_installs_the_rest()
-> Result<(), Error> {
use crate::cli::fixtures::packages::unversioned as pkg;
let (m, temp_dir) = create_model_in_temp_dir().await?;
let working_dir = temp_dir.path().join(pkg::NAMESPACE_STR);
let requested = vec![PathBuf::from(pkg::KEPT), PathBuf::from(pkg::REPLACED)];
let output = model(
m.get_local_domain(),
Input {
namespace: None,
paths: Some(requested.clone()),
uri: pkg::R1_URI.to_string(),
keep: None,
download: true,
},
)
.await?;
assert_eq!(output.to_json()["paths"], serde_json::json!([pkg::KEPT]));
assert_eq!(
output.to_json()["skipped"],
serde_json::json!([pkg::REPLACED])
);
assert_eq!(output.to_json()["hash_mismatch"], serde_json::json!([]));
assert!(
format!("{output}").ends_with(&format!(
"Skipped: \"{}\" (no longer on the remote)",
pkg::REPLACED
)),
"{output}"
);
assert_eq!(
tokio::fs::read(working_dir.join(pkg::KEPT)).await?,
pkg::KEPT_R1_BODY
);
assert!(
!working_dir.join(pkg::REPLACED).exists(),
"r2's bytes were placed under r1's row"
);
let installed = output.get_installed_package();
let lineage = installed.lineage().await?;
assert_eq!(
lineage.paths.keys().collect::<Vec<_>>(),
vec![&PathBuf::from(pkg::KEPT)]
);
let status = installed.status(None).await?;
assert!(
status.changes.is_empty(),
"untouched files read as changed: {:?}",
status.changes.keys().collect::<Vec<_>>()
);
let retry = model(
m.get_local_domain(),
Input {
namespace: None,
paths: Some(vec![PathBuf::from(pkg::REPLACED)]),
uri: pkg::R1_URI.to_string(),
keep: None,
download: true,
},
)
.await?;
assert_eq!(
retry.to_json()["skipped"],
serde_json::json!([pkg::REPLACED])
);
assert!(!working_dir.join(pkg::REPLACED).exists());
Ok(())
}
#[test(tokio::test)]
async fn json_carries_namespace_and_paths() -> Result<(), Error> {
use crate::cli::create;
use crate::cli::model::create_model_in_temp_dir;
let (m, _temp_dir) = create_model_in_temp_dir().await?;
let created = m
.create(create::Input {
namespace: ("test", "pkg").into(),
source: None,
message: None,
})
.await?;
let output = Output {
installed_package: created.installed_package,
paths: vec![
std::path::PathBuf::from("data/one.csv"),
std::path::PathBuf::from("data/two.csv"),
std::path::PathBuf::from("data/three.csv"),
],
skipped: vec![std::path::PathBuf::from("data/two.csv")],
hash_mismatch: vec![std::path::PathBuf::from("data/three.csv")],
already_installed: Vec::new(),
keep: None,
};
assert_eq!(
output.to_json().to_string(),
r#"{"namespace":"test/pkg","paths":["data/one.csv"],"skipped":["data/two.csv"],"hash_mismatch":["data/three.csv"],"already_installed":[],"keep":null}"#
);
Ok(())
}
#[test(tokio::test)]
async fn live_tracked_path_is_already_installed_and_the_rest_install() -> Result<(), Error> {
let (m, temp_dir) = create_model_in_temp_dir().await?;
let working_dir = temp_dir.path().join(pkg::NAMESPACE_STR);
let readme = PathBuf::from(pkg::README_LK);
let timestamp = PathBuf::from(pkg::TIMESTAMP_LK);
model(
m.get_local_domain(),
Input {
namespace: None,
paths: Some(vec![readme.clone()]),
uri: pkg::URI.to_string(),
keep: None,
download: true,
},
)
.await?;
tokio::fs::write(working_dir.join(&readme), b"users edit").await?;
let output = model(
m.get_local_domain(),
Input {
namespace: None,
paths: Some(vec![readme.clone(), timestamp.clone()]),
uri: pkg::URI.to_string(),
keep: None,
download: true,
},
)
.await?;
assert_eq!(
output.to_json()["paths"],
serde_json::json!([pkg::TIMESTAMP_LK])
);
assert_eq!(
output.to_json()["already_installed"],
serde_json::json!([pkg::README_LK])
);
assert!(
format!("{output}").ends_with(&format!(
"Path: \"{}\"\nAlready installed: \"{}\"",
pkg::TIMESTAMP_LK,
pkg::README_LK
)),
"{output}"
);
assert_eq!(
tokio::fs::read(working_dir.join(&readme)).await?,
b"users edit"
);
assert!(working_dir.join(×tamp).exists());
Ok(())
}
#[test(tokio::test)]
async fn text_lists_a_path_left_out_only_as_left_out() -> Result<(), Error> {
use crate::cli::create;
use crate::cli::model::create_model_in_temp_dir;
let (m, _temp_dir) = create_model_in_temp_dir().await?;
let created = m
.create(create::Input {
namespace: ("test", "pkg").into(),
source: None,
message: None,
})
.await?;
let output = Output {
installed_package: created.installed_package,
paths: vec![
PathBuf::from("a.txt"),
PathBuf::from("b.txt"),
PathBuf::from("d.txt"),
],
skipped: vec![PathBuf::from("b.txt")],
hash_mismatch: vec![PathBuf::from("d.txt")],
already_installed: vec![PathBuf::from("c.txt")],
keep: None,
};
assert!(
format!("{output}").ends_with(concat!(
"\nPath: \"a.txt\"",
"\nAlready installed: \"c.txt\"",
"\nSkipped: \"b.txt\" (no longer on the remote)",
"\nHash mismatch: \"d.txt\" (the package may be corrupt)"
)),
"{output}"
);
Ok(())
}
async fn install_keeping_one(download: bool) -> Result<(Output, tempfile::TempDir), Error> {
let (m, temp_dir) = create_model_in_temp_dir().await?;
let output = model(
m.get_local_domain(),
Input {
namespace: None,
paths: None,
uri: pkg::URI.to_string(),
keep: Some(SyncScope::folders(["./one/"])?),
download,
},
)
.await?;
Ok((output, temp_dir))
}
#[test(tokio::test)]
async fn live_install_keeping_a_folder_downloads_its_files() -> Result<(), Error> {
let (output, _temp_dir) = install_keeping_one(true).await?;
let lineage = output.installed_package.lineage().await?;
assert_eq!(
lineage.sync_scope,
SyncScope::Folders(vec!["one".to_string()])
);
assert_eq!(
lineage.paths.into_keys().collect::<Vec<_>>(),
vec![PathBuf::from(pkg::README_LK)]
);
assert_eq!(output.to_json()["keep"]["outstanding"], 0);
Ok(())
}
#[test(tokio::test)]
async fn live_install_keeping_a_folder_without_download_fetches_nothing() -> Result<(), Error> {
let (output, _temp_dir) = install_keeping_one(false).await?;
let lineage = output.installed_package.lineage().await?;
assert_eq!(
lineage.sync_scope,
SyncScope::Folders(vec!["one".to_string()])
);
assert_eq!(lineage.paths, quilt_rs::lineage::LineagePaths::new());
assert_eq!(output.to_json()["keep"]["outstanding"], 1);
Ok(())
}
#[test(tokio::test)]
async fn no_download_refuses_a_path_in_the_uri() -> Result<(), Error> {
use crate::cli::model::create_model_in_temp_dir;
let (m, _temp_dir) = create_model_in_temp_dir().await?;
let err = model(
m.get_local_domain(),
Input {
namespace: None,
paths: None,
uri: "quilt+s3://bucket#package=a/b&path=data.csv".to_string(),
keep: Some(SyncScope::EntirePackage),
download: false,
},
)
.await
.unwrap_err();
assert!(matches!(err, Error::NoDownloadWithPath), "{err}");
Ok(())
}
#[test(tokio::test)]
async fn json_paths_are_an_empty_list_when_none_installed() -> Result<(), Error> {
use crate::cli::create;
use crate::cli::model::create_model_in_temp_dir;
let (m, _temp_dir) = create_model_in_temp_dir().await?;
let created = m
.create(create::Input {
namespace: ("test", "bare").into(),
source: None,
message: None,
})
.await?;
let output = Output {
installed_package: created.installed_package,
paths: Vec::new(),
skipped: Vec::new(),
hash_mismatch: Vec::new(),
already_installed: Vec::new(),
keep: None,
};
assert_eq!(
output.to_json().to_string(),
r#"{"namespace":"test/bare","paths":[],"skipped":[],"hash_mismatch":[],"already_installed":[],"keep":null}"#
);
Ok(())
}
}