mirror of
https://github.com/kevinveenbirkenbach/docker-volume-backup.git
synced 2026-07-17 14:15:13 +00:00
Kevin Veen-Birkenbach
96e6b3ea93
Backup: a container that vanishes between the docker ps listing and the swarm-task inspect (--rm one-shots, task-history GC) no longer aborts the whole backup run; it counts as not stoppable and is skipped. Restore: the postgres replay streams the dump through a spooled temp file instead of buffering it three times in memory (multi-GB dumps OOMed the restore mid-replay), and the superuser-only line filter is COPY-aware: data rows inside COPY ... FROM stdin blocks pass through untouched, so a row that happens to start with COMMENT ON EXTENSION or ALTER DEFAULT PRIVILEGES is no longer silently dropped. The e2e runner talks to the DinD daemon through docker exec instead of a host-published tcp://127.0.0.1:2375: port publishing is unreachable from sandboxed runners and from hosts with broken loopback publishing, and the unencrypted root API port disappears from the host. The debug tmp dump shrinks to tar plus docker cp against the DinD container itself. New coverage: an e2e reproducing the swarm flake end to end (service task on the volume, nothing whitelisted: the backup must succeed, the very same task container must keep running, and the service must never replace a task), unit tests for the COPY-aware filter, the swarm-task probe including the vanished-container path, filter_stoppable ordering, and the one-session FOREIGN_KEY_CHECKS drop assembly. Full suite: 35 unit, 9 integration, 30 e2e green. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
42 lines
1.6 KiB
Python
42 lines
1.6 KiB
Python
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from baudolo.backup import docker as docker_mod
|
|
from baudolo.backup.shell import BackupException
|
|
|
|
|
|
class TestIsSwarmTask(unittest.TestCase):
|
|
@patch.object(docker_mod, "execute_shell_command", return_value=["task-id-123"])
|
|
def test_true_when_task_label_present(self, _mock) -> None:
|
|
self.assertTrue(docker_mod.is_swarm_task("c1"))
|
|
|
|
@patch.object(docker_mod, "execute_shell_command", return_value=[""])
|
|
def test_false_when_label_empty(self, _mock) -> None:
|
|
self.assertFalse(docker_mod.is_swarm_task("c1"))
|
|
|
|
@patch.object(docker_mod, "execute_shell_command", return_value=[])
|
|
def test_false_when_no_output(self, _mock) -> None:
|
|
self.assertFalse(docker_mod.is_swarm_task("c1"))
|
|
|
|
@patch.object(
|
|
docker_mod,
|
|
"execute_shell_command",
|
|
side_effect=BackupException("gone"),
|
|
)
|
|
def test_vanished_container_counts_as_not_stoppable(self, _mock) -> None:
|
|
# A container removed between listing and inspect must not abort the
|
|
# whole backup run; treating it as a swarm task keeps it out of every
|
|
# stop/start and image-inspect path.
|
|
self.assertTrue(docker_mod.is_swarm_task("gone-container"))
|
|
|
|
|
|
class TestFilterStoppable(unittest.TestCase):
|
|
@patch.object(docker_mod, "is_swarm_task", side_effect=[False, True, False])
|
|
def test_mixed_list_keeps_order_and_drops_tasks(self, _mock) -> None:
|
|
result = docker_mod.filter_stoppable(["plain-1", "swarm-task", "plain-2"])
|
|
self.assertEqual(result, ["plain-1", "plain-2"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|