Files
docker-volume-backup/tests/unit/restore/test_cli_password_stdin.py
T
kevinveenbirkenbachandClaude Opus 5.5 b30344d494 feat(backup): write database dumps gzip-compressed
Dumps stream through gzip into <db>.backup.sql.gz; no uncompressed copy touches the disk. Restore reads gzip and plain dumps by their magic bytes, verifies a gzip dump to its end before anything is dropped, and still finds the uncompressed names of older generations. The manifest names the compression.

🎯 Fix: 92 %
💥 Break: 20 %
💸 Cost: –
⏱️ Time: –
🛡️ Security: ~100 %
🏆 Score: 74
🚦 Gate: ⛔ break 20 % > 5

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-05 18:37:04 +00:00

67 lines
2.1 KiB
Python

import contextlib
import io
import unittest
from unittest.mock import patch
from baudolo.restore import __main__ as cli
BASE = [
"postgres",
"app_vol",
"hash",
"20260817000000",
"--repo-name",
"repo",
"--container",
"db",
"--db-name",
"app",
]
class TestRestorePasswordChannel(unittest.TestCase):
def call(self, extra, stdin=""):
stderr = io.StringIO()
with (
patch.object(cli, "restore_postgres_sql") as restore,
patch.object(cli, "locate", side_effect=lambda path: path),
patch("sys.stdin", io.StringIO(stdin)),
contextlib.redirect_stderr(stderr),
):
code = cli.main([*BASE, *extra])
return code, restore, stderr.getvalue()
def test_stdin_carries_the_password_without_its_newline(self):
code, restore, stderr = self.call(["--db-password-stdin"], stdin="s3cret\n")
self.assertEqual(code, 0)
self.assertEqual(restore.call_args.kwargs["password"], "s3cret")
self.assertNotIn("WARNING", stderr)
def test_the_argument_still_works_but_warns(self):
code, restore, stderr = self.call(["--db-password", "s3cret"])
self.assertEqual(code, 0)
self.assertEqual(restore.call_args.kwargs["password"], "s3cret")
self.assertIn("visible in the process list", stderr)
def test_both_ways_is_a_usage_error(self):
with self.assertRaises(SystemExit) as caught:
self.call(["--db-password", "x", "--db-password-stdin"], stdin="y")
self.assertEqual(caught.exception.code, 2)
def test_neither_way_is_a_usage_error(self):
with self.assertRaises(SystemExit) as caught:
self.call([])
self.assertEqual(caught.exception.code, 2)
def test_files_needs_no_password(self):
with patch.object(cli, "restore_volume_files", return_value=0) as restore:
code = cli.main(
["files", "app_vol", "hash", "20260817000000", "--repo-name", "repo"]
)
self.assertEqual(code, 0)
restore.assert_called_once()
if __name__ == "__main__":
unittest.main()