use quilt_rs::InstalledPackage;
use quilt_rs::PushOutcome;
use quilt_rs::io::remote::HostConfig;
use quilt_rs::io::remote::Remote;
use quilt_rs::io::remote::WorkflowIntent;
use quilt_rs::io::storage::Storage;
use quilt_uri::Host;
use quilt_uri::Namespace;
use crate::cli::Error;
use crate::cli::model::Commands;
use crate::cli::output::Render;
use crate::cli::output::Std;
#[derive(Debug)]
pub struct Input {
pub namespace: Namespace,
pub host_config: Option<HostConfig>,
pub bucket: Option<String>,
pub origin: Option<Host>,
pub workflow: WorkflowIntent,
}
#[derive(Debug)]
pub struct Output {
pub hash: String,
pub manifest_uri: quilt_uri::ManifestUri,
pub certified_latest: bool,
}
impl std::fmt::Display for Output {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, r#"New revision "{}" pushed"#, self.hash)?;
if !self.certified_latest {
write!(
f,
"\nWarning: could not update latest — remote has newer changes"
)?;
}
Ok(())
}
}
impl Render for Output {
fn to_json(&self) -> serde_json::Value {
serde_json::json!({
"hash": &self.hash,
"manifest_uri": self.manifest_uri.to_string(),
"certified_latest": self.certified_latest,
})
}
}
pub async fn command(m: impl Commands, args: Input) -> Std {
Std::from_result(m.push(args).await)
}
async fn push_package(
local_domain: &quilt_rs::LocalDomain,
namespace: Namespace,
host_config: Option<HostConfig>,
bucket: Option<String>,
origin: Option<Host>,
workflow: WorkflowIntent,
) -> Result<PushOutcome, Error> {
match local_domain.get_installed_package(&namespace).await? {
Some(installed_package) => {
set_remote_and_push(&installed_package, host_config, bucket, origin, workflow).await
}
None => Err(Error::NamespaceNotFound(namespace)),
}
}
async fn set_remote_and_push<S: Storage + Clone + Sync, R: Remote>(
installed_package: &InstalledPackage<S, R>,
host_config: Option<HostConfig>,
bucket: Option<String>,
origin: Option<Host>,
workflow: WorkflowIntent,
) -> Result<PushOutcome, Error> {
if let Some(bucket) = bucket {
let outcome = installed_package
.set_remote(bucket, origin, workflow)
.await?;
if let Some(reason) = outcome.resolution_warning {
eprintln!(
"warning: couldn't resolve the bucket's default workflow ({reason}); pushing without a workflow"
);
}
}
Ok(installed_package.push(host_config).await?)
}
pub async fn model(
local_domain: &quilt_rs::LocalDomain,
Input {
namespace,
host_config,
bucket,
origin,
workflow,
}: Input,
) -> Result<Output, Error> {
let outcome = push_package(
local_domain,
namespace,
host_config,
bucket,
origin,
workflow,
)
.await?;
Ok(Output {
hash: outcome.manifest_uri.hash.clone(),
manifest_uri: outcome.manifest_uri,
certified_latest: outcome.certified_latest,
})
}
#[cfg(test)]
mod tests {
use super::*;
use test_log::test;
use quilt_rs::flow::UserMeta;
use quilt_rs::io::remote::WorkflowIntent;
use quilt_rs::io::storage::ByteStream;
use quilt_rs::io::storage::LocalStorage;
use quilt_rs::io::storage::Storage;
use std::path::PathBuf;
use crate::cli::commit;
use crate::cli::fixtures::packages::default as pkg;
use crate::cli::model::create_model_in_temp_dir;
use crate::cli::model::install_package_into_temp_dir;
#[test(tokio::test)]
async fn test_namespace_not_found() -> Result<(), Error> {
let (m, _temp_dir) = create_model_in_temp_dir().await?;
if let Std::Err(error_str) = command(
m,
Input {
namespace: ("in", "valid").into(),
host_config: None,
bucket: None,
origin: None,
workflow: WorkflowIntent::BucketDefault,
},
)
.await
{
assert_eq!(error_str.to_string(), "Package in/valid not found");
} else {
return Err(Error::Test("Expected package not found error".to_string()));
}
Ok(())
}
#[allow(
clippy::too_many_lines,
reason = "large integration test; allowed per-test so new large tests stay flagged"
)]
#[test(tokio::test)]
async fn test_push_rejected_by_mutated_workflow_surfaces_clear_error() -> Result<(), Error> {
use quilt_rs::InstalledPackage;
use quilt_rs::io::remote::Remote;
use quilt_rs::io::remote::mocks::MockRemote;
use quilt_rs::lineage::DomainLineageIo;
use quilt_rs::lineage::Home;
use quilt_rs::lineage::PackageLineageIo;
use quilt_rs::manifest::Workflow;
use quilt_rs::manifest::WorkflowId;
use quilt_rs::paths::DomainPaths;
use quilt_uri::S3Uri;
use std::collections::BTreeMap;
use tempfile::TempDir;
let home_dir = TempDir::new()?;
let paths_dir = TempDir::new()?;
let storage = LocalStorage::new();
let remote = MockRemote::default();
let namespace: Namespace = ("reference", "push-gate").into();
let home = Home::new(home_dir.path().to_path_buf());
let paths = DomainPaths::new(paths_dir.path().to_path_buf());
paths
.scaffold_for_installing(&storage, &home, &namespace)
.await?;
let lineage_json = format!(
r#"{{
"packages": {{
"reference/push-gate": {{
"commit": null,
"remote": null,
"base_hash": "",
"latest_hash": "",
"paths": {{}}
}}
}},
"home": "{}"
}}"#,
home_dir.path().display()
);
storage
.write_byte_stream(&paths.lineage(), lineage_json.as_bytes().to_vec().into())
.await?;
let config_uri = S3Uri::try_from("s3://b/.quilt/workflows/config.yml")?;
remote
.put_object(
None,
&config_uri,
b"version: \"1\"\nworkflows:\n gate:\n name: Gate\n".to_vec(),
)
.await?;
let package = InstalledPackage {
lineage: PackageLineageIo::new(
DomainLineageIo::new(paths.lineage()),
namespace.clone(),
),
paths,
remote: std::sync::Arc::new(remote),
storage,
namespace: namespace.clone(),
state_store: None,
};
let workflow = Workflow {
config: config_uri.to_string().parse()?,
id: Some(WorkflowId {
id: "gate".to_string(),
schemas: BTreeMap::new(),
}),
};
package
.commit(
"governed commit".to_string(),
UserMeta::Clear,
Some(workflow),
None,
)
.await?;
package
.set_remote(
"b".to_string(),
None,
WorkflowIntent::Named("gate".to_string()),
)
.await?;
package
.remote
.put_object(
None,
&config_uri,
b"version: \"1\"\nworkflows:\n gate:\n name: Gate\n metadata_schema: meta\nschemas:\n meta:\n url: s3://b/schemas/meta.json\n".to_vec(),
)
.await?;
package
.remote
.put_object(
None,
&S3Uri::try_from("s3://b/schemas/meta.json")?,
br#"{"type": "object", "required": ["owner"]}"#.to_vec(),
)
.await?;
let Err(err) = package.push(None).await else {
return Err(Error::Test(
"push must be rejected by the tightened workflow config".to_string(),
));
};
let message = Error::from(err).to_string();
assert!(
message.starts_with("quilt_rs error: package does not satisfy the workflow"),
"message must announce a workflow rejection, got: {message}"
);
assert!(
message.contains("metadata_schema"),
"message must name the failing rule, got: {message}"
);
assert!(
message.contains("owner"),
"message must name the missing field, got: {message}"
);
Ok(())
}
#[test(tokio::test)]
async fn test_first_push_with_bucket_and_no_host() -> Result<(), Error> {
use quilt_rs::InstalledPackage;
use quilt_rs::io::remote::mocks::MockRemote;
use quilt_rs::lineage::DomainLineageIo;
use quilt_rs::lineage::Home;
use quilt_rs::lineage::PackageLineageIo;
use quilt_rs::paths::DomainPaths;
use tempfile::TempDir;
let home_dir = TempDir::new()?;
let paths_dir = TempDir::new()?;
let storage = LocalStorage::new();
let namespace: Namespace = ("local", "only").into();
let home = Home::new(home_dir.path().to_path_buf());
let paths = DomainPaths::new(paths_dir.path().to_path_buf());
paths
.scaffold_for_installing(&storage, &home, &namespace)
.await?;
let lineage_json = serde_json::json!({
"packages": {
"local/only": {
"commit": null,
"remote": null,
"base_hash": "",
"latest_hash": "",
"paths": {}
}
},
"home": &home,
})
.to_string();
storage
.write_byte_stream(&paths.lineage(), lineage_json.as_bytes().to_vec().into())
.await?;
let package = InstalledPackage {
lineage: PackageLineageIo::new(
DomainLineageIo::new(paths.lineage()),
namespace.clone(),
),
paths,
remote: std::sync::Arc::new(MockRemote::default()),
storage,
namespace: namespace.clone(),
state_store: None,
};
package
.commit("first".to_string(), UserMeta::Keep, None, None)
.await?;
let outcome = set_remote_and_push(
&package,
None,
Some("b".to_string()),
None,
WorkflowIntent::BucketDefault,
)
.await?;
let lineage = package.lineage().await?;
let remote_uri = lineage.remote_uri.expect("the push sets the remote");
assert_eq!(remote_uri.bucket, "b");
assert_eq!(remote_uri.origin, None);
assert_eq!(remote_uri.hash, outcome.manifest_uri.hash);
assert_eq!(outcome.manifest_uri.origin, None);
Ok(())
}
#[test(tokio::test)]
async fn live_no_commit() -> Result<(), Error> {
let uri = pkg::URI;
let (m, _, _temp_dir) = install_package_into_temp_dir(uri).await?;
if let Std::Err(error_str) = command(
m,
Input {
namespace: pkg::NAMESPACE.into(),
host_config: None,
bucket: None,
origin: None,
workflow: WorkflowIntent::BucketDefault,
},
)
.await
{
assert_eq!(
error_str.to_string(),
"quilt_rs error: Push error: No commits to push"
);
} else {
return Err(Error::Test("Expected no changes error".to_string()));
}
Ok(())
}
#[test(tokio::test)]
async fn live_push_sha256_checksum() -> Result<(), Error> {
let namespace: Namespace = ("quilt_rs", "test").into();
let uri = "quilt+s3://data-yaml-spec-tests#package=quilt_rs/test";
let host_config = Some(HostConfig::default_sha256_chunked());
let (m, installed_package, _temp_dir) = install_package_into_temp_dir(uri).await?;
let initial_lineage = installed_package.lineage().await?;
let initial_hash = initial_lineage.current_hash().unwrap();
assert_eq!(
initial_hash, "4076eb7774f5159aab212302288a2a2a9e59fab69cf4e41e827072fee80fabb4",
"Initial top hash should match expected value"
);
let working_dir = installed_package.package_home().await?;
let storage = LocalStorage::new();
let e0_file_path = working_dir.join(PathBuf::from("e0-0.txt"));
storage
.write_byte_stream(
&e0_file_path,
ByteStream::from_static(b"Emperor-Drainage8-Presoak\n"),
)
.await?;
m.commit(commit::Input {
message: "Unbounded Defy 2 Landmine".to_string(),
namespace: namespace.clone(),
user_meta: UserMeta::Set(serde_json::json!({"Naturist": "Conjure"})),
workflow: WorkflowIntent::NoWorkflow,
host_config: None,
})
.await?;
let first_push_output = m
.push(Input {
namespace: namespace.clone(),
host_config: host_config.clone(),
bucket: None,
origin: None,
workflow: WorkflowIntent::BucketDefault,
})
.await?;
assert_eq!(
first_push_output.hash,
"c8027e8697016feb74b8ea523ca55934243653b890b94d64166ef2664a71ebab",
"First push top hash should match expected value"
);
storage
.write_byte_stream(
&e0_file_path,
ByteStream::from_static(b"Thu Feb 29 19:07:56 PST 2024\n"),
)
.await?;
m.commit(commit::Input {
message: "Equate 1 Fragment Grimace".to_string(),
namespace: namespace.clone(),
user_meta: UserMeta::Set(serde_json::json!({"Antitoxic": "Mankind"})),
workflow: WorkflowIntent::NoWorkflow,
host_config: None,
})
.await?;
let final_push_output = m
.push(Input {
namespace,
host_config,
bucket: None,
origin: None,
workflow: WorkflowIntent::BucketDefault,
})
.await?;
assert_eq!(
final_push_output.hash,
"4076eb7774f5159aab212302288a2a2a9e59fab69cf4e41e827072fee80fabb4",
"Final push top hash should match original expected value"
);
Ok(())
}
#[test(tokio::test)]
async fn live_push_local_package_bucket_only() -> Result<(), Error> {
use crate::cli::create;
use crate::cli::status;
use quilt_rs::lineage::UpstreamState;
let namespace: Namespace = ("cli_test", "local_push").into();
let host_config = Some(HostConfig::default_sha256_chunked());
let (m, _temp_dir) = create_model_in_temp_dir().await?;
let create_output = m
.create(create::Input {
namespace: namespace.clone(),
source: None,
message: None,
})
.await?;
let working_dir = create_output.installed_package.package_home().await?;
let storage = LocalStorage::new();
let data_file = working_dir.join("data.txt");
storage
.write_byte_stream(
&data_file,
ByteStream::from_static(b"hello from local package\n"),
)
.await?;
m.commit(commit::Input {
message: "Add data".to_string(),
namespace: namespace.clone(),
user_meta: UserMeta::Keep,
workflow: WorkflowIntent::NoWorkflow,
host_config: None,
})
.await?;
create_output
.installed_package
.set_remote(
"fiskus-us-east-1".to_string(),
None,
WorkflowIntent::BucketDefault,
)
.await?;
let push_output = m
.push(Input {
namespace: namespace.clone(),
host_config: host_config.clone(),
bucket: None,
origin: None,
workflow: WorkflowIntent::BucketDefault,
})
.await?;
assert!(
!push_output.hash.is_empty(),
"Push should return a non-empty hash"
);
let lineage = create_output.installed_package.lineage().await?;
let remote_uri = lineage
.remote_uri
.as_ref()
.expect("remote_uri should be set");
assert_eq!(
lineage.base_hash, remote_uri.hash,
"base_hash should equal remote hash after push"
);
assert!(
!remote_uri.hash.is_empty(),
"remote hash should not be empty after push"
);
assert_eq!(remote_uri.bucket, "fiskus-us-east-1", "bucket should match");
let status_output = m
.status(status::Input {
namespace: namespace.clone(),
host_config,
list_ignored: false,
})
.await?;
assert_eq!(
status_output.status.upstream_state,
UpstreamState::UpToDate,
"Package should be up-to-date after first push"
);
Ok(())
}
#[test(tokio::test)]
async fn live_push_crc64_checksum() -> Result<(), Error> {
let namespace: Namespace = ("crc64", "s3").into();
let uri = "quilt+s3://fiskus-us-east-1#package=crc64/s3";
let host_config = Some(HostConfig::default_crc64());
let (m, installed_package, _temp_dir) = install_package_into_temp_dir(uri).await?;
let initial_lineage = installed_package.lineage().await?;
let initial_hash = initial_lineage.current_hash().unwrap();
assert_eq!(
initial_hash, "b427c3867bce2445a988f69f43ad3998237d2fedf6f5e678822acd1a1e8f580a",
"Initial top hash should match expected value"
);
let working_dir = installed_package.package_home().await?;
let storage = LocalStorage::new();
let file_path = working_dir.join(PathBuf::from("1.txt"));
storage
.write_byte_stream(
&file_path,
ByteStream::from_static(b"Emperor-Drainage8-Presoak\n"),
)
.await?;
m.commit(commit::Input {
message: "Unbounded Defy 2 Landmine".to_string(),
namespace: namespace.clone(),
user_meta: UserMeta::Set(serde_json::json!({"Naturist": "Conjure"})),
workflow: WorkflowIntent::NoWorkflow,
host_config: host_config.clone(),
})
.await?;
let first_push_output = m
.push(Input {
namespace: namespace.clone(),
host_config: host_config.clone(),
bucket: None,
origin: None,
workflow: WorkflowIntent::BucketDefault,
})
.await?;
assert_eq!(
first_push_output.hash,
"8c9beced00f51cb100da861e62688e71f77a692a1c71bce422e329706ede6e63",
"First push top hash should match expected value"
);
storage
.write_byte_stream(
&file_path,
ByteStream::from_static(b"jue 27 nov 2025 16:36:45 CET\n"),
)
.await?;
m.commit(commit::Input {
message: "Initial commit".to_string(),
namespace: namespace.clone(),
user_meta: UserMeta::Clear,
workflow: WorkflowIntent::NoWorkflow,
host_config: host_config.clone(),
})
.await?;
let final_push_output = m
.push(Input {
namespace,
host_config,
bucket: None,
origin: None,
workflow: WorkflowIntent::BucketDefault,
})
.await?;
assert_eq!(
final_push_output.hash,
"b427c3867bce2445a988f69f43ad3998237d2fedf6f5e678822acd1a1e8f580a",
"Final push top hash should match original expected value"
);
Ok(())
}
#[test]
fn json_carries_hash_uri_and_certification() {
let hash = "0123456789abcdef";
let output = Output {
hash: hash.to_string(),
manifest_uri: quilt_uri::ManifestUri {
origin: None,
bucket: "bucket".to_string(),
namespace: ("demo", "sales").into(),
hash: hash.to_string(),
},
certified_latest: false,
};
assert_eq!(
output.to_json().to_string(),
r#"{"hash":"0123456789abcdef","manifest_uri":"quilt+s3://bucket#package=demo/sales@0123456789abcdef","certified_latest":false}"#
);
}
}