solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4from __future__ import annotations
5
6import json
7import sys
8from pathlib import Path
9from typing import Any
10from unittest.mock import Mock
11
12import pytest
13from typer.testing import CliRunner
14
15from solstone.think import backup_cli, sol_cli
16from solstone.think.backup.destination import Destination, DestinationStatus
17from solstone.think.backup.engine import BackupResult, PruneResult
18from solstone.think.backup.hosted import hosted_binding_path, load_hosted_binding
19from solstone.think.backup.keys import format_recovery_key_display
20from solstone.think.backup.restore import RestoreResult
21from solstone.think.backup.rotation import RotationResult
22from solstone.think.backup.state import BackupKeys
23from solstone.think.backup.teardown import TeardownResult
24from solstone.think.offload import OffloadResult, OffloadSegmentDetail
25from solstone.think.offload_restore import OffloadRestoreResult
26
27
28def _config_path(journal: Path) -> Path:
29 return journal / "config" / "journal.json"
30
31
32def _write_config(journal: Path, payload: dict[str, Any]) -> None:
33 config_path = _config_path(journal)
34 config_path.parent.mkdir(parents=True, exist_ok=True)
35 config_path.write_text(json.dumps(payload), encoding="utf-8")
36
37
38def _destination() -> Destination:
39 return Destination(
40 repository="s3:safe-bucket/path",
41 backend="s3",
42 credentials={
43 "access_key_id": "access-key",
44 "secret_access_key": "secret-key",
45 },
46 )
47
48
49def _destination_config() -> dict[str, Any]:
50 destination = _destination()
51 return {
52 "repository": destination.repository,
53 "backend": destination.backend,
54 "credentials": destination.credentials,
55 }
56
57
58def _status(reason_code: str) -> DestinationStatus:
59 if reason_code == "repo_missing":
60 return DestinationStatus(
61 reachable=True,
62 repo_exists=False,
63 reason_code="repo_missing",
64 message="backup destination is reachable and needs setup",
65 )
66 if reason_code == "auth_failed":
67 return DestinationStatus(
68 reachable=True,
69 repo_exists=True,
70 reason_code="auth_failed",
71 message="repository password was rejected",
72 )
73 return DestinationStatus(
74 reachable=True,
75 repo_exists=True,
76 reason_code="repo_exists",
77 message="backup repository is reachable",
78 )
79
80
81def test_registry_and_command_tree(monkeypatch: pytest.MonkeyPatch) -> None:
82 assert sol_cli.COMMANDS["backup"].module == "solstone.think.backup_cli"
83 assert sol_cli.COMMANDS["backup"].surface == "service"
84
85 captured: dict[str, object] = {}
86
87 def fake_run_command(module_path: str) -> int:
88 captured["module"] = module_path
89 captured["argv"] = list(sys.argv)
90 return 0
91
92 monkeypatch.setattr(sol_cli, "run_command", fake_run_command)
93 monkeypatch.setattr(sol_cli.setproctitle, "setproctitle", lambda _title: None)
94 monkeypatch.setattr(sys, "argv", ["journal", "backup", "status"])
95
96 with pytest.raises(SystemExit) as exc_info:
97 sol_cli.journal_main()
98
99 assert exc_info.value.code == 0
100 assert captured == {
101 "module": "solstone.think.backup_cli",
102 "argv": ["journal backup", "status"],
103 }
104
105 runner = CliRunner()
106 root_help = runner.invoke(backup_cli.app, ["--help"])
107 assert root_help.exit_code == 0
108 for command in (
109 "enable",
110 "destination",
111 "run",
112 "prune",
113 "status",
114 "recovery-key",
115 "offload",
116 "restore",
117 "off",
118 ):
119 assert command in root_help.output
120
121 destination_help = runner.invoke(backup_cli.app, ["destination", "--help"])
122 assert destination_help.exit_code == 0
123 assert "set" in destination_help.output
124 assert "show" in destination_help.output
125
126 recovery_help = runner.invoke(backup_cli.app, ["recovery-key", "--help"])
127 assert recovery_help.exit_code == 0
128 assert "show" in recovery_help.output
129 assert "rotate" in recovery_help.output
130
131 offload_help = runner.invoke(backup_cli.app, ["offload", "--help"])
132 assert offload_help.exit_code == 0
133 assert "status" in offload_help.output
134 assert "run" in offload_help.output
135 assert "restore" in offload_help.output
136
137
138@pytest.mark.parametrize(
139 ("reason_code", "expected_exit"),
140 [
141 ("repo_missing", 0),
142 ("auth_failed", 1),
143 ],
144)
145def test_destination_set_reads_stdin_and_keeps_secrets_off_argv(
146 tmp_path: Path,
147 monkeypatch: pytest.MonkeyPatch,
148 reason_code: str,
149 expected_exit: int,
150) -> None:
151 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
152 _write_config(tmp_path, {})
153 captured: dict[str, Any] = {}
154
155 def fake_set_destination(destination: Destination) -> None:
156 captured["destination"] = destination
157
158 def fake_validate_destination(
159 destination: Destination,
160 password: str,
161 *,
162 restic_path: Path,
163 ) -> DestinationStatus:
164 captured["argv"] = list(sys.argv)
165 captured["password"] = password
166 captured["restic_path"] = restic_path
167 return _status(reason_code)
168
169 monkeypatch.setattr(backup_cli, "set_destination", fake_set_destination)
170 monkeypatch.setattr(backup_cli, "ensure_restic", lambda: Path("/restic"))
171 monkeypatch.setattr(backup_cli, "generate_daily_key", lambda: "probe-secret")
172 monkeypatch.setattr(backup_cli, "validate_destination", fake_validate_destination)
173
174 payload = {
175 "repository": "s3:safe-bucket/path",
176 "backend": "s3",
177 "credentials": {
178 "access_key_id": "AKIASECRET",
179 "secret_access_key": "TOPSECRET",
180 },
181 }
182 result = CliRunner().invoke(
183 backup_cli.app,
184 ["destination", "set"],
185 input=json.dumps(payload),
186 )
187
188 assert result.exit_code == expected_exit
189 destination = captured["destination"]
190 assert destination.credentials == payload["credentials"]
191 assert captured["password"] == "probe-secret"
192 assert captured["restic_path"] == Path("/restic")
193 assert "TOPSECRET" not in " ".join(captured["argv"])
194 assert "AKIASECRET" not in " ".join(captured["argv"])
195 assert "TOPSECRET" not in result.output
196 assert "AKIASECRET" not in result.output
197
198
199def test_destination_set_hosted_writes_binding_0600_without_leaking_token(
200 tmp_path: Path,
201 monkeypatch: pytest.MonkeyPatch,
202) -> None:
203 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
204 payload = {
205 "broker_endpoint": "https://broker.example",
206 "account_id": "acct",
207 "instance_id": "inst",
208 "bucket": "bkt",
209 "prefix": "users/acct/inst/",
210 "broker_token": "BTOKEN",
211 }
212
213 result = CliRunner().invoke(
214 backup_cli.app,
215 ["destination", "set-hosted"],
216 input=json.dumps(payload),
217 )
218
219 assert result.exit_code == 0
220 assert "BTOKEN" not in result.stdout
221 binding = load_hosted_binding()
222 assert binding is not None
223 assert binding.prefix == "users/acct/inst/"
224 assert hosted_binding_path().stat().st_mode & 0o777 == 0o600
225
226
227def test_destination_set_hosted_rejects_missing_field(
228 tmp_path: Path,
229 monkeypatch: pytest.MonkeyPatch,
230) -> None:
231 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
232 payload = {
233 "broker_endpoint": "https://broker.example",
234 "account_id": "acct",
235 "instance_id": "inst",
236 "bucket": "bkt",
237 "prefix": "users/acct/inst/",
238 }
239
240 result = CliRunner().invoke(
241 backup_cli.app,
242 ["destination", "set-hosted"],
243 input=json.dumps(payload),
244 )
245
246 assert result.exit_code != 0
247
248
249def test_status_and_destination_show_are_redacted(
250 tmp_path: Path,
251 monkeypatch: pytest.MonkeyPatch,
252) -> None:
253 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
254 recovery_key = "A" * 64
255 _write_config(
256 tmp_path,
257 {
258 "backup": {
259 "destination": {
260 "repository": "s3:safe-bucket/path",
261 "backend": "s3",
262 "credentials": {
263 "access_key_id": "ACCESSSECRET",
264 "secret_access_key": "BACKENDSECRET",
265 },
266 },
267 "daily_key": "DAILYSECRET",
268 "recovery_key": recovery_key,
269 }
270 },
271 )
272 runner = CliRunner()
273
274 status_result = runner.invoke(backup_cli.app, ["status"])
275 destination_result = runner.invoke(backup_cli.app, ["destination", "show"])
276
277 assert status_result.exit_code == 0
278 assert destination_result.exit_code == 0
279 output = status_result.output + destination_result.output
280 for secret in ("DAILYSECRET", recovery_key, "ACCESSSECRET", "BACKENDSECRET"):
281 assert secret not in output
282 assert "credentials_set" in output
283
284
285def test_enable_accepts_lookalike_recovery_key_confirmation(
286 tmp_path: Path,
287 monkeypatch: pytest.MonkeyPatch,
288) -> None:
289 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
290 _write_config(tmp_path, {"backup": {"destination": _destination_config()}})
291 recovery_key = ("0" * 32) + ("1" * 32)
292 display = format_recovery_key_display(recovery_key)
293 keys = BackupKeys(
294 daily_key="daily-secret",
295 recovery_key=recovery_key,
296 recovery_key_display=display,
297 )
298 calls: dict[str, Any] = {}
299
300 def fake_set_recovery_key_confirmed(confirmed: bool = True) -> None:
301 calls["confirmed"] = confirmed
302
303 def fake_set_enabled(enabled: bool) -> None:
304 calls["enabled"] = enabled
305
306 def fake_init_repository(
307 destination: Destination,
308 *,
309 daily_key: str,
310 recovery_key: str,
311 restic_path: Path,
312 ) -> None:
313 calls["init"] = {
314 "destination": destination,
315 "daily_key": daily_key,
316 "recovery_key": recovery_key,
317 "restic_path": restic_path,
318 }
319
320 monkeypatch.setattr(backup_cli, "generate_and_store_keys", lambda: keys)
321 monkeypatch.setattr(backup_cli, "ensure_restic", lambda: Path("/restic"))
322 monkeypatch.setattr(
323 backup_cli,
324 "set_recovery_key_confirmed",
325 fake_set_recovery_key_confirmed,
326 )
327 monkeypatch.setattr(backup_cli, "set_enabled", fake_set_enabled)
328 monkeypatch.setattr(backup_cli, "init_repository", fake_init_repository)
329
330 entered = display.replace("0", "O").replace("1", "I")
331 result = CliRunner().invoke(backup_cli.app, ["enable"], input=entered)
332
333 assert result.exit_code == 0
334 assert calls["confirmed"] is True
335 assert calls["enabled"] is True
336 assert calls["init"]["daily_key"] == "daily-secret"
337 assert calls["init"]["recovery_key"] == recovery_key
338
339
340def test_off_requires_yes_before_teardown(monkeypatch: pytest.MonkeyPatch) -> None:
341 teardown_backup = Mock(return_value=TeardownResult(status="ok", reason_code=None))
342 monkeypatch.setattr(backup_cli, "teardown_backup", teardown_backup)
343 runner = CliRunner()
344
345 refused = runner.invoke(backup_cli.app, ["off"])
346 accepted = runner.invoke(backup_cli.app, ["off", "--yes"])
347
348 assert refused.exit_code == 1
349 assert "Refusing" in refused.output
350 teardown_backup.assert_called_once_with()
351 assert accepted.exit_code == 0
352 assert "Backup turned off." in accepted.output
353
354
355def test_enable_power_user_skips_ceremony(
356 tmp_path: Path,
357 monkeypatch: pytest.MonkeyPatch,
358) -> None:
359 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
360 _write_config(
361 tmp_path,
362 {
363 "backup": {
364 "destination": _destination_config(),
365 "daily_key": "manual-daily",
366 "recovery_key": None,
367 }
368 },
369 )
370 set_enabled = Mock()
371 monkeypatch.setattr(
372 backup_cli,
373 "generate_and_store_keys",
374 lambda: pytest.fail("ceremony should be skipped"),
375 )
376 monkeypatch.setattr(
377 backup_cli,
378 "set_recovery_key_confirmed",
379 lambda *_args, **_kwargs: pytest.fail("confirmation should not be set"),
380 )
381 monkeypatch.setattr(backup_cli, "ensure_restic", lambda: Path("/restic"))
382 monkeypatch.setattr(
383 backup_cli, "validate_destination", lambda *a, **k: _status("ok")
384 )
385 monkeypatch.setattr(backup_cli, "set_enabled", set_enabled)
386
387 result = CliRunner().invoke(backup_cli.app, ["enable"])
388
389 assert result.exit_code == 0
390 assert "Your recovery key" not in result.output
391 set_enabled.assert_called_once_with(True)
392
393
394def test_enable_power_user_requires_existing_repository(
395 tmp_path: Path,
396 monkeypatch: pytest.MonkeyPatch,
397) -> None:
398 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
399 _write_config(
400 tmp_path,
401 {
402 "backup": {
403 "destination": _destination_config(),
404 "daily_key": "manual-daily",
405 "recovery_key": None,
406 }
407 },
408 )
409 set_enabled = Mock()
410 monkeypatch.setattr(backup_cli, "ensure_restic", lambda: Path("/restic"))
411 monkeypatch.setattr(
412 backup_cli,
413 "validate_destination",
414 lambda *a, **k: _status("repo_missing"),
415 )
416 monkeypatch.setattr(backup_cli, "set_enabled", set_enabled)
417
418 result = CliRunner().invoke(backup_cli.app, ["enable"])
419
420 assert result.exit_code == 1
421 assert "Repository not found" in result.output
422 set_enabled.assert_not_called()
423
424
425@pytest.mark.parametrize(
426 ("result", "expected_exit", "expected_text"),
427 [
428 (BackupResult("ok", "snap123", None), 0, "snap123"),
429 (BackupResult("skipped", None, None), 0, "skipped"),
430 (BackupResult("error", None, "auth_failed"), 1, "auth_failed"),
431 ],
432)
433def test_run_maps_engine_status(
434 monkeypatch: pytest.MonkeyPatch,
435 result: BackupResult,
436 expected_exit: int,
437 expected_text: str,
438) -> None:
439 monkeypatch.setattr(backup_cli, "run_backup", lambda: result)
440
441 invoke_result = CliRunner().invoke(backup_cli.app, ["run"])
442
443 assert invoke_result.exit_code == expected_exit
444 assert expected_text in invoke_result.output
445
446
447@pytest.mark.parametrize(
448 ("result", "expected_exit", "expected_text"),
449 [
450 (PruneResult("ok", None), 0, "Retention prune complete."),
451 (PruneResult("skipped", None), 0, "skipped"),
452 (PruneResult("error", "auth_failed"), 1, "auth_failed"),
453 ],
454)
455def test_prune_maps_engine_status(
456 monkeypatch: pytest.MonkeyPatch,
457 result: PruneResult,
458 expected_exit: int,
459 expected_text: str,
460) -> None:
461 monkeypatch.setattr(backup_cli, "run_prune", lambda: result)
462
463 invoke_result = CliRunner().invoke(backup_cli.app, ["prune"])
464
465 assert invoke_result.exit_code == expected_exit
466 assert expected_text in invoke_result.output
467
468
469def test_offload_status_json_delegates_to_status_builder(
470 monkeypatch: pytest.MonkeyPatch,
471) -> None:
472 payload = {
473 "offload": {"enabled": True, "budget_bytes": 100, "floor_bytes": 50},
474 "raw_media": {"total_bytes": 30, "total_files": 3},
475 "backup_only": {"total_bytes": 20, "degraded": False},
476 }
477 build_offload_status = Mock(return_value=payload)
478 monkeypatch.setattr(backup_cli, "build_offload_status", build_offload_status)
479
480 result = CliRunner().invoke(backup_cli.app, ["offload", "status", "--json"])
481
482 assert result.exit_code == 0
483 assert json.loads(result.output) == payload
484 build_offload_status.assert_called_once_with()
485
486
487def test_offload_run_delegates_to_existing_pass_and_formats_like_maintenance(
488 monkeypatch: pytest.MonkeyPatch,
489) -> None:
490 captured: dict[str, Any] = {}
491
492 def fake_run_offload(*, dry_run: bool) -> OffloadResult:
493 captured["dry_run"] = dry_run
494 return OffloadResult(
495 status="ok",
496 reason=None,
497 files_offloaded=0,
498 bytes_offloaded=0,
499 ran_out_of_media=False,
500 dry_run=True,
501 details=(
502 OffloadSegmentDetail(
503 day="20260101",
504 stream="_default",
505 segment="090000_300",
506 files=2,
507 bytes=50,
508 ),
509 ),
510 )
511
512 monkeypatch.setattr(backup_cli, "run_offload", fake_run_offload)
513
514 result = CliRunner().invoke(backup_cli.app, ["offload", "run", "--dry-run"])
515
516 assert result.exit_code == 0
517 assert captured == {"dry_run": True}
518 assert (
519 "backup offload: ok dry_run=true selected_files=2 selected_bytes=50 "
520 "ran_out_of_media=False segments=20260101/_default/090000_300:50"
521 in result.output
522 )
523
524
525def test_offload_restore_day_json_delegates_to_restore_engine(
526 monkeypatch: pytest.MonkeyPatch,
527) -> None:
528 restore_day = Mock(
529 return_value=OffloadRestoreResult(
530 status="ok",
531 reason=None,
532 scope="day",
533 day="20260228",
534 segments_selected=1,
535 segments_restored=1,
536 files_expected=2,
537 files_restored=2,
538 bytes_expected=50,
539 bytes_restored=50,
540 details=(),
541 )
542 )
543 monkeypatch.setattr(backup_cli, "restore_day", restore_day)
544
545 result = CliRunner().invoke(
546 backup_cli.app,
547 ["offload", "restore", "20260228", "--json"],
548 )
549
550 assert result.exit_code == 0
551 assert json.loads(result.output)["day"] == "20260228"
552 restore_day.assert_called_once_with("20260228")
553
554
555def test_offload_restore_all_and_invalid_scope(
556 monkeypatch: pytest.MonkeyPatch,
557) -> None:
558 restore_all = Mock(
559 return_value=OffloadRestoreResult(
560 status="no_op",
561 reason="nothing_to_restore",
562 scope="all",
563 day=None,
564 segments_selected=0,
565 segments_restored=0,
566 files_expected=0,
567 files_restored=0,
568 bytes_expected=0,
569 bytes_restored=0,
570 details=(),
571 )
572 )
573 monkeypatch.setattr(backup_cli, "restore_all", restore_all)
574
575 all_result = CliRunner().invoke(backup_cli.app, ["offload", "restore", "--all"])
576 mixed_result = CliRunner().invoke(
577 backup_cli.app,
578 ["offload", "restore", "20260228", "--all"],
579 )
580 missing_result = CliRunner().invoke(backup_cli.app, ["offload", "restore"])
581
582 assert all_result.exit_code == 0
583 assert "status=no_op reason=nothing_to_restore" in all_result.output
584 restore_all.assert_called_once_with()
585 assert mixed_result.exit_code == 1
586 assert "Use either a day or --all" in mixed_result.output
587 assert missing_result.exit_code == 1
588 assert "Provide a day or --all" in missing_result.output
589
590
591def test_recovery_key_show_prints_display_only(
592 tmp_path: Path,
593 monkeypatch: pytest.MonkeyPatch,
594) -> None:
595 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
596 recovery_key = "A" * 64
597 display = format_recovery_key_display(recovery_key)
598 _write_config(
599 tmp_path,
600 {
601 "backup": {
602 "daily_key": "daily-secret",
603 "recovery_key": recovery_key,
604 }
605 },
606 )
607
608 result = CliRunner().invoke(backup_cli.app, ["recovery-key", "show"])
609
610 assert result.exit_code == 0
611 assert "AAAA AAAA AAAA AAAA" in result.output
612 assert recovery_key not in result.output
613 assert display.replace(" ", "") not in result.output
614
615
616def test_recovery_key_show_errors_without_key(
617 tmp_path: Path,
618 monkeypatch: pytest.MonkeyPatch,
619) -> None:
620 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
621 _write_config(tmp_path, {})
622
623 result = CliRunner().invoke(backup_cli.app, ["recovery-key", "show"])
624
625 assert result.exit_code == 1
626 assert "No recovery key is set." in result.output
627
628
629def test_recovery_key_rotate_prints_display_not_canonical(
630 monkeypatch: pytest.MonkeyPatch,
631) -> None:
632 canonical = "B" * 64
633 display = format_recovery_key_display(canonical)
634 monkeypatch.setattr(
635 backup_cli,
636 "rotate_recovery_key",
637 lambda: RotationResult(
638 status="ok",
639 reason_code=None,
640 recovery_key=canonical,
641 recovery_key_display=display,
642 ),
643 )
644
645 result = CliRunner().invoke(backup_cli.app, ["recovery-key", "rotate"])
646
647 assert result.exit_code == 0
648 assert "BBBB BBBB BBBB BBBB" in result.output
649 assert canonical not in result.output
650
651
652def test_recovery_key_rotate_errors(monkeypatch: pytest.MonkeyPatch) -> None:
653 monkeypatch.setattr(
654 backup_cli,
655 "rotate_recovery_key",
656 lambda: RotationResult(
657 status="error",
658 reason_code="auth_failed",
659 recovery_key=None,
660 recovery_key_display=None,
661 ),
662 )
663
664 result = CliRunner().invoke(backup_cli.app, ["recovery-key", "rotate"])
665
666 assert result.exit_code == 1
667 assert "auth_failed" in result.output
668
669
670def test_restore_reads_secret_from_stdin_and_reports_result(
671 monkeypatch: pytest.MonkeyPatch,
672) -> None:
673 captured: dict[str, Any] = {}
674
675 def fake_restore_journal(
676 destination: Destination,
677 recovery_key: str,
678 ) -> RestoreResult:
679 captured["destination"] = destination
680 captured["recovery_key"] = recovery_key
681 captured["argv"] = list(sys.argv)
682 return RestoreResult(
683 status="ok",
684 reason_code=None,
685 integrity_ok=True,
686 resumable=True,
687 bytes_restored=123,
688 )
689
690 monkeypatch.setattr(backup_cli, "restore_journal", fake_restore_journal)
691 payload = {
692 "repository": "b2:bucket:path",
693 "backend": "b2",
694 "credentials": {
695 "account_id": "account-id",
696 "account_key": "account-secret",
697 },
698 "recovery_key": "RECOVERYSECRET",
699 }
700
701 result = CliRunner().invoke(
702 backup_cli.app,
703 ["restore"],
704 input=json.dumps(payload),
705 )
706
707 assert result.exit_code == 0
708 assert captured["destination"].backend == "b2"
709 assert captured["recovery_key"] == "RECOVERYSECRET"
710 assert "RECOVERYSECRET" not in " ".join(captured["argv"])
711 assert "RECOVERYSECRET" not in result.output
712 assert (
713 "Restore complete: 123 bytes, integrity_ok=True, resumable=True."
714 in result.output
715 )
716
717
718@pytest.mark.parametrize(
719 ("reason_code", "detail"),
720 [
721 (
722 "integrity_unverified",
723 "integrity verification could not run "
724 "(the repository was busy or timed out)",
725 ),
726 (
727 "integrity_failed",
728 "integrity verification failed — the backup copy may be damaged",
729 ),
730 ],
731)
732def test_restore_maps_degraded(
733 monkeypatch: pytest.MonkeyPatch,
734 reason_code: str,
735 detail: str,
736) -> None:
737 monkeypatch.setattr(
738 backup_cli,
739 "restore_journal",
740 lambda *_args: RestoreResult(
741 status="degraded",
742 reason_code=reason_code,
743 integrity_ok=False,
744 resumable=True,
745 bytes_restored=123,
746 ),
747 )
748 payload = {
749 "repository": "s3:safe-bucket/path",
750 "backend": "s3",
751 "credentials": {
752 "access_key_id": "access-key",
753 "secret_access_key": "secret-key",
754 },
755 "recovery_key": "RECOVERYSECRET",
756 }
757
758 result = CliRunner().invoke(
759 backup_cli.app,
760 ["restore"],
761 input=json.dumps(payload),
762 )
763
764 assert result.exit_code == 1
765 assert (
766 f"Restored 123 bytes and saved the recovery key, but {detail} "
767 f"(reason_code={reason_code})."
768 ) in result.output
769 assert "Restore failed" not in result.output
770 assert "RECOVERYSECRET" not in result.output
771
772
773def test_restore_maps_error(monkeypatch: pytest.MonkeyPatch) -> None:
774 monkeypatch.setattr(
775 backup_cli,
776 "restore_journal",
777 lambda *_args: RestoreResult(
778 status="error",
779 reason_code="invalid_key",
780 integrity_ok=False,
781 resumable=False,
782 bytes_restored=None,
783 ),
784 )
785 payload = {
786 "repository": "s3:safe-bucket/path",
787 "backend": "s3",
788 "credentials": {
789 "access_key_id": "access-key",
790 "secret_access_key": "secret-key",
791 },
792 "recovery_key": "RECOVERYSECRET",
793 }
794
795 result = CliRunner().invoke(
796 backup_cli.app,
797 ["restore"],
798 input=json.dumps(payload),
799 )
800
801 assert result.exit_code == 1
802 assert "invalid_key" in result.output
803 assert "RECOVERYSECRET" not in result.output