import concurrent.futures import io import unittest from datetime import datetime, timezone from unittest import mock import dumpling_packed_backup_init as worker class FakeS3Error(Exception): def __init__(self, code, status): super().__init__(code) self.response = { "Error": {"Code": code}, "ResponseMetadata": {"HTTPStatusCode": status}, } class ExportWorkerTests(unittest.TestCase): backup_time = datetime(2026, 7, 31, 12, 34, 56, tzinfo=timezone.utc) run_id = "dumpling-packed-export-run-1" def test_dumpling_arguments_are_appended_exactly_once(self): command = worker.dumpling_command( "s3://bucket/packed/meta.meta", "s3://bucket/export/cluster-1/", "us-east-1", "/run/gpg/public-key", ["--threads", "4", "--filter", "database.table"], ) self.assertEqual(command.count("--threads"), 1) self.assertEqual(command.count("--filter"), 4) self.assertEqual( command[-4:], ["--threads", "4", "--filter", "database.table"], ) def test_dumpling_timeout_is_parsed_as_seconds(self): args = worker.parse_args( [ "--s3-prefix", "s3://bucket/packed_backup/native/20260731/" "123456.meta/batch/meta", "--out-prefix", "s3://bucket/export", "--gpg", "/tmp/public-key", "--dumpling-timeout", "300", ] ) self.assertEqual(args.dumpling_timeout, 300) def test_dumpling_output_is_versioned_but_marker_key_is_unchanged(self): target = "s3://bucket/export/cluster-1/" output = worker.dumpling_output_uri( target, self.backup_time, self.run_id, ) marker_uri, bucket, marker_key = worker.complete_marker_location(target) self.assertEqual( output, "s3://bucket/export/cluster-1/" "2026-07-31T12:34:56Z:dumpling-packed-export-run-1/", ) self.assertEqual(marker_uri, "s3://bucket/export/cluster-1/complete") self.assertEqual(bucket, "bucket") self.assertEqual(marker_key, "export/cluster-1/complete") def test_complete_marker_replacement_uses_original_etag(self): client = mock.Mock() marker = worker.CompleteMarker( uri="s3://bucket/export/cluster-1/complete", job_instance_name="old-run", backup_time=datetime(2026, 7, 30, tzinfo=timezone.utc), etag='"old-etag"', ) worker.write_complete_marker( client, "s3://bucket/export/cluster-1/", self.run_id, self.backup_time, marker, ) request = client.put_object.call_args.kwargs self.assertEqual(request["IfMatch"], '"old-etag"') self.assertNotIn("IfNoneMatch", request) self.assertEqual( request["Body"], b"2026-07-31T12:34:56Z:dumpling-packed-export-run-1", ) self.assertEqual( request["Metadata"], { "job-instance-name": self.run_id, "backup-time": "2026-07-31T12:34:56Z", }, ) def test_complete_marker_creation_is_conditional_on_absence(self): client = mock.Mock() worker.write_complete_marker( client, "s3://bucket/export/cluster-1/", self.run_id, self.backup_time, None, ) request = client.put_object.call_args.kwargs self.assertEqual(request["IfNoneMatch"], "*") self.assertNotIn("IfMatch", request) def test_changed_complete_marker_rejects_pointer_update(self): client = mock.Mock() client.put_object.side_effect = FakeS3Error("PreconditionFailed", 412) marker = worker.CompleteMarker( uri="s3://bucket/export/cluster-1/complete", job_instance_name="old-run", backup_time=datetime(2026, 7, 30, tzinfo=timezone.utc), etag='"old-etag"', ) with self.assertRaisesRegex( worker.ExportError, "changed since it was first read", ): worker.write_complete_marker( client, "s3://bucket/export/cluster-1/", self.run_id, self.backup_time, marker, ) def test_cleanup_only_selects_older_versions_and_legacy_output(self): marker_key = "export/cluster-1/complete" prefix = "export/cluster-1/" object_keys = [ marker_key, f"{prefix}legacy.csv.gpg", f"{prefix}README", f"{prefix}unknown/nested.csv.gpg", f"{prefix}2026-07-30T12:34:56Z:old-run/data.csv.gpg", f"{prefix}2026-07-31T12:34:56Z:other-run/data.csv.gpg", f"{prefix}2026-08-01T12:34:56Z:new-run/data.csv.gpg", ] stale_keys = worker.stale_export_object_keys( marker_key, object_keys, self.backup_time, ) self.assertEqual( stale_keys, [ f"{prefix}2026-07-30T12:34:56Z:old-run/data.csv.gpg", f"{prefix}legacy.csv.gpg", ], ) def test_cleanup_runs_only_after_marker_update(self): events = [] client = mock.Mock() progress = worker.Progress(total=1) original_marker = worker.CompleteMarker( uri="s3://bucket/export/cluster-1/complete", job_instance_name="old-run", backup_time=datetime(2026, 7, 30, tzinfo=timezone.utc), etag='"old-etag"', ) def write_marker(*_args): events.append("marker") return original_marker.uri def cleanup(*_args): events.append("cleanup") with ( concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor, mock.patch.object( worker, "existing_complete_marker", return_value=original_marker, ), mock.patch.object( worker, "run_dumpling_process", return_value=(0, 123), ), mock.patch.object( worker, "write_complete_marker", side_effect=write_marker, ) as write_complete_marker, mock.patch.object( worker, "remove_stale_export_objects", side_effect=cleanup, ), ): process_id = worker.run_dumpling( ["/dumpling"], 1, "s3://bucket/source.meta", "s3://bucket/export/cluster-1/", "s3://bucket/export/cluster-1/" "2026-07-31T12:34:56Z:dumpling-packed-export-run-1/", client, progress, self.run_id, self.backup_time, None, executor, ) self.assertEqual(process_id, 123) self.assertEqual(events, ["marker", "cleanup"]) self.assertIs(write_complete_marker.call_args.args[-1], original_marker) self.assertEqual(progress.snapshot(), (1, 0, 1)) def test_failed_marker_update_never_starts_cleanup(self): client = mock.Mock() progress = worker.Progress(total=1) cleanup = mock.Mock() with ( concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor, mock.patch.object( worker, "existing_complete_marker", return_value=None, ), mock.patch.object( worker, "run_dumpling_process", return_value=(0, 123), ), mock.patch.object( worker, "write_complete_marker", side_effect=worker.ExportError("marker changed"), ), mock.patch.object( worker, "remove_stale_export_objects", cleanup, ), self.assertRaisesRegex(worker.ExportError, "marker changed"), ): worker.run_dumpling( ["/dumpling"], 1, "s3://bucket/source.meta", "s3://bucket/export/cluster-1/", "s3://bucket/export/cluster-1/" "2026-07-31T12:34:56Z:dumpling-packed-export-run-1/", client, progress, self.run_id, self.backup_time, None, executor, ) cleanup.assert_not_called() self.assertEqual(progress.snapshot(), (0, 0, 1)) def test_timed_out_process_group_is_terminated(self): class FakeProcess: pid = 4321 def __init__(self): self.stdout = io.StringIO("") self.stderr = io.StringIO("") self.wait_timeouts = [] self.wait_count = 0 def poll(self): return None def wait(self, timeout=None): self.wait_timeouts.append(timeout) self.wait_count += 1 if self.wait_count == 1: assert timeout is not None raise worker.subprocess.TimeoutExpired( ["/dumpling"], timeout, ) return -worker.signal.SIGTERM process = FakeProcess() with ( mock.patch.object( worker.subprocess, "Popen", return_value=process, ) as popen, mock.patch.object(worker.os, "killpg") as killpg, self.assertRaises(worker.DumplingTimedOut) as raised, ): worker.run_dumpling_process( ["/dumpling"], 1, 1, "s3://bucket/source.meta", "s3://bucket/export/version/", 5, ) self.assertEqual(raised.exception.process_id, process.pid) self.assertEqual(raised.exception.timeout_seconds, 5) self.assertEqual( process.wait_timeouts, [5, worker.DUMPLING_TERMINATE_GRACE_SECONDS], ) killpg.assert_called_once_with(process.pid, worker.signal.SIGTERM) self.assertTrue(popen.call_args.kwargs["start_new_session"]) def test_timed_out_process_group_is_killed_after_grace_period(self): class StubbornProcess: pid = 4321 def poll(self): return None def wait(self, timeout=None): if timeout is not None: raise worker.subprocess.TimeoutExpired( ["/dumpling"], timeout, ) return -worker.signal.SIGKILL process = StubbornProcess() with mock.patch.object(worker.os, "killpg") as killpg: returncode = worker.stop_timed_out_dumpling(process, 5) self.assertEqual(returncode, -worker.signal.SIGKILL) self.assertEqual( killpg.call_args_list, [ mock.call(process.pid, worker.signal.SIGTERM), mock.call(process.pid, worker.signal.SIGKILL), ], ) def test_timed_out_item_does_not_block_following_export(self): items = [ worker.ExportItem( source="s3://bucket/a.meta", target="s3://bucket/export/a/", output="s3://bucket/export/a/version/", command=["/dumpling", "a"], ), worker.ExportItem( source="s3://bucket/b.meta", target="s3://bucket/export/b/", output="s3://bucket/export/b/version/", command=["/dumpling", "b"], ), ] progress = worker.Progress(total=2) attempted_sources = [] def run_process(_command, _total, _attempt, source, _output, timeout): attempted_sources.append(source) self.assertEqual(timeout, 5) if source.endswith("a.meta"): raise worker.DumplingTimedOut(101, timeout) return 0, 102 with ( mock.patch.object( worker, "existing_complete_marker", return_value=None, ), mock.patch.object( worker, "run_dumpling_process", side_effect=run_process, ), mock.patch.object( worker, "write_complete_marker", return_value="s3://bucket/export/b/complete", ) as write_marker, mock.patch.object(worker, "remove_stale_export_objects"), ): worker.run_export_items( items, 1, 2, mock.Mock(), progress, self.run_id, self.backup_time, 5, ) self.assertEqual( attempted_sources, ["s3://bucket/a.meta", "s3://bucket/b.meta"], ) self.assertEqual(write_marker.call_count, 1) self.assertEqual(progress.snapshot(), (1, 1, 2)) if __name__ == "__main__": unittest.main()