use std::io::{self, Read, Write};
use std::sync::atomic::{AtomicBool, Ordering};
use rstest::rstest;
use super::*;
use crate::refs::RefName;
use crate::{
Commit, CommitFields, EntryMode, ObjectId, ObjectKind, PackLimits, Repository, Signature, Tag,
TagFields, Tree, TreeEntry,
};
struct Fixture {
_root: tempfile::TempDir,
repo: Repository,
}
impl Fixture {
fn new() -> Self {
let root = tempfile::tempdir().unwrap();
std::fs::create_dir(root.path().join("objects")).unwrap();
std::fs::create_dir(root.path().join("refs")).unwrap();
std::fs::write(root.path().join("HEAD"), b"ref: refs/heads/main\n").unwrap();
std::fs::write(root.path().join("config"), b"[core]\nbare=true\n").unwrap();
let repo = Repository::open(root.path()).unwrap();
Self { _root: root, repo }
}
fn blob(&self) -> ObjectId {
self.repo.loose_objects().write_blob(b"payload").unwrap()
}
fn tree(&self, id: ObjectId, mode: EntryMode) -> ObjectId {
let tree = Tree::new(
crate::ObjectFormat::Sha1,
vec![TreeEntry {
mode,
id,
name: b"entry".to_vec(),
}],
)
.unwrap();
self.repo.loose_objects().write_tree(&tree).unwrap()
}
fn tree_with(&self, payload: &[u8]) -> ObjectId {
let blob = self.repo.loose_objects().write_blob(payload).unwrap();
self.tree(blob, EntryMode::Blob)
}
fn commit(&self, tree: ObjectId, parents: Vec<ObjectId>) -> ObjectId {
self.commit_at(tree, parents, 0)
}
fn commit_at(&self, tree: ObjectId, parents: Vec<ObjectId>, seconds: i64) -> ObjectId {
let signature = Signature {
name: b"A".to_vec(),
email: b"a@example.com".to_vec(),
seconds,
offset_minutes: 0,
};
let commit = Commit::new(CommitFields {
tree,
parents,
author: signature.clone(),
committer: signature,
extra_headers: vec![],
message: vec![],
})
.unwrap();
self.repo.loose_objects().write_commit(&commit).unwrap()
}
fn tag(&self, target: ObjectId, kind: ObjectKind) -> ObjectId {
let tag = Tag::new(TagFields {
target,
target_kind: kind,
name: b"test".to_vec(),
tagger: None,
extra_headers: vec![],
message: vec![],
})
.unwrap();
self.repo.loose_objects().write_tag(&tag).unwrap()
}
fn prepare(
&self,
commands: Vec<PushCommand>,
limits: PushLimits,
) -> Result<PreparedPush, PushFailure> {
PreparedPush::new(
&self.repo.objects(PackLimits::default()).unwrap(),
commands,
limits,
&AtomicBool::new(false),
)
}
}
fn command(name: &str, expected: Option<ObjectId>, new: ObjectId) -> PushCommand {
PushCommand {
name: RefName::new(name).unwrap(),
expected,
new,
force: ForcePolicy::FastForwardOnly,
}
}
fn tag_command(id: ObjectId) -> PushCommand {
command("refs/tags/test", None, id)
}
fn prepared(limits: PushLimits) -> PreparedPush {
let f = Fixture::new();
f.prepare(vec![tag_command(f.blob())], limits).unwrap()
}
fn pkt(bytes: &[u8]) -> Vec<u8> {
let mut result = format!("{:04x}", bytes.len() + 4).into_bytes();
result.extend(bytes);
result
}
fn advertisement(caps: &str) -> Vec<u8> {
let mut result =
pkt(format!("{} capabilities^{{}}\0{caps}\n", ObjectId::Sha1([0; 20])).as_bytes());
result.extend(b"0000");
result
}
fn response(lines: &[&[u8]]) -> Vec<u8> {
let mut result = advertisement("report-status atomic side-band-64k object-format=sha1");
for line in lines {
result.extend(pkt(line));
}
result.extend(b"0000");
result
}
fn run(bytes: &[u8], prepared: &PreparedPush) -> Result<PushReport, PushError> {
send(
&mut &*bytes,
&mut Vec::new(),
prepared,
&AtomicBool::new(false),
)
}
#[test]
fn selects_complete_graph_and_skips_external_gitlinks() {
let f = Fixture::new();
let blob = f.blob();
let subtree = f.tree(blob, EntryMode::Blob);
let tree = f.tree(subtree, EntryMode::Tree);
let external = f.tree(
ObjectId::for_blob(crate::ObjectFormat::Sha1, b"external"),
EntryMode::Gitlink,
);
let parent = f.commit(external, vec![]);
let commit = f.commit(tree, vec![parent, parent]);
let tag = f.tag(commit, ObjectKind::Commit);
let nested = f.tag(tag, ObjectKind::Tag);
let push = f
.prepare(
vec![
tag_command(nested),
command("refs/heads/main", Some(parent), commit),
],
PushLimits::default(),
)
.unwrap();
assert_eq!(push.object_count(), 8);
assert!(push.pack_bytes() > 32);
}
#[rstest]
#[case::commit_tree(ObjectKind::Commit)]
#[case::tree_child(ObjectKind::Tree)]
#[case::tag_target(ObjectKind::Tag)]
fn rejects_missing_or_mistyped_edges(#[case] kind: ObjectKind) {
let f = Fixture::new();
let blob = f.blob();
let id = bad_edge(&f, blob, kind);
assert!(matches!(
f.prepare(vec![tag_command(id)], PushLimits::default()),
Err(PushFailure::Kind(_))
));
let missing = bad_edge(
&f,
ObjectId::for_blob(crate::ObjectFormat::Sha1, b"absent"),
kind,
);
assert!(matches!(
f.prepare(vec![tag_command(missing)], PushLimits::default()),
Err(PushFailure::Missing(_))
));
}
fn bad_edge(f: &Fixture, id: ObjectId, kind: ObjectKind) -> ObjectId {
match kind {
ObjectKind::Commit => f.commit(id, vec![]),
ObjectKind::Tree => f.tree(id, EntryMode::Tree),
ObjectKind::Tag => f.tag(id, ObjectKind::Commit),
_ => unreachable!(),
}
}
#[test]
fn branch_requires_commit_even_with_force() {
let f = Fixture::new();
let mut c = command("refs/heads/main", None, f.blob());
c.force = ForcePolicy::Allow;
assert!(matches!(
f.prepare(vec![c], PushLimits::default()),
Err(PushFailure::Kind(_))
));
}
#[test]
fn missing_parent_is_not_assumed_to_exist_remotely() {
let f = Fixture::new();
let tree = f.tree(f.blob(), EntryMode::Blob);
let id = f.commit(
tree,
vec![ObjectId::for_blob(crate::ObjectFormat::Sha1, b"missing")],
);
assert!(matches!(
f.prepare(vec![tag_command(id)], PushLimits::default()),
Err(PushFailure::Missing(_))
));
}
#[test]
fn typed_edge_to_previously_selected_root_is_validated() {
let f = Fixture::new();
let blob = f.blob();
let tag = f.tag(blob, ObjectKind::Tree);
assert!(matches!(
f.prepare(
vec![tag_command(blob), command("refs/tags/other", None, tag)],
PushLimits::default()
),
Err(PushFailure::Kind(_))
));
}
#[rstest]
#[case::branch("refs/heads/main")]
#[case::tag("refs/tags/test")]
fn replacement_requires_explicit_force(#[case] name: &str) {
let f = Fixture::new();
let tree = f.tree(f.blob(), EntryMode::Blob);
let tip = f.commit(tree, vec![]);
let mut c = command(
name,
Some(ObjectId::for_blob(crate::ObjectFormat::Sha1, b"unrelated")),
tip,
);
assert!(matches!(
f.prepare(vec![c.clone()], PushLimits::default()),
Err(PushFailure::WouldForce(_))
));
c.force = ForcePolicy::Allow;
assert!(f.prepare(vec![c], PushLimits::default()).is_ok());
}
#[test]
fn proves_ancestry_along_second_merge_parent() {
let f = Fixture::new();
let tree = f.tree(f.blob(), EntryMode::Blob);
let a = f.commit(tree, vec![]);
let b = f.commit(tree, vec![a]);
let merge = f.commit(tree, vec![a, b]);
assert!(
f.prepare(
vec![command("refs/heads/main", Some(b), merge)],
PushLimits::default()
)
.is_ok()
);
}
#[rstest]
#[case::count(0, 7, 7, false)]
#[case::payload(1, 6, 7, false)]
#[case::object(1, 7, 6, false)]
#[case::exact(1, 7, 7, true)]
fn bounds_selected_objects_and_bytes(
#[case] count: u32,
#[case] total: u64,
#[case] object: u64,
#[case] ok: bool,
) {
let f = Fixture::new();
let mut limits = PushLimits::default();
limits.pack.max_objects = count;
limits.pack.max_input_bytes = total;
limits.pack.max_object_bytes = object;
assert_eq!(f.prepare(vec![tag_command(f.blob())], limits).is_ok(), ok);
}
#[test]
fn bounds_edges_and_ancestry_proof() {
let f = Fixture::new();
let tree = f.tree(f.blob(), EntryMode::Blob);
let a = f.commit(tree, vec![]);
let b = f.commit(tree, vec![a]);
let c = command("refs/heads/main", Some(a), b);
let limits = PushLimits {
max_edges: 3,
..PushLimits::default()
};
assert!(matches!(
f.prepare(vec![c.clone()], limits),
Err(PushFailure::Limit("reachable edges"))
));
let limits = PushLimits {
max_ancestry_steps: 1,
..PushLimits::default()
};
assert!(matches!(
f.prepare(vec![c.clone()], limits),
Err(PushFailure::Limit("ancestry steps"))
));
let limits = PushLimits {
max_edges: 4,
max_ancestry_steps: 3,
..PushLimits::default()
};
assert!(f.prepare(vec![c], limits).is_ok());
}
#[rstest]
#[case::head("HEAD")]
fn rejects_unsupported_namespaces(#[case] name: &str) {
let f = Fixture::new();
assert!(matches!(
f.prepare(vec![command(name, None, f.blob())], PushLimits::default()),
Err(PushFailure::Unsupported(_))
));
}
#[test]
fn accepts_arbitrary_ref_namespace() {
let f = Fixture::new();
assert!(
f.prepare(
vec![command("refs/for/main", None, f.blob())],
PushLimits::default(),
)
.is_ok()
);
}
#[test]
fn deletion_requires_lease_and_prepares_without_objects() {
let f = Fixture::new();
let name = "refs/for/main";
let old = f.blob();
let delete = command(name, Some(old), ObjectId::null(crate::ObjectFormat::Sha1));
let prepared = f.prepare(vec![delete], PushLimits::default()).unwrap();
assert_eq!(prepared.object_count(), 0);
assert_eq!(prepared.pack_bytes(), 0);
assert!(matches!(
f.prepare(
vec![command(
name,
None,
ObjectId::null(crate::ObjectFormat::Sha1)
)],
PushLimits::default(),
),
Err(PushFailure::Command(_))
));
}
#[test]
fn deletion_capability_mismatch_sends_nothing() {
let f = Fixture::new();
let id = f.blob();
let prepared = f
.prepare(
vec![command(
"refs/tags/test",
Some(id),
ObjectId::null(crate::ObjectFormat::Sha1),
)],
PushLimits::default(),
)
.unwrap();
let mut sent = vec![];
let result = send(
&mut response(&[]).as_slice(),
&mut sent,
&prepared,
&AtomicBool::new(false),
);
assert!(matches!(
result,
Err(PushError::NotSent(PushFailure::Unsupported(_)))
));
assert!(sent.is_empty());
}
#[test]
fn push_options_reject_invalid_bytes_before_sending() {
let prepared = prepared(PushLimits::default());
assert!(matches!(
prepared.with_push_options(vec![b"bad\noption".to_vec()]),
Err(PushFailure::Command("invalid push option"))
));
}
#[test]
fn rejects_duplicate_destinations_and_zero_ids() {
let f = Fixture::new();
let c = tag_command(f.blob());
assert!(matches!(
f.prepare(vec![c.clone(), c], PushLimits::default()),
Err(PushFailure::Command(_))
));
assert!(matches!(
f.prepare(
vec![tag_command(ObjectId::Sha1([0; 20]))],
PushLimits::default()
),
Err(PushFailure::Command(_))
));
assert!(matches!(
f.prepare(
vec![command(
"refs/tags/test",
Some(ObjectId::Sha1([0; 20])),
f.blob()
)],
PushLimits::default()
),
Err(PushFailure::Command(_))
));
}
#[test]
fn sends_only_report_status_then_raw_pack_and_retains_rejections() {
let p = prepared(PushLimits::default());
let mut sent = vec![];
let bytes = response(&[b"unpack ok\n", b"ng refs/tags/test hook declined\n"]);
let report = send(
&mut bytes.as_slice(),
&mut sent,
&p,
&AtomicBool::new(false),
)
.unwrap();
assert!(!report.all_succeeded());
assert_eq!(
report.refs[0].status,
Some(Status::Rejected(b"hook declined".to_vec()))
);
assert_eq!(
report.refs[0].rejection_origin,
Some(RejectionOrigin::Receiver)
);
assert_eq!(&sent[..p.request.len()], p.request);
assert_eq!(&sent[p.request.len()..p.request.len() + 4], b"PACK");
assert!(sent.windows(14).any(|w| w == b"\0report-status"));
assert!(!sent.windows(6).any(|w| w == b"atomic"));
}
#[test]
fn preserves_unpack_failure_and_binary_reason() {
let bytes = response(&[b"unpack bad pack\n", b"ng refs/tags/test reason \xff\n"]);
let report = run(&bytes, &prepared(PushLimits::default())).unwrap();
assert_eq!(report.unpack, Some(Status::Rejected(b"bad pack".to_vec())));
assert_eq!(
report.refs[0].status,
Some(Status::Rejected(b"reason \xff".to_vec()))
);
}
#[rstest]
#[case::no_status("ofs-delta")]
#[case::sha256("report-status object-format=sha256")]
fn refuses_unsupported_capabilities_before_writing(#[case] caps: &str) {
let mut written = vec![];
let result = send(
&mut advertisement(caps).as_slice(),
&mut written,
&prepared(PushLimits::default()),
&AtomicBool::new(false),
);
assert!(matches!(
result,
Err(PushError::NotSent(PushFailure::Unsupported(_)))
));
assert!(written.is_empty());
}
#[rstest]
#[case::version(b"000eversion 2\n0000")]
#[case::header(b"zzzz")]
#[case::length(b"0001")]
#[case::truncated(b"0008abc")]
#[case::missing(b"0000")]
fn invalid_advertisement_never_sends_commands(#[case] bytes: &[u8]) {
let mut written = vec![];
assert!(matches!(
send(
&mut &*bytes,
&mut written,
&prepared(PushLimits::default()),
&AtomicBool::new(false)
),
Err(PushError::NotSent(_))
));
assert!(written.is_empty());
}
#[test]
fn advertised_stale_value_keeps_exact_lease_on_wire() {
let p = prepared(PushLimits::default());
let mut bytes = pkt(format!("{} refs/tags/test\0report-status", p.commands[0].new).as_bytes());
bytes.extend(b"0000");
bytes.extend(pkt(b"unpack ok\n"));
bytes.extend(pkt(b"ng refs/tags/test stale old value\n"));
bytes.extend(b"0000");
let mut written = vec![];
let report = send(
&mut bytes.as_slice(),
&mut written,
&p,
&AtomicBool::new(false),
)
.unwrap();
assert_eq!(
report.refs[0].status,
Some(Status::Rejected(b"stale old value".to_vec()))
);
assert_eq!(
report.refs[0].rejection_origin,
Some(RejectionOrigin::Receiver)
);
assert_eq!(
written,
[p.request.as_slice(), p.pack.to_vec().as_slice()].concat()
);
}
#[rstest]
#[case::missing_unpack(&[])]
#[case::missing_ref(&[b"unpack ok".as_slice()])]
#[case::wrong_ref(&[b"unpack ok".as_slice(), b"ok refs/tags/other".as_slice()])]
#[case::duplicate(&[b"unpack ok".as_slice(), b"ok refs/tags/test".as_slice(), b"ng refs/tags/test duplicate".as_slice()])]
#[case::v2_option(&[b"unpack ok".as_slice(), b"ok refs/tags/test".as_slice(), b"option forced-update".as_slice()])]
#[case::contradiction(&[b"unpack bad".as_slice(), b"ok refs/tags/test".as_slice()])]
#[case::empty_unpack(&[b"unpack ".as_slice()])]
#[case::empty_reason(&[b"unpack ok".as_slice(), b"ng refs/tags/test ".as_slice()])]
#[case::remote_err(&[b"ERR failure".as_slice()])]
fn malformed_status_is_uncertain(#[case] lines: &[&[u8]]) {
assert!(matches!(
run(&response(lines), &prepared(PushLimits::default())),
Err(PushError::Uncertain { .. })
));
}
#[test]
fn truncated_status_preserves_valid_prefix() {
let mut bytes = response(&[b"unpack ok", b"ok refs/tags/test"]);
bytes.truncate(bytes.len() - 2);
let Err(PushError::Uncertain { report, .. }) = run(&bytes, &prepared(PushLimits::default()))
else {
panic!("expected uncertain");
};
assert_eq!(report.unpack, Some(Status::Ok));
assert_eq!(report.refs[0].status, Some(Status::Ok));
}
#[test]
fn report_status_v2_retains_proc_receive_rewrite() {
let prepared = prepared(PushLimits::default());
let actual = prepared.commands()[0].new;
let mut bytes = advertisement("report-status-v2 object-format=sha1");
for line in [
b"unpack ok".to_vec(),
b"ok refs/tags/test".to_vec(),
b"option refname refs/heads/rewritten".to_vec(),
format!("option new-oid {actual}").into_bytes(),
b"option forced-update".to_vec(),
] {
bytes.extend(pkt(&line));
}
bytes.extend(b"0000");
let mut written = Vec::new();
let report = send(
&mut bytes.as_slice(),
&mut written,
&prepared,
&AtomicBool::new(false),
)
.unwrap();
assert!(report.all_succeeded());
assert!(
written
.windows(b"report-status-v2".len())
.any(|part| part == b"report-status-v2")
);
assert_eq!(report.refs[0].rewrites.len(), 1);
assert_eq!(
report.refs[0].rewrites[0].name,
Some(RefName::new("refs/heads/rewritten").unwrap())
);
assert_eq!(report.refs[0].rewrites[0].new, Some(actual));
assert!(report.refs[0].rewrites[0].forced);
}
#[test]
fn malformed_v2_option_leaves_current_command_uncertain() {
let prepared = prepared(PushLimits::default());
let mut bytes = advertisement("report-status-v2");
bytes.extend(pkt(b"unpack ok"));
bytes.extend(pkt(b"ok refs/tags/test"));
bytes.extend(pkt(b"option unknown value"));
bytes.extend(b"0000");
let Err(PushError::Uncertain { report, .. }) = run(&bytes, &prepared) else {
panic!("expected uncertain outcome");
};
assert_eq!(report.refs[0].status, None);
}
#[test]
fn sideband_progress_and_status_are_kept_separate() {
let prepared = prepared(PushLimits::default()).with_progress();
let mut bytes = advertisement("report-status side-band-64k");
let mut status = pkt(b"unpack ok");
status.extend(pkt(b"ok refs/tags/test"));
status.extend(b"0000");
let mut progress = vec![2];
progress.extend(b"checking objects\r");
bytes.extend(pkt(&progress));
let mut result = vec![1];
result.extend(status);
bytes.extend(pkt(&result));
bytes.extend(b"0000");
let mut written = Vec::new();
let report = send(
&mut bytes.as_slice(),
&mut written,
&prepared,
&AtomicBool::new(false),
)
.unwrap();
assert!(report.all_succeeded());
assert_eq!(report.progress, vec![b"checking objects\r".to_vec()]);
assert!(
written
.windows(b"side-band-64k".len())
.any(|part| part == b"side-band-64k")
);
}
#[test]
fn interrupted_sideband_keeps_completed_status_and_progress() {
let prepared = prepared(PushLimits::default()).with_progress();
let mut bytes = advertisement("report-status side-band-64k");
bytes.extend(pkt(b"\x02checking objects\n"));
let mut status = vec![1];
status.extend(pkt(b"unpack ok"));
status.extend(pkt(b"ok refs/tags/test"));
status.extend(b"0000");
bytes.extend(pkt(&status));
let Err(PushError::Uncertain { report, .. }) = run(&bytes, &prepared) else {
panic!("expected uncertain transport completion");
};
assert_eq!(report.refs[0].status, Some(Status::Ok));
assert_eq!(report.progress, vec![b"checking objects\n".to_vec()]);
}
#[test]
fn rejects_trailing_bytes_after_complete_report() {
let mut bytes = response(&[b"unpack ok", b"ok refs/tags/test"]);
bytes.extend(b"extra");
assert!(matches!(
run(&bytes, &prepared(PushLimits::default())),
Err(PushError::Uncertain { .. })
));
}
#[test]
fn bounds_advertisement_and_status_before_allocating_packets() {
let p = prepared(PushLimits {
max_advertisement_bytes: 4,
..PushLimits::default()
});
assert!(matches!(
run(&advertisement("report-status"), &p),
Err(PushError::NotSent(PushFailure::Limit(_)))
));
let p = prepared(PushLimits {
max_status_bytes: 4,
..PushLimits::default()
});
assert!(matches!(
run(&response(&[b"unpack ok"]), &p),
Err(PushError::Uncertain {
cause: PushFailure::Limit(_),
..
})
));
}
#[test]
fn empty_push_only_flushes() {
let f = Fixture::new();
let p = f.prepare(vec![], PushLimits::default()).unwrap();
let mut out = vec![];
assert!(
send(
&mut advertisement("").as_slice(),
&mut out,
&p,
&AtomicBool::new(false)
)
.unwrap()
.all_succeeded()
);
assert_eq!(out, b"0000");
assert_eq!(p.pack_bytes(), 0);
}
struct Interrupted;
impl Read for Interrupted {
fn read(&mut self, _: &mut [u8]) -> io::Result<usize> {
Err(io::ErrorKind::Interrupted.into())
}
}
impl Write for Interrupted {
fn write(&mut self, _: &[u8]) -> io::Result<usize> {
Err(io::ErrorKind::Interrupted.into())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
#[test]
fn interrupted_io_is_not_retried_and_is_classified_by_phase() {
let p = prepared(PushLimits::default());
assert!(
matches!(send(&mut Interrupted, &mut vec![], &p, &AtomicBool::new(false)), Err(PushError::NotSent(PushFailure::Io(e))) if e.kind() == io::ErrorKind::Interrupted)
);
assert!(
matches!(send(&mut advertisement("report-status").as_slice(), &mut Interrupted, &p, &AtomicBool::new(false)), Err(PushError::NotSent(PushFailure::Io(e))) if e.kind() == io::ErrorKind::Interrupted)
);
}
struct CancelWriter<'a>(&'a AtomicBool);
impl Write for CancelWriter<'_> {
fn write(&mut self, _: &[u8]) -> io::Result<usize> {
self.0.store(true, Ordering::Relaxed);
Ok(1)
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
#[test]
fn cancellation_before_and_during_transmission_has_distinct_outcomes() {
let p = prepared(PushLimits::default());
let cancel = AtomicBool::new(true);
assert!(matches!(
send(
&mut advertisement("report-status").as_slice(),
&mut vec![],
&p,
&cancel
),
Err(PushError::NotSent(PushFailure::Cancelled))
));
cancel.store(false, Ordering::Relaxed);
assert!(matches!(
send(
&mut advertisement("report-status").as_slice(),
&mut CancelWriter(&cancel),
&p,
&cancel
),
Err(PushError::Uncertain {
cause: PushFailure::Cancelled,
..
})
));
}
#[test]
fn preparation_cancellation_changes_no_storage() {
let f = Fixture::new();
let objects = f.repo.objects(PackLimits::default()).unwrap();
assert!(matches!(
PreparedPush::new(
&objects,
vec![tag_command(ObjectId::for_blob(
crate::ObjectFormat::Sha1,
b"absent"
))],
PushLimits::default(),
&AtomicBool::new(true)
),
Err(PushFailure::Cancelled)
));
assert_eq!(std::fs::read_dir(f.repo.object_dir()).unwrap().count(), 0);
}
#[test]
fn bounds_command_count_and_encoded_bytes_exactly() {
let f = Fixture::new();
let c = tag_command(f.blob());
let limits = PushLimits {
max_commands: 0,
..PushLimits::default()
};
assert!(matches!(
f.prepare(vec![c.clone()], limits),
Err(PushFailure::Limit("commands"))
));
let p = f.prepare(vec![c.clone()], PushLimits::default()).unwrap();
let limits = PushLimits {
max_command_bytes: p.request.len() - 1,
..PushLimits::default()
};
assert!(matches!(
f.prepare(vec![c.clone()], limits),
Err(PushFailure::Limit("command bytes"))
));
let limits = PushLimits {
max_command_bytes: p.request.len(),
..PushLimits::default()
};
assert!(f.prepare(vec![c], limits).is_ok());
}
#[test]
fn bounds_generated_pack_and_index() {
let f = Fixture::new();
let c = tag_command(f.blob());
let p = f.prepare(vec![c.clone()], PushLimits::default()).unwrap();
let mut limits = PushLimits::default();
limits.pack.max_pack_bytes = p.pack_bytes() as u64 - 1;
assert!(matches!(
f.prepare(vec![c.clone()], limits),
Err(PushFailure::Pack(_))
));
limits.pack.max_pack_bytes += 1;
limits.pack.max_index_bytes = 1099;
assert!(matches!(
f.prepare(vec![c.clone()], limits),
Err(PushFailure::Pack(_))
));
limits.pack.max_index_bytes = 1100; assert!(f.prepare(vec![c], limits).is_ok());
}
#[test]
fn exact_wire_budgets_allow_complete_report() {
let bytes = response(&[b"unpack ok", b"ok refs/tags/test"]);
let adv = advertisement("report-status atomic side-band-64k object-format=sha1").len();
let p = prepared(PushLimits {
max_advertisement_bytes: adv,
max_status_bytes: bytes.len() - adv,
..PushLimits::default()
});
assert!(run(&bytes, &p).unwrap().all_succeeded());
}
#[test]
fn bounds_advertised_refs_and_accepts_unselected_have_entries() {
let p = prepared(PushLimits {
max_refs: 0,
..PushLimits::default()
});
let mut bytes = pkt(format!("{} .have\0report-status", p.commands[0].new).as_bytes());
bytes.extend(b"0000");
assert!(matches!(
run(&bytes, &p),
Err(PushError::NotSent(PushFailure::Limit("advertised refs")))
));
bytes.extend(pkt(b"unpack ok"));
bytes.extend(pkt(b"ok refs/tags/test"));
bytes.extend(b"0000");
assert!(
run(&bytes, &prepared(PushLimits::default()))
.unwrap()
.all_succeeded()
);
}
#[test]
fn returns_partial_results_in_command_order_even_if_server_reorders() {
let f = Fixture::new();
let id = f.blob();
let p = f
.prepare(
vec![tag_command(id), command("refs/tags/other", None, id)],
PushLimits::default(),
)
.unwrap();
let bytes = response(&[
b"unpack ok",
b"ng refs/tags/other rejected",
b"ok refs/tags/test",
]);
let report = run(&bytes, &p).unwrap();
assert_eq!(report.refs[0].status, Some(Status::Ok));
assert_eq!(
report.refs[1].status,
Some(Status::Rejected(b"rejected".to_vec()))
);
}
struct ShortWriter(Vec<u8>);
impl Write for ShortWriter {
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
self.0.push(bytes[0]);
Ok(1)
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
#[test]
fn short_writes_preserve_full_request_and_pack() {
let p = prepared(PushLimits::default());
let bytes = response(&[b"unpack ok", b"ok refs/tags/test"]);
let mut writer = ShortWriter(vec![]);
assert!(
send(
&mut bytes.as_slice(),
&mut writer,
&p,
&AtomicBool::new(false)
)
.unwrap()
.all_succeeded()
);
assert_eq!(
writer.0,
[p.request.as_slice(), p.pack.to_vec().as_slice()].concat()
);
}
struct FailingWriter {
bytes: usize,
fail_flush: bool,
}
impl Write for FailingWriter {
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
if self.bytes == 0 {
return Err(io::ErrorKind::BrokenPipe.into());
}
let n = bytes.len().min(self.bytes);
self.bytes -= n;
Ok(n)
}
fn flush(&mut self) -> io::Result<()> {
if self.fail_flush {
Err(io::ErrorKind::BrokenPipe.into())
} else {
Ok(())
}
}
}
#[rstest]
#[case::pack(200, false)]
#[case::flush(usize::MAX, true)]
fn later_output_failures_are_uncertain(#[case] bytes: usize, #[case] fail_flush: bool) {
let p = prepared(PushLimits::default());
let mut writer = FailingWriter { bytes, fail_flush };
assert!(matches!(
send(
&mut advertisement("report-status").as_slice(),
&mut writer,
&p,
&AtomicBool::new(false)
),
Err(PushError::Uncertain { .. })
));
}
#[test]
fn failed_first_write_is_not_sent() {
let p = prepared(PushLimits::default());
let mut writer = FailingWriter {
bytes: 0,
fail_flush: false,
};
assert!(matches!(
send(
&mut advertisement("report-status").as_slice(),
&mut writer,
&p,
&AtomicBool::new(false)
),
Err(PushError::NotSent(PushFailure::Io(_)))
));
}
#[test]
fn partial_command_upload_marks_later_ref_not_attempted() {
let fixture = Fixture::new();
let id = fixture.blob();
let prepared = fixture
.prepare(
vec![tag_command(id), command("refs/tags/later", None, id)],
PushLimits::default(),
)
.unwrap();
let first_packet =
usize::from_str_radix(std::str::from_utf8(&prepared.request[..4]).unwrap(), 16).unwrap();
let mut writer = FailingWriter {
bytes: first_packet,
fail_flush: false,
};
let Err(PushError::Uncertain { report, .. }) = send(
&mut advertisement("report-status").as_slice(),
&mut writer,
&prepared,
&AtomicBool::new(false),
) else {
panic!("expected uncertain partial upload");
};
assert!(report.refs[0].attempted);
assert!(!report.refs[1].attempted);
assert_eq!(report.refs[0].status, None);
assert_eq!(report.refs[1].status, None);
}
#[test]
fn interrupted_read_during_report_is_uncertain() {
let p = prepared(PushLimits::default());
let ad = advertisement("report-status");
let mut reader = ad.as_slice().chain(Interrupted);
assert!(
matches!(send(&mut reader, &mut vec![], &p, &AtomicBool::new(false)), Err(PushError::Uncertain { cause: PushFailure::Io(e), .. }) if e.kind() == io::ErrorKind::Interrupted)
);
}
#[rstest]
#[case::tree(ObjectKind::Tree)]
#[case::commit(ObjectKind::Commit)]
#[case::tag(ObjectKind::Tag)]
fn malformed_reachable_payload_is_rejected(#[case] kind: ObjectKind) {
let f = Fixture::new();
let id = raw_object(&f, kind, b"invalid");
let error = f
.prepare(vec![tag_command(id)], PushLimits::default())
.unwrap_err();
assert!(error.to_string().contains(&id.to_string()));
assert!(std::error::Error::source(&error).is_some());
}
fn raw_object(f: &Fixture, kind: ObjectKind, data: &[u8]) -> ObjectId {
let id = ObjectId::for_object(kind.as_str(), data);
let hex = id.to_string();
let directory = f.repo.object_dir().join(&hex[..2]);
std::fs::create_dir_all(&directory).unwrap();
let mut encoder = flate2::write::ZlibEncoder::new(vec![], flate2::Compression::default());
encoder
.write_all(format!("{} {}\0", kind.as_str(), data.len()).as_bytes())
.unwrap();
encoder.write_all(data).unwrap();
std::fs::write(directory.join(&hex[2..]), encoder.finish().unwrap()).unwrap();
id
}
#[test]
fn ancestry_budget_charges_duplicate_parent_edges() {
let f = Fixture::new();
let tree = f.tree(f.blob(), EntryMode::Blob);
let old = f.commit(tree, vec![]);
let new = f.commit(tree, vec![old, old, old]);
let limits = PushLimits {
max_ancestry_steps: 4,
..PushLimits::default()
};
assert!(matches!(
f.prepare(vec![command("refs/heads/main", Some(old), new)], limits),
Err(PushFailure::Limit("ancestry steps"))
));
}
#[test]
fn incomplete_multi_ref_report_keeps_unknown_distinct_from_rejected() {
let f = Fixture::new();
let id = f.blob();
let p = f
.prepare(
vec![tag_command(id), command("refs/tags/other", None, id)],
PushLimits::default(),
)
.unwrap();
let bytes = response(&[b"unpack ok", b"ng refs/tags/test rejected"]);
let Err(PushError::Uncertain { report, .. }) = run(&bytes, &p) else {
panic!("expected incomplete report");
};
assert_eq!(
report.refs[0].status,
Some(Status::Rejected(b"rejected".to_vec()))
);
assert_eq!(report.refs[1].status, None);
assert!(!report.all_succeeded());
}
#[test]
fn excludes_nested_tags_and_skips_external_gitlinks() {
let f = Fixture::new();
let tree = f.tree(
ObjectId::for_blob(crate::ObjectFormat::Sha1, b"external"),
EntryMode::Gitlink,
);
let inner = f.tag(tree, ObjectKind::Tree);
let outer = f.tag(inner, ObjectKind::Tag);
let prepared = PreparedPush::new_excluding(
&f.repo.objects(PackLimits::default()).unwrap(),
vec![tag_command(outer)],
&[inner, inner],
PushLimits::default(),
&AtomicBool::new(false),
)
.unwrap();
assert_eq!(prepared.object_count(), 1);
assert_eq!(prepared.receiver_roots, vec![inner]);
}
#[test]
fn receiver_root_closure_is_trusted_without_local_validation() {
let f = Fixture::new();
let tree = f.tree(
ObjectId::for_blob(crate::ObjectFormat::Sha1, b"missing"),
EntryMode::Blob,
);
let prepared = PreparedPush::new_excluding(
&f.repo.objects(PackLimits::default()).unwrap(),
vec![tag_command(tree)],
&[tree],
PushLimits::default(),
&AtomicBool::new(false),
)
.unwrap();
assert_eq!(prepared.object_count(), 0);
assert_eq!(prepared.receiver_roots, vec![tree]);
}
#[test]
fn missing_new_object_fails_despite_receiver_roots() {
let f = Fixture::new();
let known = f.tree_with(b"known");
let old = f.commit(known, vec![]);
let tree = f.tree(
ObjectId::for_blob(crate::ObjectFormat::Sha1, b"missing"),
EntryMode::Blob,
);
let new = f.commit(tree, vec![old]);
let result = PreparedPush::new_excluding(
&f.repo.objects(PackLimits::default()).unwrap(),
vec![command("refs/heads/main", Some(old), new)],
&[old],
PushLimits::default(),
&AtomicBool::new(false),
);
assert!(matches!(result, Err(PushFailure::Missing(_))));
}
#[test]
fn unrelated_present_root_is_not_relied_on() {
let f = Fixture::new();
let unrelated = f.repo.loose_objects().write_blob(b"unrelated").unwrap();
let tip = f.commit(f.tree_with(b"tip"), vec![]);
let prepared = PreparedPush::new_excluding(
&f.repo.objects(PackLimits::default()).unwrap(),
vec![command("refs/heads/main", None, tip)],
&[unrelated],
PushLimits::default(),
&AtomicBool::new(false),
)
.unwrap();
assert_eq!(prepared.object_count(), 3);
assert!(prepared.receiver_roots.is_empty());
}
#[rstest]
#[case::exhausted(3, Some("push limit exceeded: exclusion edges"))]
#[case::exact(4, None)]
fn bounds_exclusion_edges_separately(#[case] max_edges: usize, #[case] error: Option<&str>) {
let f = Fixture::new();
let entries = [b"a", b"b", b"c"].map(|name| TreeEntry {
mode: EntryMode::Blob,
id: f.repo.loose_objects().write_blob(name).unwrap(),
name: name.to_vec(),
});
let known = Tree::new(crate::ObjectFormat::Sha1, entries.to_vec()).unwrap();
let old = f.commit(f.repo.loose_objects().write_tree(&known).unwrap(), vec![]);
let new = f.commit(f.tree_with(b"new"), vec![old]);
let result = PreparedPush::new_excluding(
&f.repo.objects(PackLimits::default()).unwrap(),
vec![command("refs/heads/main", Some(old), new)],
&[old],
PushLimits {
max_edges,
..PushLimits::default()
},
&AtomicBool::new(false),
);
assert_eq!(result.err().map(|e| e.to_string()).as_deref(), error);
}
fn history(f: &Fixture, spec: &str) -> Vec<ObjectId> {
let mut ids: Vec<ObjectId> = Vec::new();
for (index, entry) in spec.split_whitespace().enumerate() {
let (parents, seconds) = entry.split_once('@').unwrap();
let parents = parents
.split(',')
.filter(|p| !p.is_empty())
.map(|p| ids[p.parse::<usize>().unwrap()])
.collect();
let tree = f.tree_with(format!("commit {index}").as_bytes());
ids.push(f.commit_at(tree, parents, seconds.parse().unwrap()));
}
ids
}
fn prepare_over(
f: &Fixture,
old: ObjectId,
new: ObjectId,
observe: impl FnMut(PreparationProgress),
) -> Result<PreparedPush, PushFailure> {
PreparedPush::new_local_with_progress(
&f.repo.objects(PackLimits::default()).unwrap(),
vec![command("refs/heads/main", Some(old), new)],
&[old],
PushLimits::default(),
&AtomicBool::new(false),
observe,
)
}
#[rstest]
#[case::short(16)]
#[case::long(1000)]
fn incremental_push_reads_are_independent_of_known_history(#[case] length: usize) {
let f = Fixture::new();
let spec: Vec<_> = (1..length).map(|i| format!("{}@{i}", i - 1)).collect();
let old = *history(&f, &format!("@0 {}", spec.join(" ")))
.last()
.unwrap();
let new = f.commit_at(f.tree_with(b"new"), vec![old], length as i64);
let mut reads = 0;
let prepared = prepare_over(&f, old, new, |event| {
if let PreparationProgress::Reading { objects } = event {
reads = objects;
}
})
.unwrap();
assert_eq!(prepared.object_count(), 3);
assert_eq!(prepared.receiver_roots, vec![old]);
assert_eq!(reads, 10);
}
#[rstest]
#[case::ancestor_dated_after_descendants("@900 0@100 1@200", 0, 2, false)]
#[case::parent_dated_after_child("@100 0@300 1@200 2@150", 1, 3, false)]
#[case::second_parent_skewed("@100 0@900 0@200 2,1@300", 1, 3, false)]
#[case::deep_ancestor_behind_newer_side("@100 0@110 1@120 0@800 2,3@130", 0, 4, false)]
#[case::older_sibling("@100 0@150 0@200", 1, 2, true)]
#[case::newer_sibling("@100 0@900 0@200", 1, 2, true)]
#[case::rewind("@100 0@200", 1, 0, true)]
#[case::rewind_skewed("@900 0@200", 1, 0, true)]
#[case::unrelated("@100 @200 1@50", 0, 2, true)]
#[case::sibling_of_merge_parent("@100 0@300 0@50 0@60 1,2@400", 3, 4, true)]
fn fast_forward_proof_ignores_commit_dates(
#[case] spec: &str,
#[case] old: usize,
#[case] new: usize,
#[case] rejected: bool,
#[values(0, 1)] roots: usize,
) {
let f = Fixture::new();
let ids = history(&f, spec);
let result = PreparedPush::new_excluding(
&f.repo.objects(PackLimits::default()).unwrap(),
vec![command("refs/heads/main", Some(ids[old]), ids[new])],
&[ids[old]][..roots],
PushLimits::default(),
&AtomicBool::new(false),
);
assert_eq!(matches!(result, Err(PushFailure::WouldForce(_))), rejected);
assert_eq!(result.is_ok(), !rejected);
}
#[rstest]
#[case::ordered("@100 0@200 0@300 1,2@400")]
#[case::known_first_parent_newest("@100 0@900 0@300 1,2@400")]
#[case::new_side_older_than_base("@500 0@600 0@10 1,2@700")]
#[case::new_side_first("@100 0@200 0@300 2,1@400")]
#[case::merge_oldest("@100 0@200 0@300 1,2@1")]
fn merge_sends_only_new_side(#[case] spec: &str) {
let f = Fixture::new();
let ids = history(&f, spec);
let prepared = prepare_over(&f, ids[1], ids[3], |_| {}).unwrap();
assert_eq!(prepared.object_count(), 6);
assert_eq!(prepared.receiver_roots, vec![ids[1]]);
}
fn skewed_history(f: &Fixture) -> Vec<ObjectId> {
history(f, "@500 0@100 1@600 2@700 3@50")
}
fn tag_new_commit(f: &Fixture, ids: &[ObjectId]) -> ObjectId {
f.tag(ids[4], ObjectKind::Commit)
}
fn nested_tag_of_known_commit(f: &Fixture, ids: &[ObjectId]) -> ObjectId {
let inner = f.tag(ids[1], ObjectKind::Commit);
f.tag(inner, ObjectKind::Tag)
}
fn tag_of_known_tree(f: &Fixture, _: &[ObjectId]) -> ObjectId {
f.tag(f.tree_with(b"commit 2"), ObjectKind::Tree)
}
fn tag_of_known_blob(f: &Fixture, _: &[ObjectId]) -> ObjectId {
let blob = f.repo.loose_objects().write_blob(b"commit 3").unwrap();
f.tag(blob, ObjectKind::Blob)
}
fn tag_of_new_tree(f: &Fixture, _: &[ObjectId]) -> ObjectId {
f.tag(f.tree_with(b"tagged"), ObjectKind::Tree)
}
#[rstest]
#[case::new_commit(tag_new_commit as fn(&Fixture, &[ObjectId]) -> ObjectId, 4)]
#[case::nested_known_commit(nested_tag_of_known_commit, 2)]
#[case::known_tree(tag_of_known_tree, 3)]
#[case::known_blob(tag_of_known_blob, 2)]
#[case::new_tree(tag_of_new_tree, 3)]
fn tags_send_objects_outside_marked_closure(
#[case] tag: fn(&Fixture, &[ObjectId]) -> ObjectId,
#[case] objects: u32,
) {
let f = Fixture::new();
let ids = skewed_history(&f);
let tag = tag(&f, &ids);
let prepared = PreparedPush::new_excluding(
&f.repo.objects(PackLimits::default()).unwrap(),
vec![tag_command(tag)],
&[ids[3]],
PushLimits::default(),
&AtomicBool::new(false),
)
.unwrap();
assert_eq!(prepared.object_count(), objects);
}
#[test]
fn bounds_receiver_root_occurrences() {
let f = Fixture::new();
let blob = f.blob();
let result = PreparedPush::new_excluding(
&f.repo.objects(PackLimits::default()).unwrap(),
vec![tag_command(blob)],
&[blob, blob],
PushLimits {
max_refs: 1,
..PushLimits::default()
},
&AtomicBool::new(false),
);
assert!(matches!(result, Err(PushFailure::Limit("receiver roots"))));
}
#[test]
fn advertised_have_can_prove_exclusion_without_matching_a_command() {
let f = Fixture::new();
let blob = f.blob();
let prepared = PreparedPush::new_excluding(
&f.repo.objects(PackLimits::default()).unwrap(),
vec![tag_command(blob)],
&[blob],
PushLimits::default(),
&AtomicBool::new(false),
)
.unwrap();
let mut bytes = pkt(format!("{blob} .have\0report-status\n").as_bytes());
bytes.extend(b"0000");
bytes.extend(pkt(b"unpack ok"));
bytes.extend(pkt(b"ok refs/tags/test"));
bytes.extend(b"0000");
assert!(run(&bytes, &prepared).unwrap().all_succeeded());
}
#[test]
fn ref_delta_push_needs_only_report_status() {
let f = Fixture::new();
let mut data: Vec<u8> = (0..65536).map(|n| (n % 251) as u8).collect();
let loose = f.repo.loose_objects();
let first = loose.write_blob(&data).unwrap();
data[4000] ^= 255;
let second = loose.write_blob(&data).unwrap();
let commands = vec![
command("refs/tags/a", None, first),
command("refs/tags/b", None, second),
];
let ordinary = f.prepare(commands.clone(), PushLimits::default()).unwrap();
let p = f
.prepare(
commands,
PushLimits {
compression: crate::PackCompression::Delta(crate::DeltaOptions::default()),
..PushLimits::default()
},
)
.unwrap();
assert!(p.pack_bytes() < ordinary.pack_bytes());
let mut bytes = advertisement("report-status");
bytes.extend(pkt(b"unpack ok"));
bytes.extend(pkt(b"ok refs/tags/a"));
bytes.extend(pkt(b"ok refs/tags/b"));
bytes.extend(b"0000");
let mut sent = vec![];
let report = send(
&mut bytes.as_slice(),
&mut sent,
&p,
&AtomicBool::new(false),
)
.unwrap();
assert!(report.all_succeeded());
assert!(!p.request.windows(9).any(|w| w == b"ofs-delta"));
assert_eq!(&sent[p.request.len()..], p.pack.to_vec());
}
#[rstest]
#[case::new_id(None, ObjectId::Sha256([1; 32]))]
#[case::old_id(Some(ObjectId::Sha256([1; 32])), ObjectId::Sha1([1; 20]))]
fn refuses_wrong_format_before_graph_reads(#[case] old: Option<ObjectId>, #[case] new: ObjectId) {
let fixture = Fixture::new();
let result = fixture.prepare(
vec![command("refs/tags/test", old, new)],
PushLimits::default(),
);
assert!(matches!(result, Err(PushFailure::ObjectFormat(_))));
}
#[rstest]
#[case::sha1(crate::ObjectFormat::Sha1)]
#[case::sha256(crate::ObjectFormat::Sha256)]
fn preparation_progress_observes_source_and_pack(#[case] format: crate::ObjectFormat) {
let root = tempfile::tempdir().unwrap();
let repo = Repository::init(format, root.path().join("source"), crate::InitKind::Bare).unwrap();
let id = repo.loose_objects().write_blob(b"source bytes").unwrap();
let objects = repo.objects(PackLimits::default()).unwrap();
let mut events = Vec::new();
let prepared = PreparedPush::new_local_with_progress(
&objects,
vec![tag_command(id)],
&[],
PushLimits::default(),
&AtomicBool::new(false),
|event| events.push(event),
)
.unwrap();
assert_eq!(prepared.object_count(), 1);
assert_eq!(
events,
[
PreparationProgress::Reading { objects: 0 },
PreparationProgress::Reading { objects: 1 },
PreparationProgress::Packing { objects: (0, 1) },
PreparationProgress::Packing { objects: (1, 1) },
PreparationProgress::Complete
]
);
}
#[test]
fn preparation_progress_counts_shared_objects_once_and_excludes_known_pack_entries() {
let f = Fixture::new();
let blob = f.blob();
let tree = f.tree(blob, EntryMode::Blob);
let objects = f.repo.objects(PackLimits::default()).unwrap();
let mut events = Vec::new();
let prepared = PreparedPush::new_local_with_progress(
&objects,
vec![tag_command(tree), command("refs/tags/also", None, tree)],
&[blob],
PushLimits::default(),
&AtomicBool::new(false),
|event| events.push(event),
)
.unwrap();
assert_eq!(prepared.object_count(), 1);
assert_eq!(
events,
[
PreparationProgress::Reading { objects: 0 },
PreparationProgress::Reading { objects: 1 },
PreparationProgress::Reading { objects: 2 },
PreparationProgress::Packing { objects: (0, 1) },
PreparationProgress::Packing { objects: (1, 1) },
PreparationProgress::Complete
]
);
}
#[rstest]
#[case::reading_start(PreparationProgress::Reading { objects: 0 })]
#[case::reading(PreparationProgress::Reading { objects: 1 })]
#[case::packing_start(PreparationProgress::Packing { objects: (0, 1) })]
#[case::packing(PreparationProgress::Packing { objects: (1, 1) })]
fn preparation_progress_cancellation_has_no_prepared_result(#[case] stop: PreparationProgress) {
let f = Fixture::new();
let objects = f.repo.objects(PackLimits::default()).unwrap();
let cancel = AtomicBool::new(false);
let mut events = Vec::new();
let result = PreparedPush::new_local_with_progress(
&objects,
vec![tag_command(f.blob())],
&[],
PushLimits::default(),
&cancel,
|event| {
events.push(event);
if event == stop {
cancel.store(true, Ordering::Relaxed);
}
},
);
assert!(matches!(result, Err(PushFailure::Cancelled)));
assert!(!events.contains(&PreparationProgress::Complete));
assert_eq!(events.last(), Some(&stop));
}
#[test]
fn preparation_progress_completion_is_not_retroactively_cancelled() {
let f = Fixture::new();
let objects = f.repo.objects(PackLimits::default()).unwrap();
let cancel = AtomicBool::new(false);
let prepared = PreparedPush::new_local_with_progress(
&objects,
vec![tag_command(f.blob())],
&[],
PushLimits::default(),
&cancel,
|event| {
if event == PreparationProgress::Complete {
cancel.store(true, Ordering::Relaxed);
}
},
)
.unwrap();
assert_eq!(prepared.object_count(), 1);
assert!(cancel.load(Ordering::Relaxed));
}
#[test]
fn preparation_progress_empty_batch_has_no_pack_events() {
let f = Fixture::new();
let objects = f.repo.objects(PackLimits::default()).unwrap();
let mut events = Vec::new();
let prepared = PreparedPush::new_local_with_progress(
&objects,
vec![],
&[],
PushLimits::default(),
&AtomicBool::new(false),
|event| events.push(event),
)
.unwrap();
assert_eq!(prepared.pack_bytes(), 0);
assert_eq!(
events,
[
PreparationProgress::Reading { objects: 0 },
PreparationProgress::Complete
]
);
}
#[test]
fn preparation_progress_deletion_only_has_no_pack_events() {
let f = Fixture::new();
let objects = f.repo.objects(PackLimits::default()).unwrap();
let delete = command(
"refs/tags/test",
Some(f.blob()),
ObjectId::null(crate::ObjectFormat::Sha1),
);
let mut events = Vec::new();
let prepared = PreparedPush::new_local_with_progress(
&objects,
vec![delete],
&[],
PushLimits::default(),
&AtomicBool::new(false),
|event| events.push(event),
)
.unwrap();
assert_eq!(prepared.pack_bytes(), 0);
assert_eq!(
events,
[
PreparationProgress::Reading { objects: 0 },
PreparationProgress::Complete
]
);
}
#[test]
fn preparation_progress_missing_object_has_no_completion() {
let f = Fixture::new();
let objects = f.repo.objects(PackLimits::default()).unwrap();
let missing = ObjectId::for_blob(crate::ObjectFormat::Sha1, b"absent");
let mut events = Vec::new();
let result = PreparedPush::new_local_with_progress(
&objects,
vec![tag_command(missing)],
&[],
PushLimits::default(),
&AtomicBool::new(false),
|event| events.push(event),
);
assert!(matches!(result, Err(PushFailure::Missing(id)) if id == missing));
assert_eq!(events, [PreparationProgress::Reading { objects: 0 }]);
}
#[test]
fn preparation_progress_index_failure_has_no_completion() {
let f = Fixture::new();
let objects = f.repo.objects(PackLimits::default()).unwrap();
let mut limits = PushLimits::default();
limits.pack.max_index_bytes = 0;
let mut events = Vec::new();
let result = PreparedPush::new_local_with_progress(
&objects,
vec![tag_command(f.blob())],
&[],
limits,
&AtomicBool::new(false),
|event| events.push(event),
);
assert!(matches!(
result,
Err(PushFailure::Pack(crate::PackWriteError::Limit(
"index bytes"
)))
));
assert_eq!(
events.last(),
Some(&PreparationProgress::Packing { objects: (1, 1) })
);
assert!(!events.contains(&PreparationProgress::Complete));
}
#[test]
fn preparation_progress_preserves_prepared_bytes() {
let f = Fixture::new();
let objects = f.repo.objects(PackLimits::default()).unwrap();
let commands = vec![tag_command(f.blob())];
let cancel = AtomicBool::new(false);
let original = PreparedPush::new_local(
&objects,
commands.clone(),
&[],
PushLimits::default(),
&cancel,
)
.unwrap();
let observed = PreparedPush::new_local_with_progress(
&objects,
commands,
&[],
PushLimits::default(),
&cancel,
|_| {},
)
.unwrap();
assert_eq!(observed.request, original.request);
assert_eq!(observed.pack.to_vec(), original.pack.to_vec());
}
#[test]
fn prepared_pack_is_spooled_and_removed_on_drop() {
let f = Fixture::new();
let id = f.blob();
let prepared = f
.prepare(
vec![command("refs/tags/test", None, id)],
PushLimits::default(),
)
.unwrap();
let path = prepared.pack.path().expect("push must spool").to_owned();
assert_eq!(
std::fs::metadata(&path).unwrap().len() as usize,
prepared.pack_bytes()
);
drop(prepared);
assert!(!path.exists());
}
#[test]
fn fresh_transfer_deltas_obey_the_receiver_depth_limit() {
let f = Fixture::new();
let first = f
.repo
.loose_objects()
.write_blob(&b"abcdef\n".repeat(4096))
.unwrap();
let mut next = b"abcdef\n".repeat(4096);
next.extend_from_slice(b"changed\n");
let second = f.repo.loose_objects().write_blob(&next).unwrap();
let prepared = f
.prepare(
vec![
command("refs/tags/first", None, first),
command("refs/tags/second", None, second),
],
PushLimits {
read: crate::ReadLimits {
max_delta_depth: 0,
..crate::ReadLimits::default()
},
compression: crate::PackCompression::Delta(crate::DeltaOptions::default()),
..PushLimits::default()
},
)
.unwrap();
let imported = crate::fetch::ReceivedFetch::validate_known(
crate::fetch::Advertisement {
refs: vec![],
capabilities: vec![],
},
vec![first, second],
prepared.pack,
&crate::fetch::KnownHistory::default(),
vec![],
crate::fetch::FetchOptions {
limits: crate::fetch::FetchLimits {
max_delta_depth: 0,
..crate::fetch::FetchLimits::default()
},
depth: None,
},
&mut crate::fetch::ValidationObserver::new(&AtomicBool::new(false), &mut |_| {}),
)
.unwrap();
assert_eq!(imported.object_count(), 2);
}