quilt-rs 0.44.0

//! Tests for the commit/push/pull/status lifecycle of an installed package.

use std::future::Future;

use super::*;
use quilt_uri::fixtures;

use test_log::test;

use aws_sdk_s3::primitives::ByteStream;

use crate::flow::PullOutcome;
use crate::io::remote::mocks::MockRemote;
use crate::io::storage::StorageExt;
use crate::lineage::DomainLineageIo;
use crate::lineage::Home;
use crate::lineage::PackageLineageIo;
use crate::lineage::SyncScope;
use crate::object_hash::ObjectHash;
use crate::paths::DomainPaths;

#[test(tokio::test)]
async fn test_spamming_commit_writes() -> Res {
    let (home, _temp_dir1) = Home::from_temp_dir()?;
    let (paths, _temp_dir2) = DomainPaths::from_temp_dir()?;

    let storage = LocalStorage::new();
    let remote = MockRemote::default();
    let namespace: Namespace = ("test", "history").into();
    let test_hash = "deadbeef".to_string();

    paths
        .scaffold_for_installing(&storage, &home, &namespace)
        .await?;
    // Initialize domain lineage file
    let lineage_json = format!(
        r#"{{
            "packages": {{
                "test/history": {{
                    "commit": null,
                    "remote": {{
                        "bucket": "bucket",
                        "namespace": "test/history",
                        "hash": "{}",
                        "catalog": "quilt.test"
                    }},
                    "base_hash": "{}",
                    "latest_hash": "{}",
                    "paths": {{}}
                }}}},
            "home": "/tmp/working_dir"
            }}"#,
        test_hash, "foo", "bar"
    );
    storage
        .write_byte_stream(&paths.lineage(), lineage_json.as_bytes().to_vec().into())
        .await?;

    // Copy manifest to the expected path
    let test_manifest_path = paths.installed_manifest(&namespace, &test_hash);
    let test_manifest = r#"{"version": "v0"}"#;
    storage
        .write_byte_stream(
            &test_manifest_path,
            ByteStream::from_static(test_manifest.as_bytes()),
        )
        .await?;

    let domain_lineage_io = DomainLineageIo::new(paths.lineage());

    let package = InstalledPackage {
        lineage: PackageLineageIo::new(domain_lineage_io, namespace.clone()),
        paths,
        remote: std::sync::Arc::new(remote),
        storage,
        namespace,
    };

    // Make 10 commits with different content
    let mut expected_hashes = Vec::new();
    for i in 0..10 {
        let commit = package
            .commit(
                format!("Commit new1 {i}"),
                UserMeta::Set(serde_json::json!({ "count": i })),
                None,
                None,
            )
            .await?;
        expected_hashes.insert(i, commit.hash);
    }

    // Remove last, cause it's the "current" hash, not a part of `prev_hashes`
    expected_hashes.pop();

    let commit_state = package.lineage().await?.commit.unwrap();

    assert_eq!(commit_state.prev_hashes.len(), 9);
    // let hashes_to_assert: Vec<String> = expected_hashes.into_iter().rev().collect();
    assert_eq!(
        commit_state.prev_hashes,
        expected_hashes.into_iter().rev().collect::<Vec<String>>()
    );

    Ok(())
}

/// Scenario A: diverged with an unpushed local commit. The user lands on
/// the merge page because someone else moved `latest` past our install
/// base, and we have a local commit on top of that base. `certify_latest`
/// must push our commit and then tag the resulting remote hash as
/// `latest` — not roll the tag back to the install-time hash.
#[test(tokio::test)]
async fn test_certify_latest_pushes_pending_commit_then_tags() -> Res {
    let (home, _temp_dir1) = Home::from_temp_dir()?;
    let (paths, _temp_dir2) = DomainPaths::from_temp_dir()?;

    let storage = LocalStorage::new();
    let remote = MockRemote::default();
    let namespace: Namespace = ("test", "diverged").into();
    let bucket = "b";

    // L is the rebuilt-manifest hash that push will produce from an empty
    // manifest with header user_meta=null. Using a known fixture keeps
    // push's "rebuilt hash must equal commit hash" check happy without
    // wiring real objects.
    let local_hash = crate::fixtures::top_hash::EMPTY_NULL_TOP_HASH;
    let install_hash = "I_HASH";
    let other_hash = "N_HASH";

    paths
        .scaffold_for_installing(&storage, &home, &namespace)
        .await?;
    paths.scaffold_for_caching(&storage, bucket).await?;

    let lineage_json = format!(
        r#"{{
            "packages": {{
                "test/diverged": {{
                    "commit": {{
                        "timestamp": "2024-01-01T00:00:00Z",
                        "hash": "{local_hash}",
                        "prev_hashes": []
                    }},
                    "remote": {{
                        "bucket": "{bucket}",
                        "namespace": "test/diverged",
                        "hash": "{install_hash}"
                    }},
                    "base_hash": "{install_hash}",
                    "latest_hash": "{other_hash}",
                    "paths": {{}}
                }}
            }},
            "home": "/tmp/working_dir"
        }}"#
    );
    storage
        .write_byte_stream(&paths.lineage(), lineage_json.as_bytes().to_vec().into())
        .await?;

    // Local installed manifest at the commit hash — push reads it via
    // `self.manifest()`. These exact bytes are what make the rebuild
    // produce `EMPTY_NULL_TOP_HASH` (matching `local_hash` above); any
    // change to the manifest serialization or top-hash algorithm will
    // surface as a push-side "rebuilt hash != commit hash" failure
    // rather than at the final `latest_body` assertion.
    let local_manifest = b"{\"version\":\"v0\",\"message\":\"\",\"user_meta\":null}\n".to_vec();
    storage
        .write_byte_stream(
            &paths.installed_manifest(&namespace, local_hash),
            local_manifest.clone().into(),
        )
        .await?;

    // Pre-cache the install-time remote manifest so push's `flow::browse`
    // call for the previous remote_uri succeeds without a remote round-trip.
    let install_manifest_uri = ManifestUri {
        bucket: bucket.to_string(),
        namespace: namespace.clone(),
        hash: install_hash.to_string(),
        origin: None,
    };
    storage
        .write_byte_stream(
            paths.cached_manifest(&install_manifest_uri),
            local_manifest.into(),
        )
        .await?;

    // Remote `latest` tag points at someone else's hash — this is what
    // makes the state Diverged.
    remote
        .put_object(
            None,
            &S3Uri::try_from(
                format!("s3://{bucket}/.quilt/named_packages/test/diverged/latest").as_str(),
            )?,
            other_hash.as_bytes().to_vec(),
        )
        .await?;

    let domain_lineage_io = DomainLineageIo::new(paths.lineage());
    let package = InstalledPackage {
        lineage: PackageLineageIo::new(domain_lineage_io, namespace.clone()),
        paths,
        remote: std::sync::Arc::new(remote),
        storage,
        namespace,
    };

    // Certify pushes the pending commit under the lock it already holds; a
    // push that took the package's lock again would wait on itself forever.
    tokio::time::timeout(std::time::Duration::from_secs(30), package.certify_latest())
        .await
        .expect("certify must not wait on its own lock")?;

    // Remote `latest` now points at the user's revision (L), not the
    // teammate's (N) and not the install-time hash (I).
    let latest_uri = S3Uri::try_from(
        format!("s3://{bucket}/.quilt/named_packages/test/diverged/latest").as_str(),
    )?;
    let latest_body = package
        .remote
        .get_object_stream(None, &latest_uri)
        .await?
        .body
        .collect()
        .await?
        .to_vec();
    assert_eq!(latest_body, local_hash.as_bytes());

    let lineage = package.lineage().await?;
    assert_eq!(lineage.base_hash, local_hash);
    assert_eq!(lineage.latest_hash, local_hash);
    assert!(lineage.commit.is_none(), "push should have consumed commit");

    Ok(())
}

/// Scenario B: diverged because our prior push uploaded the manifest but
/// `push_package` declined to certify (`latest` had moved between
/// `base_hash` and our push). With `commit = None`, `certify_latest`
/// must skip the push and tag the already-pushed hash as `latest`.
#[test(tokio::test)]
async fn test_certify_latest_skips_push_when_no_pending_commit() -> Res {
    let (home, _temp_dir1) = Home::from_temp_dir()?;
    let (paths, _temp_dir2) = DomainPaths::from_temp_dir()?;

    let storage = LocalStorage::new();
    let remote = MockRemote::default();
    let namespace: Namespace = ("test", "pushed").into();
    let bucket = "b";

    let pushed_hash = "X_HASH";
    let other_hash = "Y_HASH";

    paths
        .scaffold_for_installing(&storage, &home, &namespace)
        .await?;
    paths.scaffold_for_caching(&storage, bucket).await?;

    let lineage_json = format!(
        r#"{{
            "packages": {{
                "test/pushed": {{
                    "commit": null,
                    "remote": {{
                        "bucket": "{bucket}",
                        "namespace": "test/pushed",
                        "hash": "{pushed_hash}"
                    }},
                    "base_hash": "{pushed_hash}",
                    "latest_hash": "{other_hash}",
                    "paths": {{}}
                }}
            }},
            "home": "/tmp/working_dir"
        }}"#
    );
    storage
        .write_byte_stream(&paths.lineage(), lineage_json.as_bytes().to_vec().into())
        .await?;

    // Remote `latest` tag currently points at someone else's hash.
    remote
        .put_object(
            None,
            &S3Uri::try_from(
                format!("s3://{bucket}/.quilt/named_packages/test/pushed/latest").as_str(),
            )?,
            other_hash.as_bytes().to_vec(),
        )
        .await?;

    let domain_lineage_io = DomainLineageIo::new(paths.lineage());
    let package = InstalledPackage {
        lineage: PackageLineageIo::new(domain_lineage_io, namespace.clone()),
        paths,
        remote: std::sync::Arc::new(remote),
        storage,
        namespace,
    };

    package.certify_latest().await?;

    // Remote `latest` now points at the previously-pushed revision (X).
    let latest_uri = S3Uri::try_from(
        format!("s3://{bucket}/.quilt/named_packages/test/pushed/latest").as_str(),
    )?;
    let latest_body = package
        .remote
        .get_object_stream(None, &latest_uri)
        .await?
        .body
        .collect()
        .await?
        .to_vec();
    assert_eq!(latest_body, pushed_hash.as_bytes());

    // No manifest was uploaded as part of certification — push was skipped.
    assert!(
        !package
            .remote
            .exists(
                None,
                &S3Uri::try_from(format!("s3://{bucket}/.quilt/packages/{pushed_hash}").as_str(),)?,
            )
            .await?,
        "push should be skipped when there is no pending commit",
    );

    let lineage = package.lineage().await?;
    assert_eq!(lineage.base_hash, pushed_hash);
    assert_eq!(lineage.latest_hash, pushed_hash);
    assert!(lineage.commit.is_none());

    Ok(())
}

#[test(tokio::test)]
async fn test_manifest_recovery_from_corruption() -> Res {
    let (home, _temp_dir1) = Home::from_temp_dir()?;
    let (paths, _temp_dir2) = DomainPaths::from_temp_dir()?;

    let storage = LocalStorage::new();
    let remote = MockRemote::default();
    let namespace: Namespace = ("test", "recovery").into();
    let test_hash = "deadbeef".to_string();

    paths
        .scaffold_for_installing(&storage, &home, &namespace)
        .await?;
    paths.scaffold_for_caching(&storage, "test-bucket").await?;

    // Initialize domain lineage file
    let lineage_json = format!(
        r#"{{
            "packages": {{
                "test/recovery": {{
                    "commit": null,
                    "remote": {{
                        "bucket": "test-bucket",
                        "namespace": "test/recovery",
                        "hash": "{}",
                        "catalog": null
                    }},
                    "base_hash": "{}",
                    "latest_hash": "{}",
                    "paths": {{}}
                }}}},
            "home": "/tmp/working_dir"
            }}"#,
        test_hash, "foo", "bar"
    );
    storage
        .write_byte_stream(&paths.lineage(), lineage_json.as_bytes().to_vec().into())
        .await?;

    // Set up a valid cached manifest
    let reference_manifest = crate::fixtures::manifest::path();
    let manifest_uri = ManifestUri {
        bucket: "test-bucket".to_string(),
        namespace: namespace.clone(),
        hash: test_hash.clone(),
        origin: None,
    };
    let cached_manifest = paths.cached_manifest(&manifest_uri);
    storage.copy(reference_manifest?, cached_manifest).await?;

    // Create a corrupted installed manifest
    let installed_manifest = paths.installed_manifest(&namespace, &test_hash);
    storage
        .write_byte_stream(
            &installed_manifest,
            ByteStream::from_static(b"corrupted data"),
        )
        .await?;

    let domain_lineage_io = DomainLineageIo::new(paths.lineage());
    let package = InstalledPackage {
        lineage: PackageLineageIo::new(domain_lineage_io, namespace.clone()),
        paths,
        remote: std::sync::Arc::new(remote),
        storage: storage.clone(),
        namespace,
    };

    // This should succeed by recovering from cache despite corrupted installed manifest
    let result = package.manifest().await;
    assert!(
        result.is_ok(),
        "Should recover from cache when installed is corrupted"
    );

    // Verify the corrupted file was replaced with good data
    let fixed_manifest_content = storage.read_bytes(&installed_manifest).await?;
    assert!(
        fixed_manifest_content.len() > 10,
        "Installed manifest should be fixed"
    );
    assert!(
        !fixed_manifest_content.starts_with(b"corrupted"),
        "Should no longer be corrupted"
    );

    Ok(())
}

/// A remote that always returns `LoginRequired`, simulating a logged-out user.
/// Refuses with the route where S3 rejected an issued credential, rather than
/// the provider refusing to vend. `status()` must treat both as a dead session.
struct RejectedCredentialRemote;

impl crate::io::remote::Remote for RejectedCredentialRemote {
    fn exists(&self, _host: Option<&Host>, _s3_uri: &S3Uri) -> impl Future<Output = Res<bool>> {
        std::future::ready(Err(rejected_credential()))
    }
    fn get_object_stream(
        &self,
        _host: Option<&Host>,
        _s3_uri: &S3Uri,
    ) -> impl Future<Output = Res<crate::io::remote::RemoteObjectStream>> {
        std::future::ready(Err(rejected_credential()))
    }
    fn resolve_url(
        &self,
        _host: Option<&Host>,
        _s3_uri: &S3Uri,
    ) -> impl Future<Output = Res<S3Uri>> {
        std::future::ready(Err(rejected_credential()))
    }
    fn put_object(
        &self,
        _host: Option<&Host>,
        _s3_uri: &S3Uri,
        _contents: impl Into<aws_sdk_s3::primitives::ByteStream>,
    ) -> impl Future<Output = Res> {
        std::future::ready(Err(rejected_credential()))
    }
    fn upload_file(
        &self,
        _host_config: &crate::io::remote::HostConfig,
        _source_path: impl AsRef<std::path::Path>,
        _dest_uri: &S3Uri,
        _size: u64,
    ) -> impl Future<Output = Res<(S3Uri, crate::object_hash::ObjectHash)>> {
        std::future::ready(Err(rejected_credential()))
    }
    fn host_config(
        &self,
        _host: Option<&Host>,
    ) -> impl Future<Output = Res<crate::io::remote::HostConfig>> {
        std::future::ready(Err(rejected_credential()))
    }
    fn verify_bucket(&self, _bucket: &str) -> impl Future<Output = Res> {
        std::future::ready(Err(rejected_credential()))
    }
    fn published_revisions(
        &self,
        _host: &Host,
        _bucket: &str,
        _namespace: &Namespace,
    ) -> impl Future<Output = Res<Vec<String>>> {
        std::future::ready(Err(rejected_credential()))
    }
}

fn rejected_credential() -> Error {
    Error::S3(crate::error::S3Error {
        host: Some(fixtures::host()),
        kind: crate::error::S3ErrorKind::InvalidCredentials("ExpiredToken: nope".to_string()),
    })
}

pub(super) struct LoggedOutRemote;

impl crate::io::remote::Remote for LoggedOutRemote {
    fn exists(&self, _host: Option<&Host>, _s3_uri: &S3Uri) -> impl Future<Output = Res<bool>> {
        std::future::ready(Err(Error::Login(LoginError::NoSession(None))))
    }
    fn get_object_stream(
        &self,
        _host: Option<&Host>,
        _s3_uri: &S3Uri,
    ) -> impl Future<Output = Res<crate::io::remote::RemoteObjectStream>> {
        std::future::ready(Err(Error::Login(LoginError::NoSession(None))))
    }
    fn resolve_url(
        &self,
        _host: Option<&Host>,
        _s3_uri: &S3Uri,
    ) -> impl Future<Output = Res<S3Uri>> {
        std::future::ready(Err(Error::Login(LoginError::NoSession(None))))
    }
    fn put_object(
        &self,
        _host: Option<&Host>,
        _s3_uri: &S3Uri,
        _contents: impl Into<aws_sdk_s3::primitives::ByteStream>,
    ) -> impl Future<Output = Res> {
        std::future::ready(Err(Error::Login(LoginError::NoSession(None))))
    }
    fn upload_file(
        &self,
        _host_config: &crate::io::remote::HostConfig,
        _source_path: impl AsRef<std::path::Path>,
        _dest_uri: &S3Uri,
        _size: u64,
    ) -> impl Future<Output = Res<(S3Uri, ObjectHash)>> {
        std::future::ready(Err(Error::Login(LoginError::NoSession(None))))
    }
    fn host_config(
        &self,
        _host: Option<&Host>,
    ) -> impl Future<Output = Res<crate::io::remote::HostConfig>> {
        std::future::ready(Ok(crate::io::remote::HostConfig::default()))
    }
    fn verify_bucket(&self, _bucket: &str) -> impl Future<Output = Res> {
        std::future::ready(Ok(()))
    }
    fn published_revisions(
        &self,
        _host: &Host,
        _bucket: &str,
        _namespace: &Namespace,
    ) -> impl Future<Output = Res<Vec<String>>> {
        std::future::ready(Err(Error::Login(LoginError::NoSession(None))))
    }
}

/// A remote that refuses every read with `AccessDenied`, simulating a role
/// that cannot reach the bucket. Distinct from [`LoggedOutRemote`]: the
/// credentials are valid, the request arrived, and it was refused.
pub(super) struct DeniedRemote;

impl crate::io::remote::Remote for DeniedRemote {
    fn exists(&self, _host: Option<&Host>, s3_uri: &S3Uri) -> impl Future<Output = Res<bool>> {
        std::future::ready(Err(denied(s3_uri)))
    }
    fn get_object_stream(
        &self,
        _host: Option<&Host>,
        s3_uri: &S3Uri,
    ) -> impl Future<Output = Res<crate::io::remote::RemoteObjectStream>> {
        std::future::ready(Err(denied(s3_uri)))
    }
    fn resolve_url(
        &self,
        _host: Option<&Host>,
        s3_uri: &S3Uri,
    ) -> impl Future<Output = Res<S3Uri>> {
        std::future::ready(Err(denied(s3_uri)))
    }
    fn put_object(
        &self,
        _host: Option<&Host>,
        s3_uri: &S3Uri,
        _contents: impl Into<aws_sdk_s3::primitives::ByteStream>,
    ) -> impl Future<Output = Res> {
        std::future::ready(Err(denied(s3_uri)))
    }
    fn upload_file(
        &self,
        _host_config: &crate::io::remote::HostConfig,
        _source_path: impl AsRef<std::path::Path>,
        dest_uri: &S3Uri,
        _size: u64,
    ) -> impl Future<Output = Res<(S3Uri, ObjectHash)>> {
        std::future::ready(Err(denied(dest_uri)))
    }
    fn host_config(
        &self,
        _host: Option<&Host>,
    ) -> impl Future<Output = Res<crate::io::remote::HostConfig>> {
        std::future::ready(Ok(crate::io::remote::HostConfig::default()))
    }
    fn verify_bucket(&self, _bucket: &str) -> impl Future<Output = Res> {
        std::future::ready(Ok(()))
    }
    fn published_revisions(
        &self,
        _host: &Host,
        bucket: &str,
        namespace: &Namespace,
    ) -> impl Future<Output = Res<Vec<String>>> {
        let pointers = S3Uri {
            bucket: bucket.to_string(),
            key: crate::paths::tag_key(namespace, ""),
            version: None,
        };
        std::future::ready(Err(denied(&pointers)))
    }
}

fn denied(s3_uri: &S3Uri) -> Error {
    crate::error::S3Error::new(crate::error::S3ErrorKind::AccessDenied(s3_uri.to_string())).into()
}

/// The latest-hash pointer read is the *first* thing a narrow role hits, so
/// it is the most common shape of a denial in practice. `status` swallows
/// every other refresh failure by design — that is what lets the app work
/// offline on stale lineage — but swallowing this one leaves the UI unable
/// to tell the user why the row will not sync, and sends it looking for an
/// auth problem that does not exist.
#[test(tokio::test)]
async fn test_status_propagates_access_denied_from_the_latest_hash_read() -> Res {
    let (home, _temp_dir1) = Home::from_temp_dir()?;
    let (paths, _temp_dir2) = DomainPaths::from_temp_dir()?;

    let storage = LocalStorage::new();
    let namespace: Namespace = ("test", "denied").into();

    paths
        .scaffold_for_installing(&storage, &home, &namespace)
        .await?;

    // An installed package with a remote and a manifest already on disk, so
    // the only remote read `status` needs is the latest-hash pointer.
    let installed_hash = "abcdef";
    let lineage_json = format!(
        r#"{{
        "packages": {{
            "test/denied": {{
                "commit": null,
                "remote": {{
                    "bucket": "locked",
                    "namespace": "test/denied",
                    "hash": "{installed_hash}",
                    "origin": "quilt.test"
                }},
                "base_hash": "{installed_hash}",
                "latest_hash": "{installed_hash}",
                "paths": {{}}
            }}
        }},
        "home": "/tmp/working_dir"
    }}"#
    );
    storage
        .write_byte_stream(&paths.lineage(), lineage_json.as_bytes().to_vec().into())
        .await?;
    storage
        .write_byte_stream(
            &paths.installed_manifest(&namespace, installed_hash),
            ByteStream::from_static(br#"{"version": "v0"}"#),
        )
        .await?;

    let domain_lineage_io = DomainLineageIo::new(paths.lineage());
    let package = InstalledPackage {
        lineage: PackageLineageIo::new(domain_lineage_io, namespace.clone()),
        paths,
        remote: std::sync::Arc::new(DeniedRemote),
        storage,
        namespace,
    };

    let result = package.status(None).await;

    let Err(err) = result else {
        panic!("a denied latest-hash read must not report a healthy status");
    };
    assert!(
        err.is_access_denied(),
        "the denial must survive as a typed denial, not a warning: {err}"
    );

    Ok(())
}

/// The other route to a dead session, asserted through `status()` itself.
///
/// **Does not isolate the guarded arm.** This stub refuses every read, and both
/// the tag refresh and the manifest read go through `get_object_stream`, so the
/// error also reaches the caller via the manifest `?`. It pins the end-to-end
/// claim — a rejected credential leaves `status()` as a session failure — not
/// the specific arm that stops the swallow-and-continue fallback. Isolating
/// that needs a stub that refuses the tag read and serves the manifest.
#[test(tokio::test)]
async fn test_status_propagates_a_rejected_credential() -> Res {
    let (home, _temp_dir1) = Home::from_temp_dir()?;
    let (paths, _temp_dir2) = DomainPaths::from_temp_dir()?;

    let storage = LocalStorage::new();
    let namespace: Namespace = ("test", "rejectedcred").into();

    paths
        .scaffold_for_installing(&storage, &home, &namespace)
        .await?;

    // Package with remote configured but never pushed (empty hash)
    let lineage_json = r#"{
        "packages": {
            "test/rejectedcred": {
                "commit": null,
                "remote": {
                    "bucket": "my-bucket",
                    "namespace": "test/rejectedcred",
                    "hash": "",
                    "origin": "quilt.test"
                },
                "base_hash": "",
                "latest_hash": "",
                "paths": {}
            }
        },
        "home": "/tmp/working_dir"
    }"#;
    storage
        .write_byte_stream(&paths.lineage(), lineage_json.as_bytes().to_vec().into())
        .await?;

    let domain_lineage_io = DomainLineageIo::new(paths.lineage());
    let package = InstalledPackage {
        lineage: PackageLineageIo::new(domain_lineage_io, namespace.clone()),
        paths,
        remote: std::sync::Arc::new(RejectedCredentialRemote),
        storage,
        namespace,
    };

    // The seam a mocked status call hides: the swallow-and-continue arm
    // below must not take this, or a dead session reports as a healthy package.
    let result = package.status(None).await;
    assert!(
        result.as_ref().is_err_and(Error::is_session_absent),
        "Expected LoginRequired error, got: {result:?}"
    );

    Ok(())
}

#[test(tokio::test)]
async fn test_status_propagates_login_required() -> Res {
    let (home, _temp_dir1) = Home::from_temp_dir()?;
    let (paths, _temp_dir2) = DomainPaths::from_temp_dir()?;

    let storage = LocalStorage::new();
    let namespace: Namespace = ("test", "needslogin").into();

    paths
        .scaffold_for_installing(&storage, &home, &namespace)
        .await?;

    // Package with remote configured but never pushed (empty hash)
    let lineage_json = r#"{
        "packages": {
            "test/needslogin": {
                "commit": null,
                "remote": {
                    "bucket": "my-bucket",
                    "namespace": "test/needslogin",
                    "hash": "",
                    "origin": "quilt.test"
                },
                "base_hash": "",
                "latest_hash": "",
                "paths": {}
            }
        },
        "home": "/tmp/working_dir"
    }"#;
    storage
        .write_byte_stream(&paths.lineage(), lineage_json.as_bytes().to_vec().into())
        .await?;

    let domain_lineage_io = DomainLineageIo::new(paths.lineage());
    let package = InstalledPackage {
        lineage: PackageLineageIo::new(domain_lineage_io, namespace.clone()),
        paths,
        remote: std::sync::Arc::new(LoggedOutRemote),
        storage,
        namespace,
    };

    // status() should propagate LoginRequired so the UI can show a Login button
    let result = package.status(None).await;
    assert!(
        matches!(result, Err(Error::Login(LoginError::NoSession(_)))),
        "Expected LoginRequired error, got: {result:?}"
    );

    Ok(())
}

/// Pull must refresh `latest_hash` from the remote before evaluating
/// `flow::pull`'s `base_hash == latest_hash` guard. Before the
/// "Stop writing lineage from `InstalledPackage::status`" refactor, a
/// prior `status` call would persist the refreshed `latest_hash`, so
/// disk was reliably fresh when `pull` ran. Without that persist,
/// the disk-stale `latest_hash` always equalled `base_hash` and
/// pull short-circuited with "already up-to-date" — both the
/// autosync watcher's pull branch and the manual Pull button were
/// affected.
#[test(tokio::test)]
async fn test_pull_refreshes_latest_hash_when_remote_moved() -> Res {
    let (home, _temp_dir1) = Home::from_temp_dir()?;
    let (paths, _temp_dir2) = DomainPaths::from_temp_dir()?;
    let storage = LocalStorage::new();
    let remote = MockRemote::default();
    let namespace: Namespace = ("test", "pull_refresh").into();
    let bucket = "bkt";
    let install_hash = "INSTALL_HASH";
    let new_hash = "NEW_HASH";

    paths
        .scaffold_for_installing(&storage, &home, &namespace)
        .await?;

    // Disk lineage at the install-time hash: latest_hash == base_hash
    // == remote.hash. Without the refresh inside `pull` this state
    // alone short-circuits the up-to-date guard.
    let lineage_json = format!(
        r#"{{
            "packages": {{
                "test/pull_refresh": {{
                    "commit": null,
                    "remote": {{
                        "bucket": "{bucket}",
                        "namespace": "test/pull_refresh",
                        "hash": "{install_hash}",
                        "catalog": "quilt.test"
                    }},
                    "base_hash": "{install_hash}",
                    "latest_hash": "{install_hash}",
                    "paths": {{}}
                }}
            }},
            "home": "{}"
        }}"#,
        home.as_ref().display(),
    );
    storage
        .write_byte_stream(&paths.lineage(), lineage_json.as_bytes().to_vec().into())
        .await?;

    // Installed manifest at the install-time hash so `package.manifest()`
    // can resolve.
    storage
        .write_byte_stream(
            paths.installed_manifest(&namespace, install_hash),
            ByteStream::from_static(br#"{"version": "v0"}"#),
        )
        .await?;

    // Remote `latest` tag has moved past the install — this is the
    // exact state that broke after the read-only-status refactor.
    remote
        .put_object(
            None,
            &S3Uri::try_from(
                format!("s3://{bucket}/.quilt/named_packages/test/pull_refresh/latest").as_str(),
            )?,
            new_hash.as_bytes().to_vec(),
        )
        .await?;

    let domain_lineage_io = DomainLineageIo::new(paths.lineage());
    let package = InstalledPackage {
        lineage: PackageLineageIo::new(domain_lineage_io, namespace.clone()),
        paths,
        remote: std::sync::Arc::new(remote),
        storage,
        namespace,
    };

    // Pull will eventually fail downstream — we have not staged the
    // manifest at `new_hash`, so `cache_remote_manifest` will return a
    // NotFound — but the *specific* error we are guarding against is
    // "package is already up-to-date". Any other failure mode proves
    // the refresh-then-check path ran.
    let err = package
        .pull(None, SyncScope::IndividualFiles)
        .await
        .expect_err("pull should fail downstream on the missing NEW_HASH manifest");
    let msg = err.to_string();
    assert!(
        !msg.contains("already up-to-date"),
        "pull must refresh latest_hash before the up-to-date guard; got: {msg}"
    );

    Ok(())
}

/// A `Behind` package: installed at `INSTALL_HASH` with one tracked `a.txt`,
/// while the remote `latest` tag has moved to `NEW_HASH`, whose manifest is
/// `latest_manifest`. No local changes.
async fn behind_package(
    latest_manifest: &str,
) -> Res<(
    InstalledPackage<LocalStorage, MockRemote>,
    (tempfile::TempDir, tempfile::TempDir),
)> {
    let (home, temp_dir1) = Home::from_temp_dir()?;
    let (paths, temp_dir2) = DomainPaths::from_temp_dir()?;
    let storage = LocalStorage::new();
    let remote = MockRemote::default();
    let namespace: Namespace = ("test", "pull_outcome").into();
    let bucket = "bkt";
    let install_hash = "INSTALL_HASH";
    let new_hash = "NEW_HASH";

    paths
        .scaffold_for_installing(&storage, &home, &namespace)
        .await?;
    paths.scaffold_for_caching(&storage, bucket).await?;

    // Disk lineage at the install-time hash: latest_hash == base_hash. The
    // remote `latest` tag (staged below) has moved, so `status`'s refresh
    // turns this into a `Behind` state.
    let lineage_json = format!(
        r#"{{
            "packages": {{
                "test/pull_outcome": {{
                    "commit": null,
                    "remote": {{
                        "bucket": "{bucket}",
                        "namespace": "test/pull_outcome",
                        "hash": "{install_hash}",
                        "catalog": "quilt.test"
                    }},
                    "base_hash": "{install_hash}",
                    "latest_hash": "{install_hash}",
                    "paths": {{}}
                }}
            }},
            "home": "{}"
        }}"#,
        home.as_ref().display(),
    );
    storage
        .write_byte_stream(&paths.lineage(), lineage_json.as_bytes().to_vec().into())
        .await?;

    // Installed (base) manifest with one tracked row.
    storage
        .write_byte_stream(
            paths.installed_manifest(&namespace, install_hash),
            ByteStream::from(BASE_MANIFEST.as_bytes().to_vec()),
        )
        .await?;

    // Remote `latest` tag has moved past the install → `Behind`.
    remote
        .put_object(
            None,
            &S3Uri::try_from(
                format!("s3://{bucket}/.quilt/named_packages/test/pull_outcome/latest").as_str(),
            )?,
            new_hash.as_bytes().to_vec(),
        )
        .await?;
    remote
        .put_object(
            None,
            &S3Uri::try_from(format!("s3://{bucket}/.quilt/packages/{new_hash}").as_str())?,
            latest_manifest.as_bytes().to_vec(),
        )
        .await?;

    let domain_lineage_io = DomainLineageIo::new(paths.lineage());
    let package = InstalledPackage {
        lineage: PackageLineageIo::new(domain_lineage_io, namespace.clone()),
        paths,
        remote: std::sync::Arc::new(remote),
        storage,
        namespace,
    };
    Ok((package, (temp_dir1, temp_dir2)))
}

/// What every early return of the dry run answers: nothing to pull, no paths,
/// and no newer revision to name.
fn empty_up_to_date() -> flow::PullPreview {
    flow::PullPreview {
        outcome: PullOutcome::UpToDate,
        added: vec![],
        changed: vec![],
        removed: vec![],
        latest_hash: None,
    }
}

/// The installed manifest [`behind_package`] starts from: one row, `a.txt`.
const BASE_MANIFEST: &str = concat!(
    "{\"version\":\"v0\",\"message\":\"\",\"user_meta\":null}\n",
    "{\"logical_key\":\"a.txt\",\"physical_keys\":[\"s3://bkt/a.txt\"],",
    "\"hash\":{\"type\":\"sha2-256-chunked\",",
    "\"value\":\"47DEQpj8HBSa+/TImW+5JCeuQeRkm5NMpJWZG3hSuFU=\"},",
    "\"size\":0,\"meta\":null}\n",
);

/// `pull_outcome` is the network-light dry-run the watcher/UI call before
/// routing a pull. A `Behind` package whose `latest` manifest is fetchable and
/// drops a tracked path (with no local changes) classifies as a clean surgical
/// update — never `UpToDate` — and names the dropped path and the newer
/// revision's hash.
#[test(tokio::test)]
async fn test_pull_outcome_behind_returns_non_up_to_date() -> Res {
    // The `latest` manifest drops `a.txt` (remote removal), so the delta is
    // non-empty and — with no local changes — resolves to a clean surgical
    // update.
    let (package, _temp_dirs) = behind_package("{\"version\":\"v0\"}").await?;

    let preview = package.pull_outcome(None).await?;
    assert!(
        matches!(
            preview.outcome,
            PullOutcome::CleanUpdate | PullOutcome::KeepsLocalChanges { .. }
        ),
        "expected a non-up-to-date outcome, got: {:?}",
        preview.outcome
    );
    assert_eq!(preview.added, [] as [PathBuf; 0]);
    assert_eq!(preview.changed, [] as [PathBuf; 0]);
    assert_eq!(preview.removed, vec![PathBuf::from("a.txt")]);
    assert_eq!(preview.latest_hash.as_deref(), Some("NEW_HASH"));

    Ok(())
}

/// The verdict-only dry run answers what the full one does, for the same
/// package, without the incoming paths.
#[test(tokio::test)]
async fn test_pull_verdict_matches_the_full_dry_run() -> Res {
    let (package, _temp_dirs) = behind_package("{\"version\":\"v0\"}").await?;

    let preview = package.pull_outcome(None).await?;
    assert_eq!(preview.removed, vec![PathBuf::from("a.txt")]);
    assert_eq!(package.pull_verdict(None).await?, preview.outcome);

    Ok(())
}

/// A newer revision that changed only the header still has a hash to link to,
/// though it brings no paths.
#[test(tokio::test)]
async fn test_pull_outcome_metadata_only_names_the_hash_and_no_paths() -> Res {
    let latest = BASE_MANIFEST.replacen(
        "\"message\":\"\"",
        "\"message\":\"newer revision message\"",
        1,
    );
    let (package, _temp_dirs) = behind_package(&latest).await?;

    assert_eq!(
        package.pull_outcome(None).await?,
        flow::PullPreview {
            outcome: PullOutcome::CleanUpdate,
            added: vec![],
            changed: vec![],
            removed: vec![],
            latest_hash: Some("NEW_HASH".to_string()),
        }
    );

    Ok(())
}

/// A local-only package (no `remote` in its lineage) has no `latest` tag to
/// resolve, so `pull_outcome` must short-circuit to `UpToDate` without touching
/// the network — never propagating the `NoRemote` error `remote()?` would raise.
#[test(tokio::test)]
async fn test_pull_outcome_local_no_remote_is_up_to_date() -> Res {
    let (home, _temp_dir1) = Home::from_temp_dir()?;
    let (paths, _temp_dir2) = DomainPaths::from_temp_dir()?;
    let storage = LocalStorage::new();
    let remote = MockRemote::default();
    let namespace: Namespace = ("test", "local_no_remote").into();

    paths
        .scaffold_for_installing(&storage, &home, &namespace)
        .await?;

    // No `remote` key → `remote_uri` is `None` → `UpstreamState::Local`.
    let lineage_json = format!(
        r#"{{
            "packages": {{
                "test/local_no_remote": {{
                    "commit": null,
                    "base_hash": "",
                    "latest_hash": "",
                    "paths": {{}}
                }}
            }},
            "home": "{}"
        }}"#,
        home.as_ref().display(),
    );
    storage
        .write_byte_stream(&paths.lineage(), lineage_json.as_bytes().to_vec().into())
        .await?;

    let domain_lineage_io = DomainLineageIo::new(paths.lineage());
    let package = InstalledPackage {
        lineage: PackageLineageIo::new(domain_lineage_io, namespace.clone()),
        paths,
        remote: std::sync::Arc::new(remote),
        storage,
        namespace,
    };

    assert_eq!(package.pull_outcome(None).await?, empty_up_to_date());

    Ok(())
}

/// A package diverged by hash (`remote.hash != base_hash` with a stale
/// `latest_hash` also differing from `base`) is `UpstreamState::Diverged` from
/// on-disk state alone — a purely lineage-local fact no tag read can cure.
/// `pull_outcome` must short-circuit to `UpToDate` WITHOUT any network, rather
/// than paying a manifest fetch + walk only to discard the result at the
/// post-walk `!= Behind` check.
#[test(tokio::test)]
async fn test_pull_outcome_diverged_by_hash_is_up_to_date_no_network() -> Res {
    let (home, _temp_dir1) = Home::from_temp_dir()?;
    let (paths, _temp_dir2) = DomainPaths::from_temp_dir()?;
    let storage = LocalStorage::new();
    let remote = MockRemote::default();
    let namespace: Namespace = ("test", "diverged_hash").into();
    let bucket = "bkt";
    let remote_hash = "R_HASH";
    let base_hash = "B_HASH";
    let latest_hash = "L_HASH";

    paths
        .scaffold_for_installing(&storage, &home, &namespace)
        .await?;

    // remote.hash != base_hash and latest_hash != base_hash, no commit →
    // `UpstreamState::from` reports `Diverged` (ahead && behind) from disk.
    let lineage_json = format!(
        r#"{{
            "packages": {{
                "test/diverged_hash": {{
                    "commit": null,
                    "remote": {{
                        "bucket": "{bucket}",
                        "namespace": "test/diverged_hash",
                        "hash": "{remote_hash}",
                        "catalog": "quilt.test"
                    }},
                    "base_hash": "{base_hash}",
                    "latest_hash": "{latest_hash}",
                    "paths": {{}}
                }}
            }},
            "home": "{}"
        }}"#,
        home.as_ref().display(),
    );
    storage
        .write_byte_stream(&paths.lineage(), lineage_json.as_bytes().to_vec().into())
        .await?;

    // Installed manifest at the current hash so that, WITHOUT the short-circuit,
    // `pull_outcome` would get past `manifest()` and pay for the tag read below.
    storage
        .write_byte_stream(
            paths.installed_manifest(&namespace, remote_hash),
            ByteStream::from_static(br#"{"version": "v0"}"#),
        )
        .await?;

    let domain_lineage_io = DomainLineageIo::new(paths.lineage());
    let package = InstalledPackage {
        lineage: PackageLineageIo::new(domain_lineage_io, namespace.clone()),
        paths,
        remote: std::sync::Arc::new(remote),
        storage,
        namespace,
    };

    assert_eq!(package.pull_outcome(None).await?, empty_up_to_date());

    // No `latest` tag was resolved — the divergence was decided from lineage
    // alone, with zero remote calls.
    let tag_uri = format!("s3://{bucket}/.quilt/named_packages/test/diverged_hash/latest");
    assert_eq!(package.remote.get_object_count(&tag_uri), 0);

    Ok(())
}

/// A package with a remote whose hash was never pushed (empty `hash` and empty
/// `latest_hash`) is `UpstreamState::Local`: there is no `latest` tag on the
/// bucket yet. `pull_outcome` must report `UpToDate` without a tag read, rather
/// than propagating the `NotFound` that resolving the absent tag would raise.
#[test(tokio::test)]
async fn test_pull_outcome_never_pushed_remote_is_up_to_date() -> Res {
    let (home, _temp_dir1) = Home::from_temp_dir()?;
    let (paths, _temp_dir2) = DomainPaths::from_temp_dir()?;
    let storage = LocalStorage::new();
    let remote = MockRemote::default();
    let namespace: Namespace = ("test", "never_pushed").into();
    let bucket = "bkt";

    paths
        .scaffold_for_installing(&storage, &home, &namespace)
        .await?;

    // Remote set but never pushed: empty `hash` + empty `latest_hash` →
    // `UpstreamState::Local`. No `latest` tag is staged on the mock remote, so
    // any tag read would 404.
    let lineage_json = format!(
        r#"{{
            "packages": {{
                "test/never_pushed": {{
                    "commit": null,
                    "remote": {{
                        "bucket": "{bucket}",
                        "namespace": "test/never_pushed",
                        "hash": "",
                        "catalog": "quilt.test"
                    }},
                    "base_hash": "",
                    "latest_hash": "",
                    "paths": {{}}
                }}
            }},
            "home": "{}"
        }}"#,
        home.as_ref().display(),
    );
    storage
        .write_byte_stream(&paths.lineage(), lineage_json.as_bytes().to_vec().into())
        .await?;

    let domain_lineage_io = DomainLineageIo::new(paths.lineage());
    let package = InstalledPackage {
        lineage: PackageLineageIo::new(domain_lineage_io, namespace.clone()),
        paths,
        remote: std::sync::Arc::new(remote),
        storage,
        namespace,
    };

    assert_eq!(package.pull_outcome(None).await?, empty_up_to_date());

    Ok(())
}

/// `latest_refreshed` says whether `status` read the remote's `latest` tip
/// this call, or fell back to the last-known one. A caller that shows the
/// status as fresh needs it: the fallback is silent, only a warning in the log.
#[test(tokio::test)]
async fn test_status_reports_whether_the_latest_tip_was_read() -> Res {
    async fn status_with(
        tag: Option<&str>,
        commit_json: &str,
        remote_json: &str,
        latest_hash: &str,
    ) -> Res<InstalledPackageStatus> {
        let (home, _temp_dir1) = Home::from_temp_dir()?;
        let (paths, _temp_dir2) = DomainPaths::from_temp_dir()?;
        let storage = LocalStorage::new();
        let remote = MockRemote::default();
        let namespace: Namespace = ("test", "refreshed").into();
        let hash = "abcdef";

        paths
            .scaffold_for_installing(&storage, &home, &namespace)
            .await?;
        let lineage_json = format!(
            r#"{{
                "packages": {{
                    "test/refreshed": {{
                        "commit": {commit_json},
                        "remote": {remote_json},
                        "base_hash": "{hash}",
                        "latest_hash": "{latest_hash}",
                        "paths": {{}}
                    }}
                }},
                "home": "{}"
            }}"#,
            home.as_ref().display(),
        );
        storage
            .write_byte_stream(&paths.lineage(), lineage_json.as_bytes().to_vec().into())
            .await?;
        storage
            .write_byte_stream(
                paths.installed_manifest(&namespace, hash),
                ByteStream::from_static(br#"{"version": "v0"}"#),
            )
            .await?;
        if let Some(tag) = tag {
            remote
                .put_object(
                    None,
                    &S3Uri::try_from("s3://bkt/.quilt/named_packages/test/refreshed/latest")?,
                    tag.as_bytes().to_vec(),
                )
                .await?;
        }

        let package = InstalledPackage {
            lineage: PackageLineageIo::new(
                DomainLineageIo::new(paths.lineage()),
                namespace.clone(),
            ),
            paths,
            remote: std::sync::Arc::new(remote),
            storage,
            namespace,
        };
        package.status(None).await
    }

    let remote_json = r#"{
        "bucket": "bkt",
        "namespace": "test/refreshed",
        "hash": "abcdef",
        "catalog": "quilt.test"
    }"#;

    let reached = status_with(Some("abcdef"), "null", remote_json, "abcdef").await?;
    assert!(reached.latest_refreshed, "the tip was read this call");
    assert_eq!(reached.upstream_state, UpstreamState::UpToDate);

    // No `latest` tag to read: the refresh fails, is swallowed, and the
    // verdict comes from the last-known tip.
    let unreachable = status_with(None, "null", remote_json, "abcdef").await?;
    assert!(!unreachable.latest_refreshed, "the tip was not read");
    assert_eq!(unreachable.upstream_state, UpstreamState::UpToDate);

    // No `latest` tag, and none ever seen: the remote was read and says the
    // package was never published, which is a fresh answer, not an outage.
    let committed_json = r#"{
        "timestamp": "2024-01-01T00:00:00Z",
        "hash": "abcdef",
        "prev_hashes": []
    }"#;
    let unpublished_json = r#"{
        "bucket": "bkt",
        "namespace": "test/refreshed",
        "hash": "",
        "catalog": "quilt.test"
    }"#;
    let unpublished = status_with(None, committed_json, unpublished_json, "").await?;
    assert!(unpublished.latest_refreshed, "the missing tag was read");
    assert_eq!(unpublished.upstream_state, UpstreamState::Local);

    let local = status_with(None, "null", "null", "abcdef").await?;
    assert!(!local.latest_refreshed, "no remote, nothing to read");

    Ok(())
}

/// The package's state, and how many times its `latest` tag was read.
async fn state_with(
    tag: Option<&str>,
    remote_json: &str,
) -> Res<(crate::lineage::PackageStateReport, usize)> {
    let (home, _temp_dir1) = Home::from_temp_dir()?;
    let (paths, _temp_dir2) = DomainPaths::from_temp_dir()?;
    let storage = LocalStorage::new();
    let remote = MockRemote::default();
    let namespace: Namespace = ("test", "state").into();
    let hash = "abcdef";
    let tag_uri = "s3://bkt/.quilt/named_packages/test/state/latest";

    paths
        .scaffold_for_installing(&storage, &home, &namespace)
        .await?;
    let lineage_json = format!(
        r#"{{
            "packages": {{
                "test/state": {{
                    "commit": null,
                    "remote": {remote_json},
                    "base_hash": "{hash}",
                    "latest_hash": "{hash}",
                    "paths": {{}}
                }}
            }},
            "home": "{}"
        }}"#,
        home.as_ref().display(),
    );
    storage
        .write_byte_stream(&paths.lineage(), lineage_json.as_bytes().to_vec().into())
        .await?;
    storage
        .write_byte_stream(
            paths.installed_manifest(&namespace, hash),
            ByteStream::from_static(br#"{"version": "v0"}"#),
        )
        .await?;
    if let Some(tag) = tag {
        remote
            .put_object(None, &S3Uri::try_from(tag_uri)?, tag.as_bytes().to_vec())
            .await?;
    }

    let package = InstalledPackage {
        lineage: PackageLineageIo::new(DomainLineageIo::new(paths.lineage()), namespace.clone()),
        paths,
        remote: std::sync::Arc::new(remote),
        storage,
        namespace,
    };
    let report = package.state().await?;
    Ok((report, package.remote.get_object_count(tag_uri)))
}

/// `state` asks the remote only when there is something to ask, and resolves
/// what it learns through `PackageState::resolve`.
#[test(tokio::test)]
async fn test_state_short_circuits_and_resolves_the_status() -> Res {
    use crate::lineage::PackageState;
    use crate::lineage::PackageStateReport;

    let (latest, reads) = state_with(
        Some("abcdef"),
        r#"{
            "bucket": "bkt",
            "namespace": "test/state",
            "hash": "abcdef",
            "origin": "quilt.test"
        }"#,
    )
    .await?;
    assert_eq!(
        latest,
        PackageStateReport {
            state: PackageState::Latest,
            upstream_state: UpstreamState::UpToDate,
            changed_files: Some(0),
            latest_refreshed: true,
        }
    );
    assert_eq!(reads, 1, "the tip was read");

    // No catalog host: nowhere to vend credentials from, so nothing is asked.
    let (misconfigured, reads) = state_with(
        Some("abcdef"),
        r#"{
            "bucket": "bkt",
            "namespace": "test/state",
            "hash": "abcdef"
        }"#,
    )
    .await?;
    assert_eq!(
        misconfigured,
        PackageStateReport {
            state: PackageState::Unknown,
            upstream_state: UpstreamState::Error,
            changed_files: None,
            latest_refreshed: false,
        }
    );
    assert_eq!(reads, 0, "a misconfigured remote is not asked");

    // No remote: resolved from lineage, the working tree not walked.
    let (local, _) = state_with(None, "null").await?;
    assert_eq!(
        local,
        PackageStateReport {
            state: PackageState::NoRemote,
            upstream_state: UpstreamState::Local,
            changed_files: None,
            latest_refreshed: false,
        }
    );

    Ok(())
}