1230 lines
52 KiB
Python
1230 lines
52 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from contextlib import nullcontext
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from PIL import Image
|
|
from starlette.datastructures import Headers, UploadFile
|
|
|
|
from . import upload_storage as upload_storage_module
|
|
from .upload_storage import (
|
|
UploadWriteFreezeGate,
|
|
UploadWriteFrozen,
|
|
assert_path_without_reparse,
|
|
build_privacy_safe_manifest,
|
|
database_target_sha256,
|
|
inspect_public_avatar_image,
|
|
is_runtime_generated_public_avatar_relative_path,
|
|
normalize_public_avatar_relative_path,
|
|
public_avatar_relative_path_sha256,
|
|
validate_current_avatar_references,
|
|
validate_upload_manifest,
|
|
)
|
|
|
|
DATABASE_TARGET_SHA256 = hashlib.sha256(b"connected-database-target").hexdigest()
|
|
|
|
|
|
def _valid_png_bytes() -> bytes:
|
|
buffer = io.BytesIO()
|
|
Image.new("RGBA", (1, 1), (0, 0, 0, 0)).save(buffer, format="PNG")
|
|
return buffer.getvalue()
|
|
|
|
|
|
class _AsyncValueContext:
|
|
def __init__(self, value):
|
|
self.value = value
|
|
|
|
async def __aenter__(self):
|
|
return self.value
|
|
|
|
async def __aexit__(self, exc_type, exc, traceback):
|
|
return False
|
|
|
|
|
|
class _RuntimeDatabaseConnection:
|
|
def __init__(self, *, avatar_urls: list[str], database_name: str = "vignette"):
|
|
self.avatar_urls = avatar_urls
|
|
self.database_name = database_name
|
|
self.transaction_options: list[dict[str, object]] = []
|
|
self.executed: list[str] = []
|
|
self.codec_names: list[str] = []
|
|
|
|
def transaction(self, **options):
|
|
self.transaction_options.append(options)
|
|
return _AsyncValueContext(self)
|
|
|
|
async def fetchrow(self, _query: str):
|
|
return {
|
|
"database_name": self.database_name,
|
|
"database_role": "vignette_app",
|
|
"server_address": "10.0.0.5",
|
|
"server_port": 5432,
|
|
}
|
|
|
|
async def execute(self, query: str):
|
|
self.executed.append(query)
|
|
|
|
async def set_type_codec(self, name: str, **_options):
|
|
self.codec_names.append(name)
|
|
|
|
async def fetch(self, _query: str):
|
|
return [{"avatar_url": value} for value in self.avatar_urls]
|
|
|
|
|
|
class _RuntimeDatabasePool:
|
|
def __init__(self, connection: _RuntimeDatabaseConnection):
|
|
self.connection = connection
|
|
self.acquire_count = 0
|
|
|
|
def acquire(self):
|
|
self.acquire_count += 1
|
|
return _AsyncValueContext(self.connection)
|
|
|
|
|
|
class UploadStorageContractTest(unittest.TestCase):
|
|
def test_broken_png_crc_is_decode_invalid_not_an_initializer_error(self) -> None:
|
|
content = bytearray(_valid_png_bytes())
|
|
idat = content.index(b"IDAT") + 4
|
|
content[idat] ^= 0x01
|
|
proof = inspect_public_avatar_image(
|
|
"profile-avatars/broken.png",
|
|
bytes(content),
|
|
)
|
|
self.assertFalse(proof.valid)
|
|
|
|
def test_api_and_release_probe_share_connected_database_identity_digest(self) -> None:
|
|
helper_path = (
|
|
Path(__file__).resolve().parents[3]
|
|
/ "scripts"
|
|
/ "public_runtime_database_identity.py"
|
|
)
|
|
spec = importlib.util.spec_from_file_location(
|
|
"public_runtime_database_identity_contract_test",
|
|
helper_path,
|
|
)
|
|
self.assertIsNotNone(spec)
|
|
self.assertIsNotNone(spec.loader if spec is not None else None)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec is not None and spec.loader is not None
|
|
spec.loader.exec_module(module)
|
|
release_database_target_sha256 = module.database_target_sha256
|
|
|
|
identity = {
|
|
"database_name": "vignette",
|
|
"database_role": "vignette_app",
|
|
"server_address": "10.0.0.5",
|
|
"server_port": 5432,
|
|
}
|
|
self.assertEqual(
|
|
release_database_target_sha256(**identity),
|
|
database_target_sha256(**identity),
|
|
)
|
|
|
|
def test_manifest_proves_required_object_without_recording_its_path(self) -> None:
|
|
with tempfile.TemporaryDirectory(prefix="vignette-upload-manifest-") as raw:
|
|
root = Path(raw) / "uploads"
|
|
object_path = root / "profile-avatars" / "private-object.jpg"
|
|
object_path.parent.mkdir(parents=True)
|
|
object_path.write_bytes(b"avatar-bytes")
|
|
token_sha = hashlib.sha256(b"freeze-token").hexdigest()
|
|
payload = build_privacy_safe_manifest(
|
|
upload_root=root,
|
|
references=[("profile-avatars/private-object.jpg", "profile_avatar")],
|
|
preserved_paths=["profile-avatars/private-object.jpg"],
|
|
database_target_sha256=DATABASE_TARGET_SHA256,
|
|
write_freeze_token_sha256=token_sha,
|
|
)
|
|
serialized = json.dumps(payload, sort_keys=True)
|
|
self.assertNotIn("private-object", serialized)
|
|
self.assertNotIn('"relative_path":', serialized)
|
|
|
|
manifest_dir = Path(raw) / "private-state" / "manifests"
|
|
manifest_dir.mkdir(parents=True)
|
|
manifest = manifest_dir / "manifest.json"
|
|
manifest.write_text(serialized, encoding="utf-8")
|
|
expected_sha = hashlib.sha256(manifest.read_bytes()).hexdigest()
|
|
proof = validate_upload_manifest(
|
|
upload_root=root,
|
|
manifest_path=manifest,
|
|
expected_manifest_sha256=expected_sha,
|
|
expected_write_freeze_path=Path(raw) / "upload-write.freeze",
|
|
expected_database_target_sha256=DATABASE_TARGET_SHA256,
|
|
require_freeze_path_binding=False,
|
|
)
|
|
self.assertEqual(1, proof.required_object_count)
|
|
public_manifest = root / "manifest.json"
|
|
public_manifest.write_text(serialized, encoding="utf-8")
|
|
public_manifest_sha = hashlib.sha256(
|
|
public_manifest.read_bytes()
|
|
).hexdigest()
|
|
with self.assertRaisesRegex(ValueError, "outside the public upload root"):
|
|
validate_upload_manifest(
|
|
upload_root=root,
|
|
manifest_path=public_manifest,
|
|
expected_manifest_sha256=public_manifest_sha,
|
|
expected_write_freeze_path=Path(raw) / "upload-write.freeze",
|
|
expected_database_target_sha256=DATABASE_TARGET_SHA256,
|
|
require_freeze_path_binding=False,
|
|
)
|
|
object_path.write_bytes(b"drift")
|
|
with self.assertRaisesRegex(ValueError, "content hash"):
|
|
validate_upload_manifest(
|
|
upload_root=root,
|
|
manifest_path=manifest,
|
|
expected_manifest_sha256=expected_sha,
|
|
expected_write_freeze_path=Path(raw) / "upload-write.freeze",
|
|
expected_database_target_sha256=DATABASE_TARGET_SHA256,
|
|
require_freeze_path_binding=False,
|
|
)
|
|
|
|
def test_duplicate_db_references_are_aggregated_without_raw_paths(self) -> None:
|
|
with tempfile.TemporaryDirectory(prefix="vignette-upload-duplicates-") as raw:
|
|
root = Path(raw) / "uploads"
|
|
object_path = root / "profile-avatars" / "shared-avatar.png"
|
|
object_path.parent.mkdir(parents=True)
|
|
object_path.write_bytes(b"shared-avatar")
|
|
token_sha = hashlib.sha256(b"freeze-token").hexdigest()
|
|
payload = build_privacy_safe_manifest(
|
|
upload_root=root,
|
|
references=[
|
|
("profile-avatars/shared-avatar.png", "profile_avatar"),
|
|
("profile-avatars/shared-avatar.png", "profile_avatar"),
|
|
],
|
|
preserved_paths=["profile-avatars/shared-avatar.png"],
|
|
database_target_sha256=DATABASE_TARGET_SHA256,
|
|
write_freeze_token_sha256=token_sha,
|
|
)
|
|
self.assertEqual(1, payload["required_object_count"])
|
|
self.assertEqual(2, payload["required_reference_count"])
|
|
self.assertEqual(
|
|
2, payload["preserved_objects"][0]["reference_count"]
|
|
)
|
|
serialized = json.dumps(payload, sort_keys=True)
|
|
self.assertNotIn("shared-avatar", serialized)
|
|
|
|
manifest_dir = Path(raw) / "private-state"
|
|
manifest_dir.mkdir()
|
|
manifest = manifest_dir / "manifest.json"
|
|
manifest.write_text(serialized, encoding="utf-8")
|
|
proof = validate_upload_manifest(
|
|
upload_root=root,
|
|
manifest_path=manifest,
|
|
expected_manifest_sha256=hashlib.sha256(
|
|
manifest.read_bytes()
|
|
).hexdigest(),
|
|
expected_write_freeze_path=Path(raw) / "write-freeze.json",
|
|
expected_database_target_sha256=DATABASE_TARGET_SHA256,
|
|
require_freeze_path_binding=False,
|
|
)
|
|
current = validate_current_avatar_references(
|
|
upload_root=root,
|
|
avatar_urls=[
|
|
"/uploads/profile-avatars/shared-avatar.png",
|
|
"/uploads/profile-avatars/shared-avatar.png",
|
|
],
|
|
initial_manifest=proof,
|
|
expected_database_target_sha256=DATABASE_TARGET_SHA256,
|
|
)
|
|
self.assertEqual(1, current.object_count)
|
|
self.assertEqual(2, current.reference_count)
|
|
self.assertEqual(proof.reference_set_sha256, current.reference_set_sha256)
|
|
|
|
def test_runtime_generated_current_references_redecode_and_count_corruption(
|
|
self,
|
|
) -> None:
|
|
with tempfile.TemporaryDirectory(prefix="vignette-upload-current-decode-") as raw:
|
|
root = Path(raw) / "uploads"
|
|
avatar_root = root / "profile-avatars"
|
|
avatar_root.mkdir(parents=True)
|
|
preserved = avatar_root / "preserved.png"
|
|
preserved.write_bytes(_valid_png_bytes())
|
|
payload = build_privacy_safe_manifest(
|
|
upload_root=root,
|
|
references=[],
|
|
preserved_paths=["profile-avatars/preserved.png"],
|
|
database_target_sha256=DATABASE_TARGET_SHA256,
|
|
write_freeze_token_sha256=hashlib.sha256(b"freeze-token").hexdigest(),
|
|
)
|
|
private = Path(raw) / "private"
|
|
private.mkdir()
|
|
manifest = private / "manifest.json"
|
|
manifest.write_text(json.dumps(payload), encoding="utf-8")
|
|
proof = validate_upload_manifest(
|
|
upload_root=root,
|
|
manifest_path=manifest,
|
|
expected_manifest_sha256=hashlib.sha256(
|
|
manifest.read_bytes()
|
|
).hexdigest(),
|
|
expected_write_freeze_path=private / "write-freeze.json",
|
|
expected_database_target_sha256=DATABASE_TARGET_SHA256,
|
|
require_freeze_path_binding=False,
|
|
)
|
|
|
|
valid_name = (
|
|
"8da1b8fc-3bdd-4e72-86f7-d17fc315c4cb-"
|
|
"AbCdEf_12345.png"
|
|
)
|
|
corrupt_name = (
|
|
"1265087b-d3d0-4e85-82bc-499d993a3f89-"
|
|
"QrStUv_67890.png"
|
|
)
|
|
(avatar_root / valid_name).write_bytes(_valid_png_bytes())
|
|
(avatar_root / corrupt_name).write_bytes(b"not-a-png")
|
|
|
|
current = validate_current_avatar_references(
|
|
upload_root=root,
|
|
avatar_urls=[
|
|
f"/uploads/profile-avatars/{valid_name}",
|
|
f"/uploads/profile-avatars/{corrupt_name}",
|
|
f"/uploads/profile-avatars/{corrupt_name}",
|
|
],
|
|
initial_manifest=proof,
|
|
expected_database_target_sha256=DATABASE_TARGET_SHA256,
|
|
)
|
|
|
|
self.assertEqual(2, current.object_count)
|
|
self.assertEqual(3, current.reference_count)
|
|
self.assertEqual(1, current.decode_invalid_object_count)
|
|
self.assertEqual(2, current.decode_invalid_reference_count)
|
|
|
|
arbitrary = avatar_root / "unbound.png"
|
|
arbitrary.write_bytes(_valid_png_bytes())
|
|
with self.assertRaisesRegex(ValueError, "not authorized"):
|
|
validate_current_avatar_references(
|
|
upload_root=root,
|
|
avatar_urls=["/uploads/profile-avatars/unbound.png"],
|
|
initial_manifest=proof,
|
|
expected_database_target_sha256=DATABASE_TARGET_SHA256,
|
|
)
|
|
|
|
def test_manifest_v3_preserves_zero_ref_upload_and_binds_connected_db(self) -> None:
|
|
with tempfile.TemporaryDirectory(prefix="vignette-upload-preserved-v2-") as raw:
|
|
root = Path(raw) / "uploads"
|
|
avatar_root = root / "profile-avatars"
|
|
avatar_root.mkdir(parents=True)
|
|
referenced = avatar_root / "referenced.png"
|
|
pending = avatar_root / "pending.webp"
|
|
referenced.write_bytes(b"referenced")
|
|
pending.write_bytes(b"pending")
|
|
payload = build_privacy_safe_manifest(
|
|
upload_root=root,
|
|
references=[("profile-avatars/referenced.png", "profile_avatar")],
|
|
preserved_paths=[
|
|
"profile-avatars/referenced.png",
|
|
"profile-avatars/pending.webp",
|
|
],
|
|
database_target_sha256=DATABASE_TARGET_SHA256,
|
|
write_freeze_token_sha256=hashlib.sha256(b"freeze").hexdigest(),
|
|
)
|
|
self.assertEqual(
|
|
"vignette.public-avatar-upload-manifest.v3",
|
|
payload["schema_version"],
|
|
)
|
|
self.assertEqual(2, payload["preserved_object_count"])
|
|
self.assertEqual(0, payload["preserved_decode_valid_count"])
|
|
self.assertEqual(2, payload["preserved_decode_invalid_count"])
|
|
self.assertEqual(1, payload["required_decode_invalid_object_count"])
|
|
self.assertEqual(
|
|
1,
|
|
payload["required_decode_invalid_reference_count"],
|
|
)
|
|
self.assertEqual(
|
|
len(b"referenced") + len(b"pending"),
|
|
payload["preserved_total_size_bytes"],
|
|
)
|
|
self.assertEqual(1, payload["required_object_count"])
|
|
self.assertEqual(
|
|
[0, 1],
|
|
sorted(
|
|
int(record["reference_count"])
|
|
for record in payload["preserved_objects"]
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
1,
|
|
payload["database_reference_contract"]["reference_count"],
|
|
)
|
|
serialized = json.dumps(payload, sort_keys=True)
|
|
self.assertNotIn("referenced.png", serialized)
|
|
self.assertNotIn("pending.webp", serialized)
|
|
private = Path(raw) / "private"
|
|
private.mkdir()
|
|
manifest = private / "manifest.json"
|
|
manifest.write_text(serialized, encoding="utf-8")
|
|
manifest_sha = hashlib.sha256(manifest.read_bytes()).hexdigest()
|
|
proof = validate_upload_manifest(
|
|
upload_root=root,
|
|
manifest_path=manifest,
|
|
expected_manifest_sha256=manifest_sha,
|
|
expected_write_freeze_path=private / "freeze.json",
|
|
expected_database_target_sha256=DATABASE_TARGET_SHA256,
|
|
require_freeze_path_binding=False,
|
|
)
|
|
self.assertEqual(2, proof.preserved_object_count)
|
|
self.assertEqual(0, len(proof.preserved_records))
|
|
self.assertEqual(2, proof.preserved_decode_invalid_count)
|
|
self.assertEqual(
|
|
len(b"referenced") + len(b"pending"),
|
|
proof.preserved_total_size_bytes,
|
|
)
|
|
self.assertEqual(DATABASE_TARGET_SHA256, proof.database_target_sha256)
|
|
with self.assertRaisesRegex(ValueError, "database target binding"):
|
|
validate_upload_manifest(
|
|
upload_root=root,
|
|
manifest_path=manifest,
|
|
expected_manifest_sha256=manifest_sha,
|
|
expected_write_freeze_path=private / "freeze.json",
|
|
expected_database_target_sha256=hashlib.sha256(
|
|
b"other-database"
|
|
).hexdigest(),
|
|
require_freeze_path_binding=False,
|
|
)
|
|
|
|
(avatar_root / "unknown.txt").write_bytes(b"must-not-be-served")
|
|
with self.assertRaisesRegex(ValueError, "invalid extra"):
|
|
validate_upload_manifest(
|
|
upload_root=root,
|
|
manifest_path=manifest,
|
|
expected_manifest_sha256=manifest_sha,
|
|
expected_write_freeze_path=private / "freeze.json",
|
|
expected_database_target_sha256=DATABASE_TARGET_SHA256,
|
|
require_freeze_path_binding=False,
|
|
)
|
|
|
|
def test_manifest_hash_and_json_parse_share_one_byte_read(self) -> None:
|
|
source = Path(__file__).with_name("upload_storage.py").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
validator = source[
|
|
source.index("def validate_upload_manifest(") :
|
|
source.index("class CurrentAvatarReferenceProof")
|
|
]
|
|
self.assertIn("manifest_bytes = manifest.read_bytes()", validator)
|
|
self.assertIn("_sha256_bytes(manifest_bytes)", validator)
|
|
self.assertIn('json.loads(manifest_bytes.decode("utf-8"))', validator)
|
|
self.assertNotIn("manifest.read_text", validator)
|
|
|
|
def test_routine_manifest_validation_rehashes_preserved_and_allows_only_generated_extras(
|
|
self,
|
|
) -> None:
|
|
with tempfile.TemporaryDirectory(
|
|
prefix="vignette-upload-routine-validation-"
|
|
) as raw:
|
|
root = Path(raw) / "uploads"
|
|
avatar_root = root / "profile-avatars"
|
|
avatar_root.mkdir(parents=True)
|
|
preserved = avatar_root / "legacy-avatar.png"
|
|
preserved.write_bytes(_valid_png_bytes())
|
|
payload = build_privacy_safe_manifest(
|
|
upload_root=root,
|
|
references=[("profile-avatars/legacy-avatar.png", "profile_avatar")],
|
|
preserved_paths=["profile-avatars/legacy-avatar.png"],
|
|
database_target_sha256=DATABASE_TARGET_SHA256,
|
|
write_freeze_token_sha256=hashlib.sha256(b"freeze").hexdigest(),
|
|
)
|
|
private = Path(raw) / "private"
|
|
private.mkdir()
|
|
manifest = private / "manifest.json"
|
|
manifest.write_text(json.dumps(payload), encoding="utf-8")
|
|
manifest_sha256 = hashlib.sha256(manifest.read_bytes()).hexdigest()
|
|
generated_name = (
|
|
"8da1b8fc-3bdd-4e72-86f7-d17fc315c4cb-"
|
|
"AbCdEf_12345.webp"
|
|
)
|
|
(avatar_root / generated_name).write_bytes(b"new-runtime-upload")
|
|
|
|
proof = validate_upload_manifest(
|
|
upload_root=root,
|
|
manifest_path=manifest,
|
|
expected_manifest_sha256=manifest_sha256,
|
|
expected_write_freeze_path=private / "freeze.json",
|
|
expected_database_target_sha256=DATABASE_TARGET_SHA256,
|
|
require_freeze_path_binding=False,
|
|
verify_preserved_objects=True,
|
|
reject_unbound_extras=False,
|
|
)
|
|
self.assertEqual(1, len(proof.preserved_records))
|
|
self.assertTrue(
|
|
is_runtime_generated_public_avatar_relative_path(
|
|
f"profile-avatars/{generated_name}"
|
|
)
|
|
)
|
|
|
|
with self.assertRaisesRegex(ValueError, "unbound extra"):
|
|
validate_upload_manifest(
|
|
upload_root=root,
|
|
manifest_path=manifest,
|
|
expected_manifest_sha256=manifest_sha256,
|
|
expected_write_freeze_path=private / "freeze.json",
|
|
expected_database_target_sha256=DATABASE_TARGET_SHA256,
|
|
require_freeze_path_binding=False,
|
|
verify_preserved_objects=True,
|
|
reject_unbound_extras=True,
|
|
)
|
|
|
|
arbitrary = avatar_root / "looks-valid-but-unbound.png"
|
|
arbitrary.write_bytes(b"unauthorized")
|
|
with self.assertRaisesRegex(ValueError, "unauthorized extra"):
|
|
validate_upload_manifest(
|
|
upload_root=root,
|
|
manifest_path=manifest,
|
|
expected_manifest_sha256=manifest_sha256,
|
|
expected_write_freeze_path=private / "freeze.json",
|
|
expected_database_target_sha256=DATABASE_TARGET_SHA256,
|
|
require_freeze_path_binding=False,
|
|
verify_preserved_objects=True,
|
|
reject_unbound_extras=False,
|
|
)
|
|
arbitrary.unlink()
|
|
preserved.write_bytes(b"tampered")
|
|
with self.assertRaisesRegex(ValueError, "content hash"):
|
|
validate_upload_manifest(
|
|
upload_root=root,
|
|
manifest_path=manifest,
|
|
expected_manifest_sha256=manifest_sha256,
|
|
expected_write_freeze_path=private / "freeze.json",
|
|
expected_database_target_sha256=DATABASE_TARGET_SHA256,
|
|
require_freeze_path_binding=False,
|
|
verify_preserved_objects=True,
|
|
reject_unbound_extras=False,
|
|
)
|
|
|
|
def test_preserved_immutable_cache_has_a_fail_closed_total_size_limit(self) -> None:
|
|
from . import upload_storage
|
|
|
|
with tempfile.TemporaryDirectory(prefix="vignette-avatar-cache-cap-") as raw:
|
|
root = Path(raw) / "uploads"
|
|
avatar = root / "profile-avatars" / "legacy.png"
|
|
avatar.parent.mkdir(parents=True)
|
|
avatar.write_bytes(_valid_png_bytes())
|
|
payload = build_privacy_safe_manifest(
|
|
upload_root=root,
|
|
references=[("profile-avatars/legacy.png", "profile_avatar")],
|
|
preserved_paths=["profile-avatars/legacy.png"],
|
|
database_target_sha256=DATABASE_TARGET_SHA256,
|
|
write_freeze_token_sha256=hashlib.sha256(b"freeze").hexdigest(),
|
|
)
|
|
private = Path(raw) / "private"
|
|
private.mkdir()
|
|
manifest = private / "manifest.json"
|
|
manifest.write_text(json.dumps(payload), encoding="utf-8")
|
|
with (
|
|
patch.object(upload_storage, "MAX_PRESERVED_AVATAR_CACHE_BYTES", 1),
|
|
self.assertRaisesRegex(ValueError, "cache size limit"),
|
|
):
|
|
validate_upload_manifest(
|
|
upload_root=root,
|
|
manifest_path=manifest,
|
|
expected_manifest_sha256=hashlib.sha256(
|
|
manifest.read_bytes()
|
|
).hexdigest(),
|
|
expected_write_freeze_path=private / "freeze.json",
|
|
expected_database_target_sha256=DATABASE_TARGET_SHA256,
|
|
require_freeze_path_binding=False,
|
|
)
|
|
def test_public_avatar_normalizer_rejects_encoded_control_and_non_image_paths(
|
|
self,
|
|
) -> None:
|
|
for value in (
|
|
"profile-avatars/encoded%2fpath.png",
|
|
"profile-avatars/control\x01.png",
|
|
"profile-avatars/not-an-image.txt",
|
|
"profile-avatars/nested/file.png",
|
|
"profile-avatars/.hidden.png",
|
|
):
|
|
with self.subTest(value=value), self.assertRaises(ValueError):
|
|
normalize_public_avatar_relative_path(value)
|
|
|
|
def test_current_inventory_cannot_be_empty_after_nonempty_initialization(
|
|
self,
|
|
) -> None:
|
|
with tempfile.TemporaryDirectory(
|
|
prefix="vignette-upload-empty-current-"
|
|
) as raw:
|
|
root = Path(raw) / "uploads"
|
|
object_path = root / "profile-avatars" / "initial.png"
|
|
object_path.parent.mkdir(parents=True)
|
|
object_path.write_bytes(b"initial")
|
|
payload = build_privacy_safe_manifest(
|
|
upload_root=root,
|
|
references=[("profile-avatars/initial.png", "profile_avatar")],
|
|
preserved_paths=["profile-avatars/initial.png"],
|
|
database_target_sha256=DATABASE_TARGET_SHA256,
|
|
write_freeze_token_sha256=hashlib.sha256(b"token").hexdigest(),
|
|
)
|
|
private = Path(raw) / "private"
|
|
private.mkdir()
|
|
manifest = private / "manifest.json"
|
|
manifest.write_text(json.dumps(payload), encoding="utf-8")
|
|
proof = validate_upload_manifest(
|
|
upload_root=root,
|
|
manifest_path=manifest,
|
|
expected_manifest_sha256=hashlib.sha256(
|
|
manifest.read_bytes()
|
|
).hexdigest(),
|
|
expected_write_freeze_path=Path(raw) / "freeze.json",
|
|
expected_database_target_sha256=DATABASE_TARGET_SHA256,
|
|
require_freeze_path_binding=False,
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "explicit reset receipt"):
|
|
validate_current_avatar_references(
|
|
upload_root=root,
|
|
avatar_urls=[],
|
|
initial_manifest=proof,
|
|
expected_database_target_sha256=DATABASE_TARGET_SHA256,
|
|
)
|
|
|
|
def test_manifest_and_freeze_state_are_rejected_inside_public_root(self) -> None:
|
|
with tempfile.TemporaryDirectory(
|
|
prefix="vignette-upload-private-state-"
|
|
) as raw:
|
|
root = Path(raw) / "uploads"
|
|
object_path = root / "profile-avatars" / "avatar.png"
|
|
object_path.parent.mkdir(parents=True)
|
|
object_path.write_bytes(b"avatar")
|
|
token_sha = hashlib.sha256(b"token").hexdigest()
|
|
with self.assertRaisesRegex(ValueError, "outside the public upload root"):
|
|
build_privacy_safe_manifest(
|
|
upload_root=root,
|
|
references=[("profile-avatars/avatar.png", "profile_avatar")],
|
|
preserved_paths=["profile-avatars/avatar.png"],
|
|
database_target_sha256=DATABASE_TARGET_SHA256,
|
|
write_freeze_token_sha256=token_sha,
|
|
write_freeze_path=root / ".private-freeze.json",
|
|
)
|
|
|
|
def test_static_mount_exposes_only_profile_avatar_directory(self) -> None:
|
|
source = Path(__file__).with_name("main.py").read_text(encoding="utf-8")
|
|
runtime_source = Path(__file__).with_name("upload_runtime.py").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
self.assertIn('app.mount(\n "/uploads/profile-avatars"', source)
|
|
self.assertIn("FlatPublicAvatarStaticFiles(directory=", source)
|
|
self.assertNotIn("\n StaticFiles(directory=", source)
|
|
self.assertNotIn('app.mount("/uploads",', source)
|
|
self.assertNotIn('name="uploads"', source)
|
|
self.assertIn(
|
|
"assert_path_without_reparse(\n public_avatar_root,",
|
|
runtime_source,
|
|
)
|
|
|
|
def test_public_avatar_root_reparse_is_rejected(self) -> None:
|
|
with tempfile.TemporaryDirectory(prefix="vignette-avatar-reparse-") as raw:
|
|
root = Path(raw) / "uploads"
|
|
avatar_root = root / "profile-avatars"
|
|
avatar_root.mkdir(parents=True)
|
|
with patch.object(
|
|
upload_storage_module,
|
|
"_is_reparse_or_symlink",
|
|
side_effect=lambda candidate: candidate == avatar_root,
|
|
):
|
|
with self.assertRaisesRegex(ValueError, "reparse point"):
|
|
assert_path_without_reparse(
|
|
avatar_root,
|
|
"initialized public avatar directory",
|
|
)
|
|
|
|
def test_write_gate_blocks_new_writes_and_counts_inflight_leases(self) -> None:
|
|
with tempfile.TemporaryDirectory(prefix="vignette-upload-freeze-") as raw:
|
|
freeze = Path(raw) / "upload-write.freeze"
|
|
gate = UploadWriteFreezeGate(freeze)
|
|
with gate.write_lease():
|
|
self.assertEqual(1, gate.status().in_flight)
|
|
self.assertEqual(0, gate.status().in_flight)
|
|
freeze.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": "vignette.public-upload-write-freeze.v1",
|
|
"token": "opaque-token-0123456789abcdef0123456789",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaises(UploadWriteFrozen):
|
|
with gate.write_lease():
|
|
pass
|
|
status = gate.status()
|
|
self.assertTrue(status.active)
|
|
self.assertTrue(status.valid)
|
|
self.assertEqual(0, status.in_flight)
|
|
|
|
def test_write_gate_status_reads_sentinel_and_counter_under_one_lock(self) -> None:
|
|
with tempfile.TemporaryDirectory(prefix="vignette-upload-freeze-race-") as raw:
|
|
freeze = Path(raw) / "upload-write.freeze"
|
|
gate = UploadWriteFreezeGate(freeze)
|
|
status_read_started = threading.Event()
|
|
allow_status_read = threading.Event()
|
|
lease_entered = threading.Event()
|
|
statuses = []
|
|
def blocking_entry_probe(candidate: Path) -> bool:
|
|
if candidate == freeze:
|
|
status_read_started.set()
|
|
if not allow_status_read.wait(2):
|
|
raise AssertionError("status sentinel read did not resume")
|
|
return False
|
|
return candidate.exists()
|
|
|
|
def read_status() -> None:
|
|
statuses.append(gate.status())
|
|
|
|
def enter_lease() -> None:
|
|
with gate.write_lease():
|
|
lease_entered.set()
|
|
|
|
with patch.object(
|
|
upload_storage_module,
|
|
"_path_entry_exists_without_following",
|
|
side_effect=blocking_entry_probe,
|
|
):
|
|
status_thread = threading.Thread(target=read_status)
|
|
status_thread.start()
|
|
self.assertTrue(status_read_started.wait(1))
|
|
lease_thread = threading.Thread(target=enter_lease)
|
|
lease_thread.start()
|
|
self.assertFalse(lease_entered.wait(0.1))
|
|
allow_status_read.set()
|
|
status_thread.join(2)
|
|
lease_thread.join(2)
|
|
|
|
self.assertFalse(status_thread.is_alive())
|
|
self.assertFalse(lease_thread.is_alive())
|
|
self.assertEqual(1, len(statuses))
|
|
self.assertEqual(0, statuses[0].in_flight)
|
|
self.assertTrue(lease_entered.is_set())
|
|
|
|
def test_unreadable_or_dangling_freeze_entry_blocks_writes(self) -> None:
|
|
with tempfile.TemporaryDirectory(prefix="vignette-upload-freeze-entry-") as raw:
|
|
freeze = Path(raw) / "dangling-freeze"
|
|
gate = UploadWriteFreezeGate(freeze)
|
|
with patch.object(
|
|
upload_storage_module,
|
|
"_path_entry_exists_without_following",
|
|
return_value=True,
|
|
):
|
|
with self.assertRaises(UploadWriteFrozen):
|
|
with gate.write_lease():
|
|
pass
|
|
status = gate.status()
|
|
self.assertTrue(status.active)
|
|
self.assertFalse(status.valid)
|
|
|
|
|
|
class RuntimeUploadDatabaseContractTest(unittest.IsolatedAsyncioTestCase):
|
|
@staticmethod
|
|
def _manifest_proof(raw: str):
|
|
root = Path(raw) / "uploads"
|
|
avatar = root / "profile-avatars" / "initial.png"
|
|
avatar.parent.mkdir(parents=True)
|
|
avatar.write_bytes(b"initial")
|
|
target_sha256 = database_target_sha256(
|
|
database_name="vignette",
|
|
database_role="vignette_app",
|
|
server_address="10.0.0.5",
|
|
server_port=5432,
|
|
)
|
|
payload = build_privacy_safe_manifest(
|
|
upload_root=root,
|
|
references=[("profile-avatars/initial.png", "profile_avatar")],
|
|
preserved_paths=["profile-avatars/initial.png"],
|
|
database_target_sha256=target_sha256,
|
|
write_freeze_token_sha256=hashlib.sha256(b"freeze").hexdigest(),
|
|
)
|
|
private = Path(raw) / "private"
|
|
private.mkdir()
|
|
manifest = private / "manifest.json"
|
|
manifest.write_text(json.dumps(payload), encoding="utf-8")
|
|
proof = validate_upload_manifest(
|
|
upload_root=root,
|
|
manifest_path=manifest,
|
|
expected_manifest_sha256=hashlib.sha256(manifest.read_bytes()).hexdigest(),
|
|
expected_write_freeze_path=private / "freeze.json",
|
|
expected_database_target_sha256=target_sha256,
|
|
require_freeze_path_binding=False,
|
|
)
|
|
return root, proof, target_sha256
|
|
|
|
async def test_lifespan_database_proof_uses_one_repeatable_read_pool_snapshot(
|
|
self,
|
|
) -> None:
|
|
from . import upload_runtime
|
|
|
|
with tempfile.TemporaryDirectory(prefix="vignette-runtime-db-proof-") as raw:
|
|
root, proof, target_sha256 = self._manifest_proof(raw)
|
|
connection = _RuntimeDatabaseConnection(
|
|
avatar_urls=["/uploads/profile-avatars/initial.png"]
|
|
)
|
|
pool = _RuntimeDatabasePool(connection)
|
|
with (
|
|
patch.object(upload_runtime, "upload_root", root),
|
|
patch.object(upload_runtime, "upload_manifest_proof", proof),
|
|
patch.object(
|
|
upload_runtime.settings,
|
|
"public_runtime_db_target_sha256",
|
|
target_sha256,
|
|
),
|
|
):
|
|
runtime_proof = (
|
|
await upload_runtime.validate_runtime_upload_database_state(pool)
|
|
)
|
|
|
|
self.assertEqual(1, pool.acquire_count)
|
|
self.assertEqual(
|
|
[{"isolation": "repeatable_read", "readonly": True}],
|
|
connection.transaction_options,
|
|
)
|
|
self.assertEqual(target_sha256, runtime_proof.database_target_sha256)
|
|
self.assertEqual(1, runtime_proof.current_references.object_count)
|
|
self.assertEqual(1, runtime_proof.current_references.reference_count)
|
|
self.assertTrue(connection.executed)
|
|
|
|
main_source = Path(__file__).with_name("main.py").read_text(encoding="utf-8")
|
|
init_index = main_source.index("await init_pool()")
|
|
proof_index = main_source.index("await validate_runtime_upload_database_state")
|
|
first_mutation_index = main_source.index("await ensure_runtime_tables()")
|
|
self.assertLess(init_index, proof_index)
|
|
self.assertLess(proof_index, first_mutation_index)
|
|
|
|
async def test_lifespan_database_proof_rejects_connected_target_drift(self) -> None:
|
|
from . import upload_runtime
|
|
|
|
with tempfile.TemporaryDirectory(prefix="vignette-runtime-db-drift-") as raw:
|
|
root, proof, target_sha256 = self._manifest_proof(raw)
|
|
connection = _RuntimeDatabaseConnection(
|
|
avatar_urls=["/uploads/profile-avatars/initial.png"],
|
|
database_name="other_database",
|
|
)
|
|
pool = _RuntimeDatabasePool(connection)
|
|
with (
|
|
patch.object(upload_runtime, "upload_root", root),
|
|
patch.object(upload_runtime, "upload_manifest_proof", proof),
|
|
patch.object(
|
|
upload_runtime.settings,
|
|
"public_runtime_db_target_sha256",
|
|
target_sha256,
|
|
),
|
|
):
|
|
with self.assertRaisesRegex(RuntimeError, "database target drift"):
|
|
await upload_runtime.validate_runtime_upload_database_state(pool)
|
|
|
|
async def test_every_new_pool_connection_is_bound_before_codec_setup(self) -> None:
|
|
from . import db
|
|
|
|
target_sha256 = database_target_sha256(
|
|
database_name="vignette",
|
|
database_role="vignette_app",
|
|
server_address="10.0.0.5",
|
|
server_port=5432,
|
|
)
|
|
connection = _RuntimeDatabaseConnection(avatar_urls=[])
|
|
with (
|
|
patch.object(db.settings, "user_upload_manifest_required", True),
|
|
patch.object(
|
|
db.settings,
|
|
"public_runtime_db_target_sha256",
|
|
target_sha256,
|
|
),
|
|
):
|
|
await db._init_connection(connection)
|
|
self.assertEqual(["jsonb", "json"], connection.codec_names)
|
|
|
|
drifted = _RuntimeDatabaseConnection(
|
|
avatar_urls=[], database_name="other_database"
|
|
)
|
|
with (
|
|
patch.object(db.settings, "user_upload_manifest_required", True),
|
|
patch.object(
|
|
db.settings,
|
|
"public_runtime_db_target_sha256",
|
|
target_sha256,
|
|
),
|
|
):
|
|
with self.assertRaisesRegex(RuntimeError, "database target drift"):
|
|
await db._init_connection(drifted)
|
|
self.assertEqual([], drifted.codec_names)
|
|
|
|
|
|
class AvatarWriteSafetyTest(unittest.IsolatedAsyncioTestCase):
|
|
@staticmethod
|
|
def _profile(users, *, avatar_url: str):
|
|
return users.UserProfileResponse(
|
|
user_id="user-1",
|
|
email="user@example.test",
|
|
display_name="사용자",
|
|
role="learner",
|
|
cohort_ids=[],
|
|
affiliation="한신대학교",
|
|
avatar_url=avatar_url,
|
|
)
|
|
|
|
async def test_static_avatar_handler_rejects_nested_hidden_and_unknown_files(
|
|
self,
|
|
) -> None:
|
|
from starlette.exceptions import HTTPException as StarletteHTTPException
|
|
|
|
from . import upload_runtime
|
|
|
|
with tempfile.TemporaryDirectory(prefix="vignette-avatar-static-") as raw:
|
|
upload_root = Path(raw) / "uploads"
|
|
root = upload_root / "profile-avatars"
|
|
root.mkdir(parents=True)
|
|
generated_name = (
|
|
"8da1b8fc-3bdd-4e72-86f7-d17fc315c4cb-"
|
|
"AbCdEf_12345.png"
|
|
)
|
|
valid = root / generated_name
|
|
valid_bytes = _valid_png_bytes()
|
|
valid.write_bytes(valid_bytes)
|
|
arbitrary = root / "avatar.png"
|
|
arbitrary.write_bytes(b"unbound")
|
|
nested = root / "nested"
|
|
nested.mkdir()
|
|
(nested / "avatar.png").write_bytes(b"nested")
|
|
(root / ".hidden.png").write_bytes(b"hidden")
|
|
(root / "notes.txt").write_bytes(b"private")
|
|
files = upload_runtime.FlatPublicAvatarStaticFiles(directory=str(root))
|
|
scope = {"method": "GET", "headers": []}
|
|
with patch.object(upload_runtime, "upload_manifest_proof", None):
|
|
response = await files.get_response(generated_name, scope)
|
|
self.assertEqual(200, response.status_code)
|
|
self.assertEqual(valid_bytes, response.body)
|
|
self.assertEqual(
|
|
f'"{hashlib.sha256(valid_bytes).hexdigest()}"',
|
|
response.headers["etag"],
|
|
)
|
|
with patch.object(upload_runtime, "upload_manifest_proof", None):
|
|
head = await files.get_response(
|
|
generated_name,
|
|
{"method": "HEAD", "headers": []},
|
|
)
|
|
self.assertEqual(b"", head.body)
|
|
self.assertEqual(str(len(valid_bytes)), head.headers["content-length"])
|
|
with (
|
|
patch.object(upload_runtime, "upload_root", upload_root),
|
|
patch.object(upload_runtime, "upload_manifest_proof", None),
|
|
):
|
|
upload_runtime.assert_public_avatar_url_exists(
|
|
f"/uploads/profile-avatars/{generated_name}"
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "not authorized"):
|
|
upload_runtime.assert_public_avatar_url_exists(
|
|
"/uploads/profile-avatars/avatar.png"
|
|
)
|
|
|
|
valid.write_bytes(b"damaged-after-upload")
|
|
with patch.object(upload_runtime, "upload_manifest_proof", None):
|
|
with self.assertRaises(StarletteHTTPException) as damaged:
|
|
await files.get_response(generated_name, scope)
|
|
self.assertEqual(404, damaged.exception.status_code)
|
|
|
|
corrupt_name = (
|
|
"1265087b-d3d0-4e85-82bc-499d993a3f89-"
|
|
"QrStUv_67890.png"
|
|
)
|
|
(root / corrupt_name).write_bytes(b"not-a-png")
|
|
with patch.object(upload_runtime, "upload_manifest_proof", None):
|
|
with self.assertRaises(StarletteHTTPException) as corrupt:
|
|
await files.get_response(corrupt_name, scope)
|
|
self.assertEqual(404, corrupt.exception.status_code)
|
|
|
|
mismatch_name = (
|
|
"02f01847-1d0c-48c8-9559-4ba791d00275-"
|
|
"MnOpQr_24680.png"
|
|
)
|
|
mismatch_buffer = io.BytesIO()
|
|
Image.new("RGB", (1, 1), (255, 255, 255)).save(
|
|
mismatch_buffer,
|
|
format="JPEG",
|
|
)
|
|
(root / mismatch_name).write_bytes(mismatch_buffer.getvalue())
|
|
with patch.object(upload_runtime, "upload_manifest_proof", None):
|
|
with self.assertRaises(StarletteHTTPException) as mismatch:
|
|
await files.get_response(mismatch_name, scope)
|
|
self.assertEqual(404, mismatch.exception.status_code)
|
|
for invalid in (
|
|
"avatar.png",
|
|
"nested/avatar.png",
|
|
".hidden.png",
|
|
"notes.txt",
|
|
):
|
|
with self.subTest(path=invalid):
|
|
with patch.object(upload_runtime, "upload_manifest_proof", None):
|
|
with self.assertRaises(StarletteHTTPException) as raised:
|
|
await files.get_response(invalid, scope)
|
|
self.assertEqual(404, raised.exception.status_code)
|
|
|
|
legacy_name = "legacy-avatar.png"
|
|
legacy = root / legacy_name
|
|
legacy.write_bytes(b"legacy")
|
|
path_sha256 = public_avatar_relative_path_sha256(
|
|
f"profile-avatars/{legacy_name}"
|
|
)
|
|
proof = SimpleNamespace(
|
|
preserved_records=(
|
|
(
|
|
path_sha256,
|
|
hashlib.sha256(b"legacy").hexdigest(),
|
|
len(b"legacy"),
|
|
"image/png",
|
|
b"legacy",
|
|
),
|
|
),
|
|
preserved_path_decode_records=((path_sha256, True),),
|
|
)
|
|
with patch.object(upload_runtime, "upload_manifest_proof", proof):
|
|
response = await files.get_response(legacy_name, scope)
|
|
self.assertEqual(200, response.status_code)
|
|
self.assertEqual(b"legacy", response.body)
|
|
self.assertEqual("6", response.headers["content-length"])
|
|
self.assertEqual("image/png", response.headers["content-type"])
|
|
self.assertEqual(
|
|
f'"{hashlib.sha256(b"legacy").hexdigest()}"',
|
|
response.headers["etag"],
|
|
)
|
|
legacy.write_bytes(b"drift")
|
|
cached = await files.get_response(legacy_name, scope)
|
|
self.assertEqual(b"legacy", cached.body)
|
|
head = await files.get_response(
|
|
legacy_name,
|
|
{"method": "HEAD", "headers": []},
|
|
)
|
|
self.assertEqual(b"", head.body)
|
|
self.assertEqual("6", head.headers["content-length"])
|
|
|
|
forensic_name = "forensic-avatar.png"
|
|
forensic_path_sha256 = public_avatar_relative_path_sha256(
|
|
f"profile-avatars/{forensic_name}"
|
|
)
|
|
(root / forensic_name).write_bytes(b"\x89PNG\r\n\x1a\ncorrupt")
|
|
forensic_proof = SimpleNamespace(
|
|
preserved_records=(),
|
|
preserved_path_decode_records=(
|
|
(forensic_path_sha256, False),
|
|
),
|
|
)
|
|
with patch.object(
|
|
upload_runtime,
|
|
"upload_manifest_proof",
|
|
forensic_proof,
|
|
):
|
|
with self.assertRaises(StarletteHTTPException) as raised:
|
|
await files.get_response(forensic_name, scope)
|
|
self.assertEqual(404, raised.exception.status_code)
|
|
|
|
async def test_upload_root_is_revalidated_immediately_before_write(self) -> None:
|
|
from fastapi import HTTPException
|
|
|
|
from .routes import users
|
|
|
|
with tempfile.TemporaryDirectory(prefix="vignette-avatar-root-drift-") as raw:
|
|
root = Path(raw)
|
|
with (
|
|
patch.object(users, "public_avatar_root", root),
|
|
patch.object(
|
|
users,
|
|
"assert_path_without_reparse",
|
|
side_effect=ValueError("reparse point"),
|
|
),
|
|
):
|
|
with self.assertRaises(HTTPException) as raised:
|
|
users._upload_root()
|
|
self.assertEqual(503, raised.exception.status_code)
|
|
|
|
target = root / "avatar.png"
|
|
with patch.object(
|
|
users,
|
|
"assert_path_without_reparse",
|
|
side_effect=ValueError("reparse point"),
|
|
):
|
|
with self.assertRaisesRegex(ValueError, "reparse point"):
|
|
users._write_new_avatar_atomically(target, b"avatar")
|
|
self.assertFalse(target.exists())
|
|
|
|
async def test_unrelated_profile_patch_does_not_restore_stale_avatar(self) -> None:
|
|
from .deps import Principal, Role
|
|
from .routes import users
|
|
|
|
principal = Principal(
|
|
user_id="user-1",
|
|
role=Role.LEARNER,
|
|
email="user@example.test",
|
|
display_name="사용자",
|
|
)
|
|
stale_profile = self._profile(
|
|
users, avatar_url="/uploads/profile-avatars/stale.png"
|
|
)
|
|
update = AsyncMock(return_value=object())
|
|
with (
|
|
patch.object(users, "_profile_for", AsyncMock(return_value=stale_profile)),
|
|
patch.object(users, "update_managed_user", update),
|
|
patch.object(
|
|
users,
|
|
"_profile_avatar_write_lease",
|
|
side_effect=AssertionError("avatar lease must not be needed"),
|
|
),
|
|
):
|
|
await users.patch_me(users.UserProfilePatch(display_name="새 이름"), principal)
|
|
|
|
managed_patch = update.await_args.args[1]
|
|
self.assertIsNone(managed_patch.avatar_url)
|
|
|
|
async def test_explicit_same_avatar_patch_still_uses_write_lease(self) -> None:
|
|
from .deps import Principal, Role
|
|
from .routes import users
|
|
|
|
principal = Principal(
|
|
user_id="user-1",
|
|
role=Role.LEARNER,
|
|
email="user@example.test",
|
|
display_name="사용자",
|
|
)
|
|
avatar_url = "/uploads/profile-avatars/current.png"
|
|
profile = self._profile(users, avatar_url=avatar_url)
|
|
lease = patch.object(
|
|
users, "_profile_avatar_write_lease", return_value=nullcontext()
|
|
)
|
|
with (
|
|
patch.object(users, "_profile_for", AsyncMock(return_value=profile)),
|
|
patch.object(users, "update_managed_user", AsyncMock(return_value=object())),
|
|
patch.object(users, "assert_public_avatar_url_exists", return_value=None),
|
|
lease as lease_mock,
|
|
):
|
|
await users.patch_me(users.UserProfilePatch(avatar_url=avatar_url), principal)
|
|
|
|
lease_mock.assert_called_once_with()
|
|
|
|
async def test_onboarding_always_participates_in_avatar_write_freeze(self) -> None:
|
|
from .deps import Principal, Role
|
|
from .routes import users
|
|
|
|
principal = Principal(
|
|
user_id="user-1",
|
|
role=Role.LEARNER,
|
|
email="user@example.test",
|
|
display_name="사용자",
|
|
consent_at=1.0,
|
|
)
|
|
profile = self._profile(users, avatar_url="")
|
|
updated = SimpleNamespace(
|
|
display_name="학습자", profile_completed_at=1.0
|
|
)
|
|
lease = patch.object(
|
|
users, "_profile_avatar_write_lease", return_value=nullcontext()
|
|
)
|
|
body = users.OnboardingRequest(
|
|
legal_name="학습자",
|
|
affiliation="한신대학교",
|
|
department="상담심리학과",
|
|
grade_level="3학년",
|
|
phone="010-1234-5678",
|
|
contact_address="경기도 오산시",
|
|
nickname="학습자",
|
|
self_introduction="소개",
|
|
avatar_url="",
|
|
terms_accepted=True,
|
|
privacy_accepted=True,
|
|
)
|
|
with (
|
|
patch.object(users, "_profile_for", AsyncMock(return_value=profile)),
|
|
patch.object(users, "update_managed_user", AsyncMock(return_value=updated)),
|
|
lease as lease_mock,
|
|
):
|
|
await users.complete_onboarding(body, principal)
|
|
|
|
lease_mock.assert_called_once_with()
|
|
|
|
async def test_patch_failure_preserves_old_db_referenced_avatar(self) -> None:
|
|
from .deps import Principal, Role
|
|
from .routes import users
|
|
|
|
with tempfile.TemporaryDirectory(prefix="vignette-avatar-write-") as raw:
|
|
root = Path(raw)
|
|
old = root / "user-1-old.png"
|
|
old.write_bytes(b"old-avatar")
|
|
user_id = "8da1b8fc-3bdd-4e72-86f7-d17fc315c4cb"
|
|
principal = Principal(
|
|
user_id=user_id,
|
|
role=Role.LEARNER,
|
|
email="user@example.test",
|
|
display_name="사용자",
|
|
)
|
|
upload = UploadFile(
|
|
io.BytesIO(_valid_png_bytes()),
|
|
filename="avatar.png",
|
|
headers=Headers({"content-type": "image/png"}),
|
|
)
|
|
with patch.object(users, "_upload_root", return_value=root):
|
|
uploaded = await users.upload_my_avatar(principal, upload)
|
|
|
|
new_name = uploaded.avatar_url.rsplit("/", 1)[-1]
|
|
new_path = root / new_name
|
|
self.assertTrue(new_path.is_file())
|
|
self.assertEqual(b"old-avatar", old.read_bytes())
|
|
|
|
profile = users.UserProfileResponse(
|
|
user_id="user-1",
|
|
email="user@example.test",
|
|
display_name="사용자",
|
|
role="learner",
|
|
cohort_ids=[],
|
|
affiliation="한신대학교",
|
|
avatar_url="/uploads/profile-avatars/user-1-old.png",
|
|
)
|
|
with (
|
|
patch.object(users, "_profile_for", AsyncMock(return_value=profile)),
|
|
patch.object(
|
|
users,
|
|
"update_managed_user",
|
|
AsyncMock(side_effect=RuntimeError("DB write failed")),
|
|
),
|
|
patch.object(
|
|
users, "assert_public_avatar_url_exists", return_value=None
|
|
),
|
|
):
|
|
with self.assertRaisesRegex(RuntimeError, "DB write failed"):
|
|
await users.patch_me(
|
|
users.UserProfilePatch(avatar_url=uploaded.avatar_url),
|
|
principal,
|
|
)
|
|
|
|
self.assertEqual(b"old-avatar", old.read_bytes())
|
|
self.assertTrue(new_path.is_file())
|
|
|
|
async def test_avatar_upload_rejects_non_uuid_principal_before_file_creation(
|
|
self,
|
|
) -> None:
|
|
from fastapi import HTTPException
|
|
|
|
from .deps import Principal, Role
|
|
from .routes import users
|
|
|
|
with tempfile.TemporaryDirectory(prefix="vignette-avatar-user-id-") as raw:
|
|
root = Path(raw)
|
|
principal = Principal(
|
|
user_id="not-a-uuid",
|
|
role=Role.LEARNER,
|
|
email="user@example.test",
|
|
display_name="사용자",
|
|
)
|
|
upload = UploadFile(
|
|
io.BytesIO(_valid_png_bytes()),
|
|
filename="avatar.png",
|
|
headers=Headers({"content-type": "image/png"}),
|
|
)
|
|
with patch.object(users, "_upload_root", return_value=root):
|
|
with self.assertRaises(HTTPException) as raised:
|
|
await users.upload_my_avatar(principal, upload)
|
|
self.assertEqual(503, raised.exception.status_code)
|
|
self.assertEqual([], list(root.iterdir()))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|