From 058570c11adc45b2af2b553fe18507bdd5d4492b Mon Sep 17 00:00:00 2001 From: Simon Pinfold Date: Wed, 26 Aug 2026 03:23:37 -0700 Subject: [PATCH] fix(assets): store blake3-prefixed hashes; unify every read and comparison (D7) --- app/assets/api/routes.py | 6 +- app/assets/helpers.py | 5 + app/assets/scanner.py | 23 +- app/assets/scanner_changes.py | 12 +- app/assets/services/asset_management.py | 11 +- app/assets/services/hash_mode_state.py | 9 +- app/assets/services/ingest.py | 31 +- .../services/test_detection_gate.py | 15 +- .../services/test_enrichment_snapshot.py | 3 +- .../assets_test/services/test_from_hash.py | 5 +- .../services/test_recovery_gate.py | 16 +- .../assets_test/services/test_scanner_b.py | 5 +- .../assets_test/services/test_serving_b.py | 15 +- .../services/test_stored_hash_prefix.py | 281 ++++++++++++++++++ .../services/test_transition_drain.py | 15 +- .../assets_test/test_upload_hashing_modes.py | 27 +- 16 files changed, 401 insertions(+), 78 deletions(-) create mode 100644 tests-unit/assets_test/services/test_stored_hash_prefix.py diff --git a/app/assets/api/routes.py b/app/assets/api/routes.py index 08b243d33..884ff505a 100644 --- a/app/assets/api/routes.py +++ b/app/assets/api/routes.py @@ -351,14 +351,10 @@ def _build_record_response( else: preview_url = None - content_hash = content.hash - if content_hash is not None and not content_hash.startswith("blake3:"): - content_hash = f"blake3:{content_hash}" - return schemas_out.Asset( id=record.id, name=record.name, - hash=content_hash, + hash=content.hash, loader_path=record.loader_path, size=content.size_bytes, mime_type=record.mime_type, diff --git a/app/assets/helpers.py b/app/assets/helpers.py index 6cf417604..2e964162b 100644 --- a/app/assets/helpers.py +++ b/app/assets/helpers.py @@ -47,6 +47,11 @@ def normalize_tags(tags: list[str] | None) -> list[str]: return list(dict.fromkeys(t.strip() for t in (tags or []) if (t or "").strip())) +def to_stored_hash(digest: str) -> str: + """Convert a bare BLAKE3 digest to its stored form.""" + return f"blake3:{digest}" + + def validate_blake3_hash(s: str) -> str: """Validate and normalize a blake3 hash string. diff --git a/app/assets/scanner.py b/app/assets/scanner.py index 7eba50bfd..ce96c29b8 100644 --- a/app/assets/scanner.py +++ b/app/assets/scanner.py @@ -14,6 +14,7 @@ from app.assets.database.queries import ( create_record, ) from app.assets.database.models import Asset, AssetContent +from app.assets.helpers import to_stored_hash from app.assets.scanner_changes import ( clear_pending_verifications, detect_content_change, @@ -24,12 +25,12 @@ from app.assets.scanner_changes import ( recover_missing_content, ) from app.assets.scanner_admission import ( - PARTIAL_DOWNLOAD_EXTENSIONS, - _WATCH_LIST, - _WatchEntry, + PARTIAL_DOWNLOAD_EXTENSIONS as PARTIAL_DOWNLOAD_EXTENSIONS, + _WATCH_LIST as _WATCH_LIST, + _WatchEntry as _WatchEntry, _should_skip_extension, _two_stat_admit, - tick_watch_list, + tick_watch_list as tick_watch_list, ) from app.assets.services.bulk_ingest import SeedAssetSpec from app.assets.services.file_utils import get_mtime_ns, is_visible, list_files_recursively @@ -449,16 +450,18 @@ def enrich_asset( if metadata: mime_type = metadata.content_type - full_hash: str | None = None + digest: str | None = None + stored_hash: str | None = None if compute_hash: try: - full_hash = snapshot_hash(file_path) - if full_hash is None: + digest = snapshot_hash(file_path) + if digest is None: logging.warning( "File modified during hashing (snapshot unstable), discarding hash: %s", file_path, ) return False + stored_hash = to_stored_hash(digest) except Exception as e: logging.warning("Failed to hash %s: %s", file_path, e) @@ -483,14 +486,14 @@ def enrich_asset( system_metadata.update(dims) record.system_metadata = {**(record.system_metadata or {}), **system_metadata} - if full_hash: - content.hash = full_hash + if stored_hash: + content.hash = stored_hash if mime_type: record.mime_type = mime_type session.commit() - return full_hash is not None or metadata is not None or mime_type is not None + return stored_hash is not None or metadata is not None or mime_type is not None def enrich_assets_batch( diff --git a/app/assets/scanner_changes.py b/app/assets/scanner_changes.py index 02e3896ac..dd1d7737b 100644 --- a/app/assets/scanner_changes.py +++ b/app/assets/scanner_changes.py @@ -16,6 +16,7 @@ from app.assets.database.queries.records import ( mark_content_missing, unset_content_missing, ) +from app.assets.helpers import to_stored_hash from app.assets.services.path_utils import compute_loader_path, get_name_and_tags_from_asset_path from app.assets.services.snapshot_hash import snapshot_hash @@ -48,12 +49,13 @@ def recover_missing_content( if path not in _pending_recovery_paths: _pending_recovery_paths.append(path) return "unstable" + stored_hash = to_stored_hash(digest) matches = list( session.scalars( sa.select(AssetContent).where( AssetContent.path == path, AssetContent.is_missing.is_(True), - AssetContent.hash == digest, + AssetContent.hash == stored_hash, ) ) ) @@ -130,14 +132,14 @@ def drain_pending_verifications(session: Session, limit: int | None = None) -> i if digest is None: queue_pending_verification(content_id) continue + stored_hash = to_stored_hash(digest) - current_hash = digest - if content.hash == current_hash or content.hash is None: - content.hash = current_hash + if content.hash == stored_hash or content.hash is None: + content.hash = stored_hash content.size_bytes = stat_result.st_size content.mtime_ns = stat_result.st_mtime_ns else: - split_content(session, content, stat_result, hash_value=current_hash) + split_content(session, content, stat_result, hash_value=stored_hash) processed += 1 return processed diff --git a/app/assets/services/asset_management.py b/app/assets/services/asset_management.py index fedbf3a4a..741266997 100644 --- a/app/assets/services/asset_management.py +++ b/app/assets/services/asset_management.py @@ -202,9 +202,8 @@ def asset_exists(asset_hash: str) -> bool: canonical = validate_blake3_hash(asset_hash) except ValueError: return False - digest = canonical.partition(":")[2] with create_session() as session: - return lookup_for_view(session, digest) is not None + return lookup_for_view(session, canonical) is not None def get_asset_by_hash(asset_hash: str) -> AssetData | None: @@ -347,12 +346,16 @@ def resolve_hash_to_path( from sqlalchemy import select from app.assets.database.models import Asset + from app.assets.helpers import validate_blake3_hash from app.assets.services.lookup import lookup_for_view del tenant_id - digest = asset_hash.partition(":")[2] or asset_hash + try: + canonical = validate_blake3_hash(asset_hash) + except ValueError: + return None with create_session() as session: - content = lookup_for_view(session, digest) + content = lookup_for_view(session, canonical) if content is None: return None diff --git a/app/assets/services/hash_mode_state.py b/app/assets/services/hash_mode_state.py index 308afae3e..4f546c614 100644 --- a/app/assets/services/hash_mode_state.py +++ b/app/assets/services/hash_mode_state.py @@ -10,6 +10,7 @@ from sqlalchemy.orm import Session from app.assets import mode as _mode from app.assets.database.models import AssetContent, AssetSystemState from app.assets.database.queries.records import create_content, create_record, mark_content_missing +from app.assets.helpers import to_stored_hash from app.assets.services.path_utils import compute_loader_path, get_name_and_tags_from_asset_path from app.assets.services.snapshot_hash import snapshot_hash @@ -75,6 +76,7 @@ def drain_transition_queue(session: Session) -> None: if digest is None: _PENDING_QUEUE.append(path) continue + stored_hash = to_stored_hash(digest) content = session.scalars( select(AssetContent).where( AssetContent.path == path, AssetContent.is_missing.is_(False) @@ -82,17 +84,16 @@ def drain_transition_queue(session: Session) -> None: ).first() if content is None: continue - current_hash = digest if content.hash is None: - content.hash = current_hash - elif content.hash != current_hash: + content.hash = stored_hash + elif content.hash != stored_hash: mark_content_missing(session, content.id) stat = os.stat(path) name, tags = get_name_and_tags_from_asset_path(path) replacement = create_content( session, path=path, - hash=current_hash, + hash=stored_hash, size_bytes=stat.st_size, mtime_ns=stat.st_mtime_ns, ) diff --git a/app/assets/services/ingest.py b/app/assets/services/ingest.py index e7065b6f2..8b8278bdd 100644 --- a/app/assets/services/ingest.py +++ b/app/assets/services/ingest.py @@ -27,7 +27,7 @@ from app.assets.database.queries import ( upsert_reference as _legacy_upsert_reference, # wave-3-fixes: replaced by B-schema write paths in Wave 3 validate_tags_exist, ) -from app.assets.helpers import get_utc_now, normalize_tags +from app.assets.helpers import get_utc_now, normalize_tags, to_stored_hash from app.assets.services.bulk_ingest import batch_insert_seed_assets from app.assets.services.file_utils import get_mtime_ns, get_size_and_mtime_ns from app.assets.services.image_dimensions import extract_image_dimensions @@ -53,10 +53,12 @@ from app.database.db import create_session def _normalize_hash_input(hash_str: str) -> str: - """Strip blake3: prefix from client-supplied hashes for DB lookup.""" - if hash_str and hash_str.lower().startswith("blake3:"): - return hash_str[7:] - return hash_str + """Canonicalise a client-supplied hash to stored form.""" + if not hash_str: + return hash_str + normalized = hash_str.strip().lower() + digest = normalized.partition(":")[2] or normalized + return to_stored_hash(digest) def _extract_system_metadata_sync( @@ -663,12 +665,13 @@ def upload_from_temp_path( except UploadUnstableError: _remove_temp_path(temp_path) raise - if expected_hash and digest != _normalize_hash_input(expected_hash).lower(): + stored_hash = to_stored_hash(digest) + if expected_hash and stored_hash != _normalize_hash_input(expected_hash): _remove_temp_path(temp_path) raise HashMismatchError("Uploaded file hash does not match provided hash.") with create_session() as session: - dedup = lookup_for_upload_dedup(session, digest, display_name) + dedup = lookup_for_upload_dedup(session, stored_hash, display_name) if isinstance(dedup, Asset): _remove_temp_path(temp_path) @@ -706,7 +709,7 @@ def upload_from_temp_path( size_bytes, mtime_ns = get_size_and_mtime_ns(dest_abs) with create_session() as session: content = create_content( - session, dest_abs, digest, size_bytes, mtime_ns + session, dest_abs, stored_hash, size_bytes, mtime_ns ) record = _create_upload_record( session, @@ -752,8 +755,9 @@ def register_file_in_place( size_bytes, mtime_ns = get_size_and_mtime_ns(locator) digest = _snapshot_hash_with_retry(locator) + stored_hash = to_stored_hash(digest) with create_session() as session: - dedup = lookup_for_upload_dedup(session, digest, display_name) + dedup = lookup_for_upload_dedup(session, stored_hash, display_name) if isinstance(dedup, Asset): with create_session() as session: @@ -779,7 +783,7 @@ def register_file_in_place( with create_session() as session: content = create_content( - session, locator, digest, size_bytes, mtime_ns + session, locator, stored_hash, size_bytes, mtime_ns ) record = _create_upload_record( session, @@ -810,13 +814,14 @@ def create_from_hash( if not mode.hashing_enabled(): return None - digest = _normalize_hash_input(hash_str) + stored_hash = _normalize_hash_input(hash_str) + bare_digest = stored_hash.partition(":")[2] or stored_hash display_name = _sanitize_filename( - name, fallback=digest + name, fallback=bare_digest ) with create_session() as session: - content = lookup_for_from_hash(session, digest) + content = lookup_for_from_hash(session, stored_hash) if content is None: logging.warning("create_from_hash: no asset found for hash %s", hash_str) return None diff --git a/tests-unit/assets_test/services/test_detection_gate.py b/tests-unit/assets_test/services/test_detection_gate.py index 3c4165884..580ece25b 100644 --- a/tests-unit/assets_test/services/test_detection_gate.py +++ b/tests-unit/assets_test/services/test_detection_gate.py @@ -6,6 +6,7 @@ import pytest from sqlalchemy import select from app.assets.database.models import Asset, AssetContent +from app.assets.helpers import to_stored_hash from app.assets.scanner import ( clear_pending_verifications, drain_pending_verifications, @@ -42,6 +43,12 @@ def _bump_mtime(path: Path) -> None: os.utime(path, ns=(stat.st_atime_ns, stat.st_mtime_ns + 1_000_000)) +def _stored_hash(path: Path) -> str: + digest = snapshot_hash(str(path)) + assert digest is not None + return to_stored_hash(digest) + + def test_off_mode_touch_splits(session, temp_dir: Path): input_root = temp_dir / "input" input_root.mkdir() @@ -68,7 +75,7 @@ def test_hash_mode_touch_refreshes_mtime(session, temp_dir: Path): input_root.mkdir() path = input_root / "touched.bin" path.write_bytes(b"same bytes") - old_content, _ = _seed_content(session, path, snapshot_hash(str(path))) + old_content, _ = _seed_content(session, path, _stored_hash(path)) _bump_mtime(path) with ( @@ -91,7 +98,7 @@ def test_hash_mode_real_edit_splits(session, temp_dir: Path): input_root.mkdir() path = input_root / "edited.bin" path.write_bytes(b"old bytes") - old_content, _ = _seed_content(session, path, snapshot_hash(str(path))) + old_content, _ = _seed_content(session, path, _stored_hash(path)) path.write_bytes(b"new bytes with a different length") with ( @@ -105,7 +112,7 @@ def test_hash_mode_real_edit_splits(session, temp_dir: Path): contents = list(session.scalars(select(AssetContent).order_by(AssetContent.created_at))) assert len(contents) == 2 assert session.get(AssetContent, old_content.id).is_missing is True - assert next(content for content in contents if not content.is_missing).hash == snapshot_hash(str(path)) + assert next(content for content in contents if not content.is_missing).hash == _stored_hash(path) def test_old_record_id_resolves_to_missing_content_after_split(session, temp_dir: Path): @@ -113,7 +120,7 @@ def test_old_record_id_resolves_to_missing_content_after_split(session, temp_dir input_root.mkdir() path = input_root / "edited.bin" path.write_bytes(b"old bytes") - old_content, old_record = _seed_content(session, path, snapshot_hash(str(path))) + old_content, old_record = _seed_content(session, path, _stored_hash(path)) path.write_bytes(b"replacement bytes") with ( diff --git a/tests-unit/assets_test/services/test_enrichment_snapshot.py b/tests-unit/assets_test/services/test_enrichment_snapshot.py index be11668d0..d3311f3b0 100644 --- a/tests-unit/assets_test/services/test_enrichment_snapshot.py +++ b/tests-unit/assets_test/services/test_enrichment_snapshot.py @@ -2,6 +2,7 @@ from pathlib import Path from unittest.mock import patch from app.assets.database.models import Asset, AssetContent +from app.assets.helpers import to_stored_hash from app.assets.scanner import enrich_asset @@ -38,7 +39,7 @@ def test_enrichment_uses_snapshot_hash_not_direct_blake3(session, temp_dir: Path ) assert enriched is True - assert session.get(AssetContent, content.id).hash == "snapshot-digest" + assert session.get(AssetContent, content.id).hash == to_stored_hash("snapshot-digest") mocked_snapshot_hash.assert_called_once_with(str(path)) diff --git a/tests-unit/assets_test/services/test_from_hash.py b/tests-unit/assets_test/services/test_from_hash.py index 8aa1f3676..7daf46931 100644 --- a/tests-unit/assets_test/services/test_from_hash.py +++ b/tests-unit/assets_test/services/test_from_hash.py @@ -2,6 +2,7 @@ from sqlalchemy import select from app.assets.database.models import Asset from app.assets.database.queries.records import create_content +from app.assets.helpers import to_stored_hash from app.assets.services.ingest import create_from_hash @@ -14,7 +15,7 @@ def test_create_from_hash_with_prefixed_hash_finds_existing_content( monkeypatch.setattr("app.assets.mode.hashing_enabled", lambda: True) with mock_create_session() as session: - content = create_content(session, str(path), digest, path.stat().st_size) + content = create_content(session, str(path), to_stored_hash(digest), path.stat().st_size) content_id = content.id session.commit() @@ -38,7 +39,7 @@ def test_create_from_hash_with_bare_hash_also_works( monkeypatch.setattr("app.assets.mode.hashing_enabled", lambda: True) with mock_create_session() as session: - content = create_content(session, str(path), digest, path.stat().st_size) + content = create_content(session, str(path), to_stored_hash(digest), path.stat().st_size) content_id = content.id session.commit() diff --git a/tests-unit/assets_test/services/test_recovery_gate.py b/tests-unit/assets_test/services/test_recovery_gate.py index bcfec47ea..ce9c71aea 100644 --- a/tests-unit/assets_test/services/test_recovery_gate.py +++ b/tests-unit/assets_test/services/test_recovery_gate.py @@ -5,11 +5,13 @@ import pytest from sqlalchemy import select from app.assets.database.models import Asset, AssetContent, AssetTag, Tag +from app.assets.helpers import to_stored_hash from app.assets.scanner import ( clear_pending_verifications, pending_recovery_count, seed_asset_specs, ) +from app.assets.services.bulk_ingest import SeedAssetSpec from app.assets.services.snapshot_hash import snapshot_hash @@ -34,7 +36,7 @@ def _missing_content(session, path: Path, hash_value: str) -> tuple[AssetContent return content, record -def _spec(path: Path) -> dict: +def _spec(path: Path) -> SeedAssetSpec: stat = path.stat() return { "abs_path": str(path), @@ -50,10 +52,16 @@ def _spec(path: Path) -> dict: } +def _stored_hash(path: Path) -> str: + digest = snapshot_hash(str(path)) + assert digest is not None + return to_stored_hash(digest) + + def test_single_hash_match_recovers(session, temp_dir: Path): path = temp_dir / "restored.bin" path.write_bytes(b"restored bytes") - content, record = _missing_content(session, path, snapshot_hash(str(path))) + content, record = _missing_content(session, path, _stored_hash(path)) with patch("app.assets.scanner.mode.hashing_enabled", return_value=True): created = seed_asset_specs(session, [_spec(path)]) @@ -67,7 +75,7 @@ def test_single_hash_match_recovers(session, temp_dir: Path): def test_ambiguous_hash_match_recovers_nothing(session, temp_dir: Path): path = temp_dir / "ambiguous.bin" path.write_bytes(b"same bytes") - digest = snapshot_hash(str(path)) + digest = _stored_hash(path) first, _ = _missing_content(session, path, digest) second, _ = _missing_content(session, path, digest) @@ -98,7 +106,7 @@ def test_no_hash_match_creates_fresh_rows(session, temp_dir: Path): def test_off_mode_no_recovery(session, temp_dir: Path): path = temp_dir / "off.bin" path.write_bytes(b"bytes") - missing, _ = _missing_content(session, path, snapshot_hash(str(path))) + missing, _ = _missing_content(session, path, _stored_hash(path)) with ( patch("app.assets.scanner.mode.hashing_enabled", return_value=False), diff --git a/tests-unit/assets_test/services/test_scanner_b.py b/tests-unit/assets_test/services/test_scanner_b.py index c52ba78b7..136395d15 100644 --- a/tests-unit/assets_test/services/test_scanner_b.py +++ b/tests-unit/assets_test/services/test_scanner_b.py @@ -8,6 +8,7 @@ from sqlalchemy import select from sqlalchemy.orm import Session from app.assets.database.models import Asset, AssetContent, AssetTag +from app.assets.helpers import to_stored_hash from app.assets.scanner import ( build_asset_specs, enrich_asset, @@ -213,7 +214,7 @@ def test_enrichment_keeps_equal_hash_contents_distinct(session, temp_dir: Path): ) existing = AssetContent( path=str(temp_dir / "existing.bin"), - hash=digest, + hash=to_stored_hash(digest), size_bytes=path.stat().st_size, mtime_ns=path.stat().st_mtime_ns, ) @@ -234,6 +235,6 @@ def test_enrichment_keeps_equal_hash_contents_distinct(session, temp_dir: Path): ) assert enriched is True - assert session.get(AssetContent, content.id).hash == digest + assert session.get(AssetContent, content.id).hash == to_stored_hash(digest) assert session.get(Asset, record.id).content_id == content.id assert session.get(Asset, existing_record.id).content_id == existing.id diff --git a/tests-unit/assets_test/services/test_serving_b.py b/tests-unit/assets_test/services/test_serving_b.py index ae0390a00..52c161ea0 100644 --- a/tests-unit/assets_test/services/test_serving_b.py +++ b/tests-unit/assets_test/services/test_serving_b.py @@ -17,6 +17,7 @@ from app.assets.database.queries.records import ( get_record_by_id, mark_content_missing, ) +from app.assets.helpers import to_stored_hash from app.assets.services.asset_management import ( asset_exists, get_asset_detail, @@ -103,7 +104,7 @@ def test_resolve_hash_to_path_refuses_temp_content(mock_create_session, session, digest = "e" * 64 f = temp_dir / "temp_only.bin" f.write_bytes(b"temp") - create_content(session, path=str(f), hash=digest) + create_content(session, path=str(f), hash=to_stored_hash(digest)) session.commit() with patch("app.assets.services.lookup.is_temp_path", return_value=True): @@ -147,7 +148,7 @@ def test_view_hash_read_updates_last_access_time( digest = "a" * 64 f = temp_dir / "view.bin" f.write_bytes(b"view") - content = create_content(session, path=str(f), hash=digest) + content = create_content(session, path=str(f), hash=to_stored_hash(digest)) record = create_record(session, content_id=content.id, name="view.bin") session.commit() record_id = record.id @@ -169,11 +170,11 @@ def test_lookup_for_view_returns_none_for_temp_content(session, temp_dir): digest = "b" * 64 f = temp_dir / "view_temp.bin" f.write_bytes(b"temp") - create_content(session, path=str(f), hash=digest) + create_content(session, path=str(f), hash=to_stored_hash(digest)) session.commit() with patch("app.assets.services.lookup.is_temp_path", return_value=True): - assert lookup_for_view(session, digest) is None + assert lookup_for_view(session, to_stored_hash(digest)) is None def test_asset_exists_false_for_temp_content(mock_create_session, session, temp_dir): @@ -181,7 +182,7 @@ def test_asset_exists_false_for_temp_content(mock_create_session, session, temp_ digest = "c" * 64 f = temp_dir / "exists_temp.bin" f.write_bytes(b"temp") - create_content(session, path=str(f), hash=digest) + create_content(session, path=str(f), hash=to_stored_hash(digest)) session.commit() with patch("app.assets.services.lookup.is_temp_path", return_value=True): @@ -196,7 +197,7 @@ async def test_head_hash_route_404_for_temp_content( digest = "d" * 64 f = temp_dir / "head_temp.bin" f.write_bytes(b"temp") - create_content(session, path=str(f), hash=digest) + create_content(session, path=str(f), hash=to_stored_hash(digest)) session.commit() monkeypatch.setattr(routes, "_ASSETS_ENABLED", True) @@ -220,7 +221,7 @@ def test_resolve_hash_to_path_temp_does_not_bump_last_access_time( digest = "f" * 64 f = temp_dir / "noaccess_temp.bin" f.write_bytes(b"temp") - content = create_content(session, path=str(f), hash=digest) + content = create_content(session, path=str(f), hash=to_stored_hash(digest)) record = create_record(session, content_id=content.id, name="noaccess_temp.bin") session.commit() record_id = record.id diff --git a/tests-unit/assets_test/services/test_stored_hash_prefix.py b/tests-unit/assets_test/services/test_stored_hash_prefix.py new file mode 100644 index 000000000..5d097c2cd --- /dev/null +++ b/tests-unit/assets_test/services/test_stored_hash_prefix.py @@ -0,0 +1,281 @@ +from __future__ import annotations + +import json +import os +import uuid +from unittest.mock import AsyncMock + +import pytest +from aiohttp import web +from aiohttp.test_utils import make_mocked_request +from sqlalchemy import func, select + +import app.assets.mode as mode_module +from app.assets.database.models import Asset, AssetContent +from app.assets.database.queries.records import create_record +from app.assets.scanner import enrich_asset +from app.assets.scanner_changes import recover_missing_content +from app.assets.services.ingest import ( + HashMismatchError, + upload_from_temp_path, +) +from app.assets.services.snapshot_hash import snapshot_hash + + +@pytest.fixture +def hashing_on(): + class FakeArgs: + enable_asset_hashing = True + + class DisabledArgs: + enable_asset_hashing = False + + mode_module.init(FakeArgs()) + yield + mode_module.init(DisabledArgs()) + + +def _write_temp(content: bytes) -> str: + import folder_paths + + uploads_root = os.path.join( + folder_paths.get_temp_directory(), "uploads", uuid.uuid4().hex + ) + os.makedirs(uploads_root, exist_ok=True) + path = os.path.join(uploads_root, ".upload.part") + with open(path, "wb") as file: + file.write(content) + return path + + +def test_upload_stores_prefixed_hash_expected_hash_succeeds_and_dedups( + mock_create_session, hashing_on +): + content_bytes = b"prefixed-stored-upload-bytes" + temp1 = _write_temp(content_bytes) + digest = snapshot_hash(temp1) + assert digest is not None + expected = f"blake3:{digest}" + assert expected.startswith("blake3:") + + temp2 = _write_temp(content_bytes) + try: + r1 = upload_from_temp_path( + temp_path=temp1, + name="pref.bin", + tags=["output"], + client_filename="pref.bin", + expected_hash=expected, + ) + assert r1.created_new is True + + with mock_create_session() as session: + live = list( + session.scalars( + select(AssetContent).where(AssetContent.is_missing.is_(False)) + ) + ) + assert len(live) == 1 + assert live[0].hash == expected + assert live[0].hash.startswith("blake3:") + + assert r1.asset.hash == expected + + r2 = upload_from_temp_path( + temp_path=temp2, + name="pref.bin", + tags=["output"], + client_filename="pref.bin", + expected_hash=expected, + ) + assert r2.created_new is False + assert r2.ref.id == r1.ref.id + with mock_create_session() as session: + assert session.scalar(select(func.count()).select_from(AssetContent)) == 1 + finally: + for path in (temp1, temp2): + if os.path.exists(path): + os.unlink(path) + + +def test_upload_expected_hash_mismatch_still_rejected(mock_create_session, hashing_on): + temp = _write_temp(b"mismatch-bytes") + try: + with pytest.raises(HashMismatchError): + upload_from_temp_path( + temp_path=temp, + name="mismatch.bin", + tags=["output"], + client_filename="mismatch.bin", + expected_hash=f"blake3:{'0' * 64}", + ) + with mock_create_session() as session: + assert session.scalar(select(func.count()).select_from(Asset)) == 0 + assert session.scalar(select(func.count()).select_from(AssetContent)) == 0 + finally: + if os.path.exists(temp): + os.unlink(temp) + + +def test_enrich_fills_deferred_hash_prefixed(session, temp_dir): + path = temp_dir / "deferred.bin" + path.write_bytes(b"deferred output bytes") + stat = path.stat() + + content = AssetContent( + path=str(path), + hash=None, + size_bytes=stat.st_size, + mtime_ns=stat.st_mtime_ns, + ) + session.add(content) + session.flush() + record = create_record(session, content.id, path.name) + session.commit() + + enriched = enrich_asset( + session, + file_path=str(path), + content_id=content.id, + record_id=record.id, + extract_metadata=False, + compute_hash=True, + ) + + assert enriched is True + stored = session.get(AssetContent, content.id).hash + assert stored is not None + assert stored.startswith("blake3:") + digest = snapshot_hash(str(path)) + assert digest is not None + assert stored == f"blake3:{digest}" + + +def test_recovery_matches_prefixed_stored_hash(session, temp_dir): + path = temp_dir / "recover.bin" + original_bytes = b"recover-me-bytes" + path.write_bytes(original_bytes) + digest = snapshot_hash(str(path)) + assert digest is not None + stored = f"blake3:{digest}" + assert stored.startswith("blake3:") + + content = AssetContent( + path=str(path), + hash=stored, + is_missing=True, + size_bytes=0, + mtime_ns=None, + ) + session.add(content) + session.flush() + create_record(session, content.id, path.name) + session.commit() + + path.unlink() + path.write_bytes(original_bytes) + stat = os.stat(str(path)) + result = recover_missing_content( + session, str(path), stat, hashing_is_enabled=True + ) + + assert result == "recovered" + recovered = session.get(AssetContent, content.id) + assert recovered is not None + assert recovered.is_missing is False + assert recovered.hash == stored + + +@pytest.mark.asyncio +async def test_all_read_surfaces_agree_on_prefixed_hash( + db_engine, monkeypatch, hashing_on +): + from contextlib import contextmanager + + from sqlalchemy.orm import Session as SASession + + from app.assets import mode + from app.assets.api import routes + from app.assets.services import asset_management, ingest + from app.assets.services.asset_management import get_asset_detail + + @contextmanager + def _factory(): + with SASession(db_engine) as sess: + yield sess + + monkeypatch.setattr(routes, "create_session", lambda: SASession(db_engine)) + monkeypatch.setattr(routes, "_ASSETS_ENABLED", True) + monkeypatch.setattr(mode, "hashing_enabled", lambda: True) + monkeypatch.setattr(ingest, "create_session", _factory) + monkeypatch.setattr(asset_management, "create_session", _factory) + + content_bytes = b"one-asset-all-surfaces-agree" + temp = _write_temp(content_bytes) + digest = snapshot_hash(temp) + assert digest is not None + expected = f"blake3:{digest}" + + try: + upload_result = ingest.upload_from_temp_path( + temp_path=temp, + name="surf.bin", + tags=["output"], + client_filename="surf.bin", + ) + asset_id = upload_result.ref.id + + assert upload_result.asset is not None + assert upload_result.asset.hash == expected + upload_resp = routes._build_asset_response(upload_result, {}) + assert upload_resp.hash == expected + + detail = get_asset_detail(asset_id) + assert detail is not None + assert detail.asset is not None + assert detail.asset.hash == expected + + get_resp = await routes.get_asset_route( + make_mocked_request( + "GET", f"/api/assets/{asset_id}", match_info={"id": asset_id} + ) + ) + assert isinstance(get_resp, web.Response) + assert isinstance(get_resp.body, bytes | bytearray) + get_body = json.loads(get_resp.body) + assert get_body["hash"] == expected + + list_resp = await routes.list_assets_route( + make_mocked_request("GET", "/api/assets") + ) + assert isinstance(list_resp, web.Response) + assert isinstance(list_resp.body, bytes | bytearray) + list_body = json.loads(list_resp.body) + item = next(a for a in list_body["assets"] if a["id"] == asset_id) + assert item["hash"] == expected + + head_resp = await routes.head_asset_by_hash( + make_mocked_request( + "HEAD", + f"/api/assets/hash/{expected}", + match_info={"hash": expected}, + ) + ) + assert isinstance(head_resp, web.Response) + assert head_resp.status == 200 + + fh_req = AsyncMock(spec=web.Request) + fh_req.json.return_value = { + "hash": expected, + "name": "fromhash.bin", + "tags": ["output"], + } + fh_resp = await routes.create_asset_from_hash_route(fh_req) + assert isinstance(fh_resp, web.Response) + assert fh_resp.status == 201 + assert isinstance(fh_resp.body, bytes | bytearray) + fh_body = json.loads(fh_resp.body) + assert fh_body["hash"] == expected + finally: + if os.path.exists(temp): + os.unlink(temp) diff --git a/tests-unit/assets_test/services/test_transition_drain.py b/tests-unit/assets_test/services/test_transition_drain.py index 58f5ceca6..12343be4c 100644 --- a/tests-unit/assets_test/services/test_transition_drain.py +++ b/tests-unit/assets_test/services/test_transition_drain.py @@ -1,8 +1,11 @@ +from pathlib import Path + import pytest from sqlalchemy import select from app.assets.database.models import Asset, AssetContent from app.assets.database.queries.records import create_content, create_record +from app.assets.helpers import to_stored_hash from app.assets.services import hash_mode_state from app.assets.services.hash_mode_state import ( clear_transition_queue, @@ -22,6 +25,12 @@ def transition_queue(): clear_transition_queue() +def _stored_hash(path: Path) -> str: + digest = snapshot_hash(str(path)) + assert digest is not None + return to_stored_hash(digest) + + def test_off_to_on_transition_hashes_null_rows_and_persists_mode(session, temp_dir, monkeypatch): paths = [temp_dir / "first.bin", temp_dir / "second.bin"] for index, path in enumerate(paths): @@ -37,7 +46,7 @@ def test_off_to_on_transition_hashes_null_rows_and_persists_mode(session, temp_d session.commit() contents = list(session.scalars(select(AssetContent))) - assert {content.hash for content in contents} == {snapshot_hash(str(path)) for path in paths} + assert {content.hash for content in contents} == {_stored_hash(path) for path in paths} assert read_stored_mode(session) == "on" @@ -49,7 +58,7 @@ def test_transition_drain_splits_changed_content(session, temp_dir, monkeypatch) assert old_digest is not None stat = path.stat() old_content = create_content( - session, str(path), old_digest, stat.st_size, stat.st_mtime_ns + session, str(path), to_stored_hash(old_digest), stat.st_size, stat.st_mtime_ns ) old_content_id = old_content.id create_record(session, old_content_id, "changed.bin") @@ -63,6 +72,6 @@ def test_transition_drain_splits_changed_content(session, temp_dir, monkeypatch) live_content = next(content for content in contents if not content.is_missing) records = list(session.scalars(select(Asset))) assert session.get(AssetContent, old_content_id).is_missing is True - assert live_content.hash == snapshot_hash(str(path)) + assert live_content.hash == _stored_hash(path) assert len(records) == 2 assert any(record.content_id == live_content.id for record in records) diff --git a/tests-unit/assets_test/test_upload_hashing_modes.py b/tests-unit/assets_test/test_upload_hashing_modes.py index 7a2d26a47..38785a4c3 100644 --- a/tests-unit/assets_test/test_upload_hashing_modes.py +++ b/tests-unit/assets_test/test_upload_hashing_modes.py @@ -17,10 +17,10 @@ modes), the scanner assertion skips (see ``server_hashing_enabled``), and the output assertion is self-contained (it drives the ingest function in-process with the flag forced off). -The authoritative hash signal is the ``asset_contents.hash`` column: a raw -64-char BLAKE3 hex digest when hashed, or ``NULL`` when not. (The HTTP layer -omits null hashes entirely via ``exclude_none=True`` and prefixes non-null ones -with ``blake3:``, so the DB column is the stable thing to assert on.) +The authoritative hash signal is the ``asset_contents.hash`` column: the stored +canonical ``blake3:`` form when hashed, or ``NULL`` when not. (The HTTP +layer omits null hashes entirely via ``exclude_none=True`` and returns that same +stored form, so the DB column is the stable thing to assert on.) """ from __future__ import annotations @@ -63,7 +63,7 @@ def _query(db_path: str, sql: str, params: tuple[Any, ...] = ()) -> list[tuple[A def _content_hash_for_asset(db_path: str, asset_id: str) -> str | None: - """Return the ``asset_contents.hash`` backing an asset record (raw hex or None).""" + """Return the ``asset_contents.hash`` backing an asset record (stored ``blake3:`` or None).""" rows = _query( db_path, "SELECT c.hash FROM asset_contents c " @@ -75,13 +75,12 @@ def _content_hash_for_asset(db_path: str, asset_id: str) -> str | None: return rows[0][0] -def _is_blake3_hex(value: str | None) -> bool: - """True iff ``value`` is a bare 64-char lowercase BLAKE3 hex digest.""" - return ( - isinstance(value, str) - and len(value) == 64 - and all(c in "0123456789abcdef" for c in value.lower()) - ) +def _is_stored_blake3_hash(value: str | None) -> bool: + """True iff ``value`` is the stored canonical ``blake3:`` form.""" + if not isinstance(value, str) or not value.startswith("blake3:"): + return False + digest = value[len("blake3:") :] + return len(digest) == 64 and all(c in "0123456789abcdef" for c in digest.lower()) # --------------------------------------------------------------------------- # @@ -144,7 +143,7 @@ def test_upload_via_api_hashes_in_on_mode(http, api_base, comfy_tmp_base_dir, re ) assert status in (200, 201), body digest = _content_hash_for_asset(db_path, body["id"]) - assert _is_blake3_hex(digest), f"expected a blake3 digest, got {digest!r}" + assert _is_stored_blake3_hash(digest), f"expected a stored blake3 hash, got {digest!r}" def test_upload_via_image_hashes_in_on_mode(http, api_base, comfy_tmp_base_dir, request): @@ -156,7 +155,7 @@ def test_upload_via_image_hashes_in_on_mode(http, api_base, comfy_tmp_base_dir, asset = body.get("asset") assert asset and asset.get("id"), f"/upload/image did not register an asset: {body}" digest = _content_hash_for_asset(db_path, asset["id"]) - assert _is_blake3_hex(digest), f"expected a blake3 digest, got {digest!r}" + assert _is_stored_blake3_hash(digest), f"expected a stored blake3 hash, got {digest!r}" def test_upload_dedup_works_in_on_mode(http, api_base):