675 lines
29 KiB
Python
675 lines
29 KiB
Python
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()
|