fix(image): clean partial mounts and refuse mounted export deletion

tests/test_image.py

@@ -97,6 +97,58 @@ class ImageTests(unittest.TestCase):
         with self.assertRaises(image.ImageError):
             image.kernel_version(self.root)
 
+    def test_mount_faults_clean_only_successful_bindings(self):
+        source = Path(__file__).resolve().parents[1]
+        tools = Path(self.tmp.name) / 'tools'
+        tools.mkdir()
+        for name in ('python3', 'chroot', 'ln', 'systemctl'):
+            script = tools / name
+            script.write_text('#!/bin/sh\nexit 0\n')
+            script.chmod(0o755)
+        mount = tools / 'mount'
+        mount.write_text('#!/bin/sh\nn=$(cat "$STATE/count" 2>/dev/null || echo 0)\n'
+                         'n=$((n+1)); echo "$n" > "$STATE/count"\n'
+                         'test "$n" != "$FAIL_MOUNT"\n')
+        mount.chmod(0o755)
+        unmount = tools / 'umount'
+        unmount.write_text('#!/bin/sh\nprintf "%s\\n" "$@" >> "$STATE/unmounts"\n'
+                           'test "$FAIL_UNMOUNT" != 1\n')
+        unmount.chmod(0o755)
+        for failed, cleanup_failed in ((1, 0), (2, 0), (3, 0), (0, 0), (0, 1)):
+            with self.subTest(failed=failed, cleanup_failed=cleanup_failed):
+                state = Path(self.tmp.name) / f'state-{failed}-{cleanup_failed}'
+                state.mkdir()
+                env = dict(os.environ, PATH=str(tools) + ':' + os.environ['PATH'],
+                           STATE=str(state), FAIL_MOUNT=str(failed), FAIL_UNMOUNT=str(cleanup_failed))
+                result = subprocess.run(['bash', str(source / 'image/configure-rootfs.sh'), '/hydra-output/rootfs'],
+                                        env=env, capture_output=True, text=True)
+                expected = ['/hydra-output/rootfs/dev', '/hydra-output/rootfs/proc', '/hydra-output/rootfs/sys']
+                expected = list(reversed(expected[:failed - 1 if failed else 3]))
+                actual = (state / 'unmounts').read_text().splitlines() if (state / 'unmounts').exists() else []
+                self.assertEqual(actual, expected)
+                self.assertEqual(result.returncode == 0, not failed and not cleanup_failed)
+
+    def test_retry_refuses_removing_owned_rootfs_with_stale_child_mount(self):
+        from hydra.build import rootfs_retry_guard_script
+        marker = self.file('.hydra-export-owned')
+        preserved = self.file('dev/guest-data')
+        tools = Path(self.tmp.name) / 'guard-tools'
+        tools.mkdir()
+        findmnt = tools / 'findmnt'
+        findmnt.write_text('#!/bin/sh\nprintf "%s\\n" / "$STALE_MOUNT"\n')
+        findmnt.chmod(0o755)
+        env = dict(os.environ, PATH=str(tools) + ':' + os.environ['PATH'], STALE_MOUNT=str(self.root / 'dev'))
+        script = 'set -eu\nroot=' + shlex.quote(str(self.root)) + '\n' + rootfs_retry_guard_script()
+        result = subprocess.run(['bash', '-c', script], env=env, capture_output=True, text=True)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertTrue(marker.is_file())
+        self.assertTrue(preserved.is_file())
+        self.assertIn('mounted', result.stderr)
+        env['STALE_MOUNT'] = str(Path(self.tmp.name) / 'different-rootfs')
+        result = subprocess.run(['bash', '-c', script], env=env, capture_output=True, text=True)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertFalse(self.root.exists())
+
     def test_export_sanitization_preserves_update_inputs(self):
         preserved = ['var/db/pkg/sys-apps/portage-3/USE', 'var/lib/portage/world',
                      'var/db/repos/gentoo/profiles/repo_name', 'usr/share/openpgp-keys/gentoo-release.asc',