Skip to content

test(gmail): close #1503's patch-coverage gaps in sync's live-progress and reconciliation paths #1506

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
newhoggy wants to merge 1 commit into
base: main
Choose a base branch
from
Open

test(gmail): close #1503's patch-coverage gaps in sync's live-progress and reconciliation paths #1506

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
76 changes: 75 additions & 1 deletion src/cli/gmail/sync.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,7 @@ impl SyncCommand {
},
self.quiet,
&self.output,
std::io::stderr().is_terminal(),
)
.await
}
Expand All @@ -120,13 +121,19 @@ impl SyncCommand {
/// render it, and only *then* decide the process exit condition — a
/// non-empty `errors` becomes a failing exit code after everything already
/// ran and printed, never silently.
///
/// Takes `stderr_is_terminal` as a value for the same reason
/// [`should_show_progress`] does — so tests can force the live-progress
/// branch without depending on the test runner's own (never-a-terminal)
/// stderr.
async fn run_sync_command(
client: &GmailClient,
opts: SyncOptions,
quiet: bool,
output: &OutputFormat,
stderr_is_terminal: bool,
) -> Result<()> {
let show_progress = should_show_progress(quiet, output, std::io::stderr().is_terminal());
let show_progress = should_show_progress(quiet, output, stderr_is_terminal);
let report = if show_progress {
let (tx, rx) = mpsc::unbounded_channel();
let bars = SyncProgressBars::new();
Expand Down Expand Up @@ -548,6 +555,7 @@ mod tests {
},
false,
&OutputFormat::Table,
false,
)
.await
.unwrap();
Expand Down Expand Up @@ -594,12 +602,76 @@ mod tests {
},
false,
&OutputFormat::Table,
false,
)
.await
.unwrap_err();
assert!(err.to_string().contains("1 message(s) failed"));
}

#[tokio::test]
async fn run_sync_command_with_stderr_is_terminal_takes_the_live_progress_path() {
// `stderr_is_terminal: true` is what `should_show_progress` needs to
// pick the live-progress branch — the test runner's own stderr is
// never a terminal, hence injecting it here rather than relying on
// `std::io::IsTerminal` (see `run_sync_command`'s doc comment).
let server = wiremock::MockServer::start().await;
let client = client_with_bootstrapped_token(&server).await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/gmail/v1/users/me/profile"))
.respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
"emailAddress": "user@example.com", "messagesTotal": 1, "threadsTotal": 1, "historyId": "1"
})))
.mount(&server)
.await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/gmail/v1/users/me/messages"))
.respond_with(
wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
"messages": [{"id": "m1", "threadId": "t1"}]
})),
)
.mount(&server)
.await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/gmail/v1/users/me/messages/m1"))
.and(wiremock::matchers::query_param("format", "raw"))
.respond_with(
wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
"id": "m1", "threadId": "t1", "labelIds": ["INBOX"],
"internalDate": "1700000000000", "historyId": "500",
"raw": base64::engine::general_purpose::URL_SAFE_NO_PAD
.encode("From: a@example.com\r\n\r\nBody of m1."),
})),
)
.mount(&server)
.await;

let dir = tempfile::tempdir().unwrap();
let output_dir = dir.path().join("archive");
run_sync_command(
&client,
SyncOptions {
output_dir: output_dir.clone(),
query: None,
full: false,
concurrency: 4,
dry_run: false,
extract_attachments: false,
},
false,
&OutputFormat::Table,
true,
)
.await
.unwrap();

assert!(
output_dir.exists(),
"the live-progress run must still archive the message"
);
}

#[tokio::test]
async fn execute_passes_flags_through() {
let server = wiremock::MockServer::start().await;
Expand Down Expand Up @@ -722,6 +794,7 @@ Content-Disposition: attachment; filename=\"report.pdf\"\r\n\
},
false,
&OutputFormat::Table,
false,
)
.await
.unwrap();
Expand Down Expand Up @@ -750,6 +823,7 @@ Content-Disposition: attachment; filename=\"report.pdf\"\r\n\
},
false,
&OutputFormat::Table,
false,
)
.await
.unwrap();
Expand Down
132 changes: 132 additions & 0 deletions src/cli/gmail/sync/engine.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1588,6 +1588,81 @@ not-really-a-pdf\r\n\
assert!(err.to_string().contains("Failed to parse manifest"));
}

// ── `--full` forces reconciliation even with a valid watermark ──────

#[tokio::test]
async fn run_sync_full_flag_forces_reconciliation_even_with_a_valid_watermark() {
let server = wiremock::MockServer::start().await;
let client = client_with_bootstrapped_token(&server).await;
let dir = tempfile::tempdir().unwrap();
let output_dir = dir.path().join("archive");
std::fs::create_dir_all(&output_dir).unwrap();
state::save(
&ArchiveState {
history_id: "100".to_string(),
email_address: "user@example.com".to_string(),
last_sync: Utc::now(),
query: None,
},
&state_path(&output_dir),
)
.unwrap();

mount_profile(&server, "user@example.com", "999").await;
mount_message_list(&server, &["m1"]).await;
mount_raw_get(&server, "m1", "Hello").await;
// Deliberately no mock for GET .../history — `--full` must take the
// full-listing path even though a valid watermark exists, never
// fall through to `run_incremental`.

let mut o = opts(output_dir.clone());
o.full = true;
let report = run_sync(&client, &o).await.unwrap();

assert!(report.errors.is_empty());
assert!(report
.actions
.iter()
.any(|a| matches!(a, SyncAction::Fetched { id, .. } if id == "m1")));
match state::load(&state_path(&output_dir)) {
LoadOutcome::Present(s) => assert_eq!(
s.history_id, "999",
"the reconciliation's historyId, not the pre-existing watermark, must be persisted"
),
_ => panic!("expected state.json to still be present after --full"),
}
}

// ── a corrupt state.json is a reconciliation trigger, not a hard error ──

#[tokio::test]
async fn run_sync_reconciles_when_state_json_is_corrupt() {
let server = wiremock::MockServer::start().await;
let client = client_with_bootstrapped_token(&server).await;
let dir = tempfile::tempdir().unwrap();
let output_dir = dir.path().join("archive");
std::fs::create_dir_all(&output_dir).unwrap();
std::fs::write(state_path(&output_dir), "not json\n").unwrap();

mount_profile(&server, "user@example.com", "999").await;
mount_message_list(&server, &["m1"]).await;
mount_raw_get(&server, "m1", "Hello").await;

let report = run_sync(&client, &opts(output_dir.clone())).await.unwrap();

assert!(report.errors.is_empty());
assert!(
report.actions.iter().any(
|a| matches!(a, SyncAction::Note { message } if message.contains("state.json unreadable"))
),
"a corrupt watermark should be reported as a reconciliation note"
);
match state::load(&state_path(&output_dir)) {
LoadOutcome::Present(s) => assert_eq!(s.history_id, "999"),
_ => panic!("expected a fresh state.json to replace the corrupt one"),
}
}

// ── reconciliation must not truncate at HARD_CAP (#1467) ────────────

#[tokio::test]
Expand Down Expand Up @@ -1984,6 +2059,63 @@ not-really-a-pdf\r\n\
assert_eq!(completed_failed, 1);
}

#[tokio::test]
async fn fetch_and_archive_messages_streaming_dedupes_a_repeated_id() {
// A page-boundary race in the live-streamed listing can re-observe
// the same id twice — the second occurrence must be a no-op, not a
// second fetch, and `listed_ids` must still only contain it once.
let server = wiremock::MockServer::start().await;
let client = client_with_bootstrapped_token(&server).await;
let dir = tempfile::tempdir().unwrap();
let output_dir = dir.path().join("archive");
std::fs::create_dir_all(&output_dir).unwrap();

wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/gmail/v1/users/me/messages/m1"))
.and(wiremock::matchers::query_param("format", "raw"))
.respond_with(
wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
"id": "m1", "threadId": "t1", "labelIds": ["INBOX"],
"internalDate": "1700000000000", "historyId": "500",
"raw": raw_message_body("m1", "Hello"),
})),
)
.expect(1)
.mount(&server)
.await;

let mut manifest = Manifest::default();
let limiter = TokenBucket::new(1_000_000, 1_000_000);
let mut report = SyncReport::default();
let o = opts(output_dir.clone());

let (ids_tx, ids_rx) = mpsc::unbounded_channel();
for id in ["m1", "m1"] {
ids_tx.send(id.to_string()).unwrap();
}
drop(ids_tx);

let listed_ids = fetch_and_archive_messages_streaming(
&client,
&mut manifest,
ids_rx,
&limiter,
&o,
&mut report,
None,
)
.await
.unwrap();

assert_eq!(listed_ids.len(), 1, "listed_ids must dedupe too");
assert_eq!(
report.actions.len(),
1,
"the duplicate id must not be fetched a second time"
);
assert!(matches!(&report.actions[0], SyncAction::Fetched { id, .. } if id == "m1"));
}

#[tokio::test]
async fn run_sync_with_progress_emits_listing_events_ending_in_listing_done() {
let server = wiremock::MockServer::start().await;
Expand Down
Loading