아바타 저장소 승격 계약을 완성

scripts/test_initialize_public_runtime_upload_root.py

@@ -0,0 +1,675 @@
+from __future__ import annotations
+
+import importlib.util
+import base64
+import io
+import json
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from contextlib import redirect_stdout
+from unittest.mock import patch
+
+
+SCRIPT = Path(__file__).with_name("initialize-public-runtime-upload-root.py")
+WRAPPER = Path(__file__).with_name("initialize-public-runtime-upload-root.ps1")
+DATABASE_IDENTITY = Path(__file__).with_name("public_runtime_database_identity.py")
+
+
+def _load_module():
+    spec = importlib.util.spec_from_file_location("upload_initializer", SCRIPT)
+    if spec is None or spec.loader is None:
+        raise AssertionError("initializer module is not importable")
+    module = importlib.util.module_from_spec(spec)
+    script_parent = str(SCRIPT.parent)
+    inserted = script_parent not in sys.path
+    if inserted:
+        sys.path.insert(0, script_parent)
+    try:
+        spec.loader.exec_module(module)
+    finally:
+        if inserted:
+            sys.path.remove(script_parent)
+    return module
+
+
+def _load_database_identity_module():
+    spec = importlib.util.spec_from_file_location(
+        "public_runtime_database_identity_test", DATABASE_IDENTITY
+    )
+    if spec is None or spec.loader is None:
+        raise AssertionError("database identity module is not importable")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class PublicRuntimeUploadInitializerTest(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.module = _load_module()
+        cls.database_identity = _load_database_identity_module()
+
+    def test_connected_database_identity_is_stable_and_cross_target_distinct(
+        self,
+    ) -> None:
+        first = self.database_identity.database_target_sha256(
+            database_name="vignette",
+            database_role="vignette_app",
+            server_address="10.0.0.8",
+            server_port=5432,
+        )
+        same = self.database_identity.database_target_sha256(
+            database_name="vignette",
+            database_role="vignette_app",
+            server_address="10.0.0.8",
+            server_port=5432,
+        )
+        other_database = self.database_identity.database_target_sha256(
+            database_name="vignette_shadow",
+            database_role="vignette_app",
+            server_address="10.0.0.8",
+            server_port=5432,
+        )
+        other_server = self.database_identity.database_target_sha256(
+            database_name="vignette",
+            database_role="vignette_app",
+            server_address="10.0.0.9",
+            server_port=5432,
+        )
+        self.assertEqual(first, same)
+        self.assertRegex(first, r"^[a-f0-9]{64}$")
+        self.assertNotEqual(first, other_database)
+        self.assertNotEqual(first, other_server)
+
+    def test_reference_inventory_deduplicates_objects_but_counts_rows(self) -> None:
+        one = "/uploads/profile-avatars/00000000-0000-0000-0000-000000000001-a.png"
+        two = "/uploads/profile-avatars/00000000-0000-0000-0000-000000000002-b.webp"
+        inventory = self.module.build_reference_inventory(
+            [one, one, two], expected_reference_count=3
+        )
+        self.assertEqual(3, inventory.reference_count)
+        self.assertEqual(2, inventory.unique_object_count)
+        self.assertRegex(inventory.reference_set_sha256, r"^[a-f0-9]{64}$")
+
+    def test_zero_database_references_are_valid_and_digest_bound(self) -> None:
+        inventory = self.module.build_reference_inventory(
+            [], expected_reference_count=0
+        )
+        self.assertEqual(0, inventory.reference_count)
+        self.assertEqual(0, inventory.unique_object_count)
+        self.assertRegex(inventory.reference_set_sha256, r"^[a-f0-9]{64}$")
+
+    def test_reference_validation_rejects_encoded_control_and_bad_extensions(
+        self,
+    ) -> None:
+        invalid = (
+            "/uploads/profile-avatars/name%2Fescape.png",
+            "/uploads/profile-avatars/name.exe",
+            "/uploads/profile-avatars/name\x1f.png",
+            "/uploads/profile-avatars/nested/name.png",
+            "/uploads/profile-avatars/name.png/",
+            "/uploads//profile-avatars/name.png",
+            "/uploads/other/name.png",
+        )
+        for value in invalid:
+            with self.subTest(value=repr(value)):
+                with self.assertRaises(self.module.InitializationError):
+                    self.module.normalize_avatar_reference(value)
+
+    def test_copy_is_create_only_hash_verified_and_conflicts_fail(self) -> None:
+        name = "profile-avatars/00000000-0000-0000-0000-000000000001-a.png"
+        url = f"/uploads/{name}"
+        with tempfile.TemporaryDirectory(prefix="vignette-upload-init-") as raw:
+            root = Path(raw)
+            source = root / "source"
+            target = root / "target"
+            (source / "profile-avatars").mkdir(parents=True)
+            target.mkdir()
+            (source / name).write_bytes(b"source-avatar")
+            inventory = self.module.scan_preserved_inventory([source])
+
+            first = self.module.copy_required_objects(
+                inventory=inventory, source_roots=[source], upload_root=target
+            )
+            self.assertEqual(1, first.copied_count)
+            self.assertEqual(b"source-avatar", (target / name).read_bytes())
+
+            second = self.module.copy_required_objects(
+                inventory=inventory, source_roots=[source], upload_root=target
+            )
+            self.assertEqual(1, second.reused_exact_count)
+
+            (target / name).write_bytes(b"different")
+            with self.assertRaisesRegex(
+                self.module.InitializationError,
+                "target_inventory_not_empty_or_exact",
+            ):
+                self.module.copy_required_objects(
+                    inventory=inventory,
+                    source_roots=[source],
+                    upload_root=target,
+                )
+
+    def test_source_conflict_and_missing_are_fail_closed(self) -> None:
+        name = "profile-avatars/00000000-0000-0000-0000-000000000001-a.jpg"
+        url = f"/uploads/{name}"
+        with tempfile.TemporaryDirectory(prefix="vignette-upload-source-") as raw:
+            root = Path(raw)
+            first = root / "first"
+            second = root / "second"
+            target = root / "target"
+            for directory in (first, second, target):
+                directory.mkdir()
+            (first / "profile-avatars").mkdir()
+            (second / "profile-avatars").mkdir()
+            (first / name).write_bytes(b"one")
+            (second / name).write_bytes(b"two")
+            with self.assertRaisesRegex(
+                self.module.InitializationError, "source_object_conflict"
+            ):
+                self.module.scan_preserved_inventory([first, second])
+
+            (first / name).unlink()
+            (second / name).unlink()
+            inventory = self.module.build_reference_inventory(
+                [url], expected_reference_count=1
+            )
+            preserved = self.module.scan_preserved_inventory([first, second])
+            with self.assertRaisesRegex(
+                self.module.InitializationError, "source_object_missing"
+            ):
+                self.module.assert_database_references_preserved(
+                    references=inventory,
+                    preserved=preserved,
+                )
+
+    def test_explicit_source_union_preserves_93_objects_for_8_db_refs(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="vignette-upload-union-") as raw:
+            root = Path(raw)
+            sources = [root / f"source-{index}" for index in range(3)]
+            for source in sources:
+                (source / "profile-avatars").mkdir(parents=True)
+            names = [f"avatar-{index:03d}.png" for index in range(93)]
+            for index, name in enumerate(names):
+                destination = sources[index % len(sources)] / "profile-avatars" / name
+                destination.write_bytes(f"avatar-{index}".encode("ascii"))
+            # 동일 상대 경로/동일 내용은 여러 명시 root에 있어도 하나로 보존한다.
+            duplicate = sources[1] / "profile-avatars" / names[0]
+            duplicate.write_bytes(b"avatar-0")
+            references = self.module.build_reference_inventory(
+                [f"/uploads/profile-avatars/{name}" for name in names[:8]],
+                expected_reference_count=8,
+            )
+            preserved = self.module.scan_preserved_inventory(sources)
+            self.module.assert_database_references_preserved(
+                references=references,
+                preserved=preserved,
+            )
+            self.assertEqual(93, preserved.object_count)
+            self.assertEqual(
+                sum(len(f"avatar-{index}".encode("ascii")) for index in range(93)),
+                preserved.total_size_bytes,
+            )
+            self.assertEqual(8, references.reference_count)
+            self.assertRegex(preserved.inventory_sha256, r"^[a-f0-9]{64}$")
+            self.module.assert_expected_preserved_inventory(
+                inventory=preserved,
+                expected_object_count=93,
+                expected_total_size_bytes=preserved.total_size_bytes,
+                expected_inventory_sha256=preserved.inventory_sha256,
+            )
+            with self.assertRaisesRegex(
+                self.module.InitializationError,
+                "preserved_inventory_pin_mismatch",
+            ):
+                self.module.assert_expected_preserved_inventory(
+                    inventory=preserved,
+                    expected_object_count=92,
+                    expected_total_size_bytes=preserved.total_size_bytes,
+                    expected_inventory_sha256=preserved.inventory_sha256,
+                )
+
+    def test_preserved_inventory_rejects_invalid_nested_and_changed_sources(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="vignette-upload-stability-") as raw:
+            source = Path(raw) / "source"
+            avatar_root = source / "profile-avatars"
+            avatar_root.mkdir(parents=True)
+            avatar = avatar_root / "valid.png"
+            avatar.write_bytes(b"before")
+            expected = self.module.scan_preserved_inventory([source])
+
+            avatar.write_bytes(b"after")
+            actual = self.module.scan_preserved_inventory([source])
+            with self.assertRaisesRegex(
+                self.module.InitializationError,
+                "source_inventory_changed_during_copy",
+            ):
+                self.module.assert_preserved_inventory_stable(
+                    expected=expected,
+                    actual=actual,
+                )
+
+            avatar.unlink()
+            (avatar_root / "nested").mkdir()
+            with self.assertRaisesRegex(
+                self.module.InitializationError,
+                "source_inventory_entry_invalid",
+            ):
+                self.module.scan_preserved_inventory([source])
+
+    def test_preserved_inventory_probe_requires_caller_pin_without_raw_paths(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="vignette-upload-probe-") as raw:
+            source = Path(raw) / "private-user-source"
+            avatar_root = source / "profile-avatars"
+            avatar_root.mkdir(parents=True)
+            (avatar_root / "private-avatar-name.png").write_bytes(b"avatar")
+            inventory = self.module.scan_preserved_inventory([source])
+            output = io.StringIO()
+            with redirect_stdout(output):
+                exit_code = self.module._verify_preserved_inventory_cli(
+                    [
+                        "--source-root",
+                        str(source),
+                        "--expected-preserved-object-count",
+                        "1",
+                        "--expected-preserved-total-size-bytes",
+                        str(inventory.total_size_bytes),
+                        "--expected-preserved-inventory-sha256",
+                        inventory.inventory_sha256,
+                    ]
+                )
+            payload = json.loads(output.getvalue())
+            self.assertEqual(0, exit_code)
+            self.assertEqual("verified", payload["status"])
+            self.assertEqual(
+                inventory.total_size_bytes,
+                payload["preserved_total_size_bytes"],
+            )
+            serialized = json.dumps(payload, sort_keys=True)
+            self.assertNotIn(str(source), serialized)
+            self.assertNotIn("private-avatar-name.png", serialized)
+
+    def test_failed_copy_cleanup_preserves_a_replacement_after_handle_close(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="vignette-upload-cleanup-race-") as raw:
+            root = Path(raw)
+            source = root / "source.bin"
+            target = root / "target.bin"
+            source.write_bytes(b"source-avatar")
+            expected_size, expected_sha256 = self.module._hash_regular_file(source)
+
+            def replace_after_close(*_args, **_kwargs) -> bool:
+                target.unlink()
+                target.write_bytes(b"replacement-owned-by-other-process")
+                return False
+
+            with patch.object(
+                self.module, "_target_matches", side_effect=replace_after_close
+            ):
+                with self.assertRaisesRegex(
+                    self.module.InitializationError,
+                    "target_object_verification_failed",
+                ):
+                    self.module._copy_one_create_only(
+                        source=source,
+                        target=target,
+                        expected_size=expected_size,
+                        expected_sha256=expected_sha256,
+                    )
+
+            self.assertEqual(b"replacement-owned-by-other-process", target.read_bytes())
+
+    def test_active_private_audio_always_requires_a_separate_migration(self) -> None:
+        self.module.assert_no_active_private_audio(0)
+        for count in (1, 5):
+            with self.subTest(count=count):
+                with self.assertRaisesRegex(
+                    self.module.InitializationError,
+                    "active_private_audio_requires_separate_migration",
+                ):
+                    self.module.assert_no_active_private_audio(count)
+
+    def test_offline_capture_rejects_forged_legacy_source_pins(self) -> None:
+        commit = "a" * 40
+        tree = "b" * 40
+        identity = {
+            "pid": 42,
+            "started_at_utc": "2026-08-29T00:00:00Z",
+            "executable_sha256": "c" * 64,
+            "command_line_sha256": "d" * 64,
+            "cwd_sha256": "e" * 64,
+        }
+        with tempfile.TemporaryDirectory(prefix="vignette-offline-source-") as raw:
+            source = Path(raw).resolve()
+            capture = {
+                "schema_version": self.module.OFFLINE_CAPTURE_SCHEMA_VERSION,
+                "status": "quiesced",
+                "captured_at_utc": "2026-08-29T00:00:01Z",
+                "source_commit": commit,
+                "source_tree": tree,
+                "source_root_sha256s": [
+                    self.module.canonical_path_sha256(source)
+                ],
+                "source_root_set_sha256": self.module._source_root_bindings(
+                    [source]
+                )[1],
+                "api_identity": identity,
+                "tunnel_identity": identity,
+                "listener_absent": True,
+                "tunnel_absent": True,
+                "listener_endpoint_sha256": "f" * 64,
+                "tunnel_config_sha256": "1" * 64,
+            }
+            encoded = base64.b64encode(
+                json.dumps(capture, separators=(",", ":")).encode("utf-8")
+            ).decode("ascii")
+            decoded = self.module.decode_offline_quiescence_capture(
+                encoded,
+                legacy_source_roots=[source],
+                expected_source_commit=commit,
+                expected_source_tree=tree,
+            )
+            self.assertEqual(commit, decoded["source_commit"])
+            with self.assertRaisesRegex(
+                self.module.InitializationError,
+                "offline_quiescence_capture_invalid",
+            ):
+                self.module.decode_offline_quiescence_capture(
+                    encoded,
+                    legacy_source_roots=[source],
+                    expected_source_commit="9" * 40,
+                    expected_source_tree=tree,
+                )
+
+    def test_offline_receipt_is_private_hash_bound_and_path_free(self) -> None:
+        commit = "a" * 40
+        tree = "b" * 40
+        identity = {
+            "pid": 42,
+            "started_at_utc": "2026-08-29T00:00:00Z",
+            "executable_sha256": "c" * 64,
+            "command_line_sha256": "d" * 64,
+            "cwd_sha256": "e" * 64,
+        }
+        with tempfile.TemporaryDirectory(prefix="vignette-offline-receipt-") as raw:
+            root = Path(raw).resolve()
+            source = root / "legacy-secret-user"
+            source_two = root / "legacy-secret-user-two"
+            source_three = root / "legacy-secret-user-three"
+            upload = root / "public"
+            state = root / "private"
+            for directory in (
+                source,
+                source_two,
+                source_three,
+                upload,
+                state,
+            ):
+                directory.mkdir()
+            (source / "profile-avatars").mkdir()
+            (source / "profile-avatars" / "private-name.png").write_bytes(
+                b"avatar"
+            )
+            inventory = self.module.build_reference_inventory(
+                ["/uploads/profile-avatars/private-name.png"],
+                expected_reference_count=1,
+            )
+            source_roots = [source, source_two, source_three]
+            source_root_sha256s, source_root_set_sha256 = (
+                self.module._source_root_bindings(source_roots)
+            )
+            capture = {
+                "captured_at_utc": "2026-08-29T00:00:01Z",
+                "source_commit": commit,
+                "source_tree": tree,
+                "source_root_sha256s": list(source_root_sha256s),
+                "source_root_set_sha256": source_root_set_sha256,
+                "api_identity": identity,
+                "tunnel_identity": identity,
+                "listener_endpoint_sha256": "f" * 64,
+                "tunnel_config_sha256": "1" * 64,
+            }
+            preserved = self.module.scan_preserved_inventory(source_roots)
+            invalid_objects, invalid_references = (
+                self.module.required_decode_invalid_counts(
+                    references=inventory,
+                    preserved=preserved,
+                )
+            )
+            payload = self.module.build_offline_quiescence_receipt(
+                capture=capture,
+                inventory=inventory,
+                preserved=preserved,
+                database_target_digest="2" * 64,
+            )
+            receipt, digest = self.module.write_offline_quiescence_receipt_create_only(
+                manifest_state_dir=state,
+                payload=payload,
+            )
+            proof = self.module.validate_offline_quiescence_receipt(
+                receipt_path=receipt,
+                expected_receipt_sha256=digest,
+                stable_source_root=source,
+                upload_root=upload,
+                expected_source_commit=commit,
+                expected_source_tree=tree,
+                expected_source_roots=source_roots,
+                expected_database_target_sha256="2" * 64,
+                expected_preserved_object_count=1,
+                expected_preserved_total_size_bytes=preserved.total_size_bytes,
+                expected_preserved_inventory_sha256=preserved.inventory_sha256,
+                expected_preserved_decode_valid_count=preserved.decode_valid_count,
+                expected_preserved_decode_invalid_count=(
+                    preserved.decode_invalid_count
+                ),
+                expected_required_decode_invalid_object_count=invalid_objects,
+                expected_required_decode_invalid_reference_count=(
+                    invalid_references
+                ),
+                expected_reference_count=1,
+                expected_unique_object_count=1,
+                expected_reference_set_sha256=inventory.reference_set_sha256,
+            )
+            hash_only_proof = self.module.validate_offline_quiescence_receipt(
+                receipt_path=receipt,
+                expected_receipt_sha256=digest,
+                stable_source_root=source,
+                upload_root=upload,
+                expected_source_commit=commit,
+                expected_source_tree=tree,
+                expected_source_roots=None,
+                expected_source_root_sha256s=source_root_sha256s,
+                expected_source_root_set_sha256=source_root_set_sha256,
+                expected_database_target_sha256="2" * 64,
+                expected_preserved_object_count=1,
+                expected_preserved_total_size_bytes=preserved.total_size_bytes,
+                expected_preserved_inventory_sha256=preserved.inventory_sha256,
+                expected_preserved_decode_valid_count=preserved.decode_valid_count,
+                expected_preserved_decode_invalid_count=(
+                    preserved.decode_invalid_count
+                ),
+                expected_required_decode_invalid_object_count=invalid_objects,
+                expected_required_decode_invalid_reference_count=(
+                    invalid_references
+                ),
+                expected_reference_count=1,
+                expected_unique_object_count=1,
+                expected_reference_set_sha256=inventory.reference_set_sha256,
+            )
+            serialized = json.dumps(payload, sort_keys=True)
+            self.assertRegex(proof.receipt_sha256, r"^[a-f0-9]{64}$")
+            self.assertEqual(tuple(source_root_sha256s), proof.source_root_sha256s)
+            self.assertEqual(source_root_set_sha256, proof.source_root_set_sha256)
+            self.assertEqual(proof.receipt_sha256, hash_only_proof.receipt_sha256)
+            self.assertNotIn(str(root), serialized)
+            self.assertNotIn("private-name.png", serialized)
+            self.assertNotIn("/uploads/", serialized)
+
+    def test_runtime_freeze_requires_exact_loopback_health_and_token(self) -> None:
+        token_sha256 = "b" * 64
+        health = json.dumps(
+            {
+                "upload_write_freeze": {
+                    "capable": True,
+                    "active": True,
+                    "valid": True,
+                    "in_flight": 0,
+                    "token_sha256": token_sha256,
+                }
+            }
+        ).encode("utf-8")
+
+        class Response:
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *unused):
+                return False
+
+            def read(self):
+                return health
+
+        class Opener:
+            def open(self, request, timeout):
+                return Response()
+
+        with patch.object(
+            self.module.urllib.request,
+            "build_opener",
+            return_value=Opener(),
+        ):
+            self.module.wait_for_drained_runtime_freeze(
+                health_url="http://127.0.0.1:8001/health",
+                token_sha256=token_sha256,
+                timeout_seconds=0.1,
+            )
+        with self.assertRaisesRegex(
+            self.module.InitializationError, "runtime_health_url_invalid"
+        ):
+            self.module.wait_for_drained_runtime_freeze(
+                health_url="https://example.invalid/health",
+                token_sha256=token_sha256,
+                timeout_seconds=0.1,
+            )
+
+    def test_manifest_and_result_are_privacy_safe(self) -> None:
+        name = "profile-avatars/00000000-0000-0000-0000-000000000001-secret.png"
+        url = f"/uploads/{name}"
+        with tempfile.TemporaryDirectory(prefix="vignette-upload-manifest-") as raw:
+            root = Path(raw)
+            upload = root / "upload"
+            state = root / "state"
+            freeze = state / "freeze.json"
+            (upload / "profile-avatars").mkdir(parents=True)
+            state.mkdir()
+            (upload / name).write_bytes(b"avatar")
+            freeze.write_text("{}", encoding="utf-8")
+            inventory = self.module.build_reference_inventory(
+                [url, url], expected_reference_count=2
+            )
+            preserved = self.module.scan_preserved_inventory([upload])
+            payload = self.module.privacy_safe_manifest(
+                upload_root=upload,
+                inventory=inventory,
+                preserved=preserved,
+                database_target_sha256="b" * 64,
+                freeze_token_sha256="a" * 64,
+                freeze_path=freeze,
+            )
+            _, digest = self.module.write_manifest_create_only(
+                manifest_state_dir=state, payload=payload
+            )
+            serialized = json.dumps(payload, sort_keys=True)
+            self.assertNotIn("secret.png", serialized)
+            self.assertNotIn("00000000-0000", serialized)
+            self.assertNotIn("/uploads/", serialized)
+            self.assertNotIn(str(root), serialized)
+            self.assertRegex(digest, r"^[a-f0-9]{64}$")
+            contract = payload["database_reference_contract"]
+            self.assertEqual(2, contract["reference_count"])
+            self.assertEqual(1, contract["unique_object_count"])
+            self.assertEqual(2, payload["required_reference_count"])
+            self.assertEqual(
+                2, payload["preserved_objects"][0]["reference_count"]
+            )
+            self.assertEqual(1, payload["preserved_object_count"])
+            self.assertEqual(
+                preserved.inventory_sha256,
+                payload["preserved_object_set_sha256"],
+            )
+            self.assertEqual("b" * 64, payload["database_target_sha256"])
+            self.assertEqual(
+                inventory.reference_set_sha256, payload["reference_set_sha256"]
+            )
+            self.assertEqual(
+                payload["reference_set_sha256"], contract["reference_set_sha256"]
+            )
+            self.assertEqual(
+                [
+                    {
+                        "path_sha256": payload["preserved_objects"][0]["path_sha256"],
+                        "reference_count": 2,
+                    }
+                ],
+                contract["objects"],
+            )
+            result = self.module.privacy_safe_result(
+                manifest_path=state / f"public-avatar-upload-{digest}.json",
+                manifest_sha256=digest,
+                upload_root=upload,
+                inventory=inventory,
+                preserved=preserved,
+                copy_proof=self.module.CopyProof(1, 1, 0),
+                freeze_token_sha256="a" * 64,
+            )
+            serialized_result = json.dumps(result, sort_keys=True)
+            self.assertNotIn("secret.png", serialized_result)
+            self.assertNotIn(str(root), serialized_result)
+            self.assertEqual(
+                {
+                    "status",
+                    "manifest_sha256",
+                    "manifest_path_sha256",
+                    "root_path_sha256",
+                    "preserved_object_count",
+                    "preserved_total_size_bytes",
+                    "preserved_inventory_sha256",
+                    "preserved_decode_valid_count",
+                    "preserved_decode_invalid_count",
+                    "required_decode_invalid_object_count",
+                    "required_decode_invalid_reference_count",
+                    "required_object_count",
+                    "database_reference_count",
+                    "database_reference_set_sha256",
+                    "copied_object_count",
+                    "reused_exact_object_count",
+                    "write_freeze_token_sha256",
+                },
+                set(result),
+            )
+
+    def test_wrapper_owns_creation_and_freeze_create_new(self) -> None:
+        source = WRAPPER.read_text(encoding="utf-8")
+        self.assertIn("-CreateIfMissing", source)
+        self.assertIn("[System.IO.FileMode]::CreateNew", source)
+        self.assertIn("ExpectedReferenceCount", source)
+        self.assertIn("ExpectedPreservedObjectCount", source)
+        self.assertIn("ExpectedPreservedTotalSizeBytes", source)
+        self.assertIn("ExpectedPreservedInventorySha256", source)
+        self.assertIn('"--expected-preserved-object-count"', source)
+        self.assertIn('"--expected-preserved-total-size-bytes"', source)
+        self.assertIn('"--expected-preserved-inventory-sha256"', source)
+        self.assertIn("parsedOutput.preserved_total_size_bytes", source)
+        self.assertIn("Remove-OwnedWriteFreeze", source)
+        self.assertIn("Assert-StableInitializerSourceProvenance", source)
+        self.assertIn("symbolic-ref -q HEAD", source)
+        self.assertIn("status --porcelain=v1 --untracked-files=all", source)
+        worker = SCRIPT.read_text(encoding="utf-8")
+        self.assertIn('open("xb")', worker)
+        self.assertNotIn("shutil.move", worker)
+        self.assertNotIn("os.replace", worker)
+
+
+if __name__ == "__main__":
+    unittest.main()