mirror of
https://github.com/kevinveenbirkenbach/docker-volume-backup.git
synced 2026-08-20 13:12:48 +00:00
Kevin Veen-Birkenbach
efcfe88e7f
The package carried no ruff configuration at all, so it ran on the defaults (E4+E7+E9+F) while infinito-nexus-core, its only consumer, holds itself to a far wider selection. Measured against that selection this tree had 224 findings. It now has none. The selector list is core's verbatim so both repositories answer to one bar. target-version stays py39 rather than core's py311, because requires-python still declares >=3.9 and pyupgrade would otherwise propose syntax the declared minimum cannot run. Every ignore carries its reason: S603/S607 in particular, since running docker and dump binaries from PATH in list form is this tool's whole job and is already injection-safe. Two conversions are judgement rather than mechanics. os.path.join(dir, '') was the rsync idiom for a trailing separator, which Path drops, so it becomes an explicit os.sep. os.path.abspath stays where Path.resolve() would follow symlinks and let a symlinked volume test as inside the snapshot subject. BREAKING CHANGE: VersionMismatch is renamed VersionMismatchError. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
128 lines
4.8 KiB
Python
128 lines
4.8 KiB
Python
"""Which volumes a snapshot of the subject actually contains.
|
|
|
|
The failure this guards against is silent: a volume with a backing store of
|
|
its own is present inside the snapshot as an empty directory, so rsync
|
|
succeeds, the generation is stamped complete, and the volume is empty in it.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
from baudolo.backup.snapshot import SnapshotError, snapshot_source, unsnapshotted
|
|
from baudolo.backup.volume import Backing
|
|
|
|
|
|
class TestUnsnapshotted(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.subject = tempfile.mkdtemp()
|
|
self.mountpoint = str(Path(self.subject) / "volumes" / "app" / "_data")
|
|
Path(self.mountpoint).mkdir(parents=True)
|
|
|
|
def backing(self, **kwargs) -> Backing:
|
|
return Backing(kwargs.pop("mountpoint", self.mountpoint), **kwargs)
|
|
|
|
def test_a_plain_local_volume_is_captured(self) -> None:
|
|
self.assertIsNone(unsnapshotted(self.backing(), self.subject))
|
|
|
|
def test_a_foreign_driver_is_not(self) -> None:
|
|
reason = unsnapshotted(self.backing(driver="rexray"), self.subject)
|
|
self.assertIn("rexray", reason)
|
|
|
|
def test_declared_driver_options_are_not(self) -> None:
|
|
reason = unsnapshotted(
|
|
self.backing(options={"type": "nfs", "device": ":/exports/app"}),
|
|
self.subject,
|
|
)
|
|
self.assertIn("backing store", reason)
|
|
|
|
def test_the_declaration_decides_not_the_mount_table(self) -> None:
|
|
"""Docker unmounts an NFS volume when its last container stops."""
|
|
with mock.patch.object(os.path, "ismount", return_value=False):
|
|
reason = unsnapshotted(self.backing(options={"type": "nfs"}), self.subject)
|
|
self.assertIsNotNone(reason)
|
|
|
|
def test_a_volume_without_a_mountpoint_is_not(self) -> None:
|
|
reason = unsnapshotted(Backing(""), self.subject)
|
|
self.assertIn("no mountpoint", reason)
|
|
|
|
def test_an_own_mount_is_not(self) -> None:
|
|
with mock.patch.object(os.path, "ismount", return_value=True):
|
|
reason = unsnapshotted(self.backing(), self.subject)
|
|
self.assertIn("own mount", reason)
|
|
|
|
def test_a_filesystem_boundary_is_not(self) -> None:
|
|
real = os.stat
|
|
|
|
def crossing(path, *args, **kwargs):
|
|
info = real(path, *args, **kwargs)
|
|
if os.path.realpath(path) == os.path.realpath(self.mountpoint):
|
|
return os.stat_result(
|
|
(info.st_mode, info.st_ino, info.st_dev + 1, *tuple(info)[3:])
|
|
)
|
|
return info
|
|
|
|
with mock.patch.object(os, "stat", side_effect=crossing):
|
|
reason = unsnapshotted(self.backing(), self.subject)
|
|
self.assertIn("filesystem boundary", reason)
|
|
|
|
def test_an_unreadable_mountpoint_is_not(self) -> None:
|
|
reason = unsnapshotted(
|
|
self.backing(mountpoint=str(Path(self.subject) / "gone")), self.subject
|
|
)
|
|
self.assertIn("could not be read", reason)
|
|
|
|
|
|
class TestSnapshotSource(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.subject = tempfile.mkdtemp()
|
|
self.mountpoint = str(Path(self.subject) / "volumes" / "app" / "_data")
|
|
Path(self.mountpoint).mkdir(parents=True)
|
|
self.snapshot = str(
|
|
Path(self.subject) / ".baudolo-tag" / "volumes" / "app" / "_data"
|
|
)
|
|
Path(self.snapshot).mkdir(parents=True)
|
|
self.backing = Backing(self.mountpoint)
|
|
|
|
def test_a_captured_volume_reads_from_the_snapshot(self) -> None:
|
|
source, reason = snapshot_source(
|
|
lambda path: self.snapshot + "/", self.backing, self.subject
|
|
)
|
|
self.assertEqual(source, self.snapshot + "/")
|
|
self.assertEqual(reason, "")
|
|
|
|
def test_an_uncaptured_volume_is_refused_before_the_resolver_runs(self) -> None:
|
|
def resolve(path):
|
|
raise AssertionError("must not resolve a volume the snapshot misses")
|
|
|
|
source, reason = snapshot_source(
|
|
resolve, Backing(self.mountpoint, options={"type": "nfs"}), self.subject
|
|
)
|
|
self.assertIsNone(source)
|
|
self.assertIn("backing store", reason)
|
|
|
|
def test_a_volume_outside_the_subject_degrades_instead_of_raising(self) -> None:
|
|
def resolve(path):
|
|
raise SnapshotError(f"{path} lies outside the snapshot subject")
|
|
|
|
source, reason = snapshot_source(resolve, self.backing, self.subject)
|
|
self.assertIsNone(source)
|
|
self.assertIn("lies outside", reason)
|
|
|
|
def test_a_volume_created_after_the_snapshot_degrades(self) -> None:
|
|
source, reason = snapshot_source(
|
|
lambda path: str(Path(self.subject) / "absent") + "/",
|
|
self.backing,
|
|
self.subject,
|
|
)
|
|
self.assertIsNone(source)
|
|
self.assertIn("created after", reason)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|