Initial curated projectbrain source from 8a7b1e0651dbdab97d744cbd2e653ab3970ac6c5

scripts/legacy-deployment-migration/test_storage.py

@@ -0,0 +1,62 @@
+"""Linux filesystem qualification with isolated temporary data, never appdata."""
+import os
+from pathlib import Path
+import stat
+import subprocess
+import tempfile
+import unittest
+
+from storage import clone_cold, private_write, tree_manifest
+
+
+@unittest.skipUnless(os.name == "posix", "Linux ownership/fsync qualification")
+class StorageTests(unittest.TestCase):
+    def test_private_replacement_is_durable_and_private(self):
+        with tempfile.TemporaryDirectory(prefix="migration-storage-test-") as directory:
+            path = Path(directory) / "journal.json"
+            private_write(path, b"first")
+            private_write(path, b"second")
+            self.assertEqual(path.read_bytes(), b"second")
+            self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600)
+            target = Path(directory) / "linked"
+            target.symlink_to(path)
+            with self.assertRaises(ValueError):
+                private_write(target, b"must not replace")
+            self.assertEqual(path.read_bytes(), b"second")
+
+    def test_real_cold_copy_preserves_content_modes_and_hardlinks(self):
+        with tempfile.TemporaryDirectory(prefix="migration-storage-test-") as directory:
+            source = Path(directory) / "original"
+            source.mkdir(mode=0o750)
+            payload = source / "database"
+            payload.write_bytes(b"representative fixture data")
+            payload.chmod(0o640)
+            os.link(payload, source / "hardlink")
+            destination = Path(directory) / "candidate"
+
+            def runner(command, **kwargs):
+                subprocess.run(command, check=True, capture_output=True, **kwargs)
+
+            checksum = clone_cold(source, destination, runner)
+            self.assertEqual(tree_manifest(source), tree_manifest(destination))
+            self.assertEqual(len(checksum), 64)
+            self.assertNotEqual(payload.stat().st_ino, (destination / "database").stat().st_ino)
+            (destination / "database").write_bytes(b"candidate mutation")
+            self.assertEqual(payload.read_bytes(), b"representative fixture data")
+            with self.assertRaises(ValueError):
+                clone_cold(source, destination, runner)
+
+    def test_special_files_are_rejected_before_copy(self):
+        with tempfile.TemporaryDirectory(prefix="migration-storage-test-") as directory:
+            root = Path(directory)
+            (root / "link").symlink_to("/tmp")
+            with self.assertRaises(ValueError):
+                tree_manifest(root)
+            (root / "link").unlink()
+            os.mkfifo(root / "fifo")
+            with self.assertRaises(ValueError):
+                tree_manifest(root)
+
+
+if __name__ == "__main__":
+    unittest.main()