upstream-archive/ComfyUI
1
0
Fork 0
mirror of https://github.com/Comfy-Org/ComfyUI.git synced 2026-09-21 05:27:57 -05:00
Code Issues Packages Projects Releases Wiki Activity

feat: structured event log lines for the assets system (#16306)

Browse Source 
* feat(assets): structured event log lines for the assets system

* test(assets): AST check that tagged event lines stay inside the closed vocabulary

* feat(assets): structured event logging for the seeder lifecycle

* feat(assets): structured event logging for scanner and ingest failure paths

* feat(assets): structured event logging for API request failures

* fix(seeder): finalize checkpoint no longer parks completed scans

* fix(scanner): enrich_failed counts exceptions, not benign races

* fix(seeder): idle reset survives emit/assert failures

* refactor(assets): private emit-once bookkeeping helper; clean Progress DTO

* test(event-log): call-site registry as frozenset; drop dead pending machinery

* feat(event-log): scanner.stat_failed with emit-once discipline; drop dead vocabulary

* refactor(assets): drop API request failure events

* refactor(assets): drop ingest failure events

* refactor(assets): narrow event vocabulary to scan pipeline

* test(assets): per-call-site event tests collapse to mechanism-once

* refactor(assets): event log lines as logfmt, matching the log's own idiom

* test(assets): adapt hash-failure log assertion to privacy-safe scan-error logging

The rebase onto master picked up the privacy-safe _log_scan_error from
PR 16096, which no longer includes the file path in the log message.
Assert on the generic 'Asset scan error' message and that the path does
not leak, instead of asserting the path is present.

* fix(assets): enforce event vocabulary and scan counters

---------

Co-authored-by: Simon Pinfold <synap5e@users.noreply.github.com>
Co-authored-by: guill <jacob.e.segal@gmail.com>
This commit is contained in:
Christian Byrne
2026-09-13 22:24:33 -07:00
committed by GitHub
co-authored by Simon Pinfold guill
parent eecbfb4046
commit f42b24efbe
10 changed files with 1903 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/assets/event_log.py
+163
View File
@@ -0,0 +1,163 @@
"""Structured event log lines for the assets system.
Every line is ``[assets-event] <event> key=value ...`` on the standard logging
INFO channel, with fields sorted by name and omitted when an event has none. This
mirrors the ``assets.seed.*`` events the seeder already puts on the PromptServer
bus. A log-tailing launcher can pick assets health signals out of core's output
without parsing prose, and the existing human-readable lines stay exactly as
they are.
The field vocabulary is closed. Only the names in :data:`ALLOWED_FIELDS` may be
carried, each has a validator, and no string value may contain a path separator
or logfmt delimiter — so file names, paths, asset ids and content hashes cannot
ride along.
"""
import logging
import os
import traceback
from collections.abc import Callable
from typing import Any
TAG = "[assets-event]"
MAX_STRING_LENGTH = 64
FORBIDDEN_STRING_CHARS = ("/", "\\", ":", " ", "=", '"')
ROOTS = frozenset({"models", "input", "output", "user", "temp"})
PHASES = frozenset({"fast", "enrich", "full"})
STAGES = frozenset({"mark_missing", "pruning", "fast_scan", "enrich", "finalize"})
STAT_SITES = frozenset({"discovery", "enrich"})
ALLOWED_EVENTS = frozenset({
"assets.enabled",
"seeder.scan_started",
"seeder.scan_completed",
"seeder.scan_failed",
"seeder.scan_cancelled",
"seeder.marked_missing",
"seeder.batch_insert_failed",
"scanner.hash_failed",
"scanner.enrich_failed",
"scanner.hash_discarded_modified",
"scanner.fast_scan_failed",
"scanner.temp_sync_failed",
"scanner.mark_missing_failed",
"scanner.stat_failed",
})
class EventLogError(ValueError):
"""An emit() call that would break the closed event vocabulary."""
def _is_safe_string(value: Any) -> bool:
return (
isinstance(value, str)
and 0 < len(value) <= MAX_STRING_LENGTH
and not any(char in value for char in FORBIDDEN_STRING_CHARS)
)
def _one_of(allowed: frozenset[str]) -> Callable[[Any], bool]:
def validate(value: Any) -> bool:
return _is_safe_string(value) and value in allowed
return validate
def _is_count(value: Any) -> bool:
# bool subclasses int, so it has to be excluded before the int check.
return isinstance(value, int) and not isinstance(value, bool)
def _is_flag(value: Any) -> bool:
return isinstance(value, bool)
ALLOWED_FIELDS: dict[str, Callable[[Any], bool]] = {
"root": _one_of(ROOTS),
"phase": _one_of(PHASES),
"stage": _one_of(STAGES),
"elapsed_ms": _is_count,
"created": _is_count,
"enriched": _is_count,
"skipped": _is_count,
"hash_failed": _is_count,
"enrich_failed": _is_count,
"permission_denied": _is_count,
"count": _is_count,
"error_type": _is_safe_string,
"hashing_enabled": _is_flag,
"site": _one_of(STAT_SITES),
}
_warned_call_sites: set[tuple[str, int]] = set()
def _find_problem(event: Any, fields: dict[str, Any]) -> str | None:
if not isinstance(event, str) or event not in ALLOWED_EVENTS:
return "invalid event name"
for name, value in fields.items():
validate = ALLOWED_FIELDS.get(name)
if validate is None:
return f"field {name!r} is not in the allowed vocabulary"
if not validate(value):
return f"field {name!r} has a value its validator rejected"
return None
def _strict_mode() -> bool:
return (
"PYTEST_CURRENT_TEST" in os.environ
or os.environ.get("COMFYUI_ASSETS_EVENT_LOG_STRICT") == "1"
)
def _caller_call_site() -> tuple[str, int]:
"""Identify emit()'s caller so a bad call site warns at most once."""
caller = traceback.extract_stack(limit=3)[0]
return (caller.filename, caller.lineno or 0)
def emit(event: str, *, root: str | None = None, **fields: Any) -> None:
"""Log one tagged event line.
An invalid call raises in strict mode (under pytest, or with
COMFYUI_ASSETS_EVENT_LOG_STRICT=1) so a bad call site fails the test suite.
In production it warns once per call site and drops the event, so a
vocabulary mistake can never break a running server.
"""
if root is not None:
fields["root"] = root
problem = _find_problem(event, fields)
if problem is None:
pairs = " ".join(
f"{name}={str(value).lower() if isinstance(value, bool) else value}"
for name, value in sorted(fields.items())
)
line = f"{TAG} {event}" + (f" {pairs}" if pairs else "")
logging.info("%s", line)
return
if _strict_mode():
raise EventLogError(problem)
call_site = _caller_call_site()
if call_site not in _warned_call_sites:
_warned_call_sites.add(call_site)
logging.warning(
"Dropped an invalid assets event at %s:%d: %s",
call_site[0],
call_site[1],
problem,
)
def error_type(exc: BaseException) -> str:
"""The only sanctioned description of an exception: its class name.
Stringifying the exception itself is banned here, because FileNotFoundError
and friends embed the path that triggered them.
"""
return type(exc).__name__
app/assets/scanner.py
+80 -16
View File
@@ -11,13 +11,14 @@ import logging
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Literal, TypedDict
from typing import Callable, Literal, Protocol, TypedDict
import folder_paths
import sqlalchemy as sa
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.assets import mode
from app.assets.event_log import emit, error_type
from app.assets.database.queries import (
create_content_reporting_insert,
mark_content_missing,
@@ -66,6 +67,14 @@ __all__ = [
RootType = Literal["models", "input", "output"]
class _ScanProgress(Protocol):
hash_failed: int
enrich_failed: int
permission_denied: int
def mark_emitted(self, key: str) -> bool: ...
class SeedAssetSpec(TypedDict):
abs_path: str
@@ -147,11 +156,13 @@ def sync_references_with_filesystem(
session,
root: RootType,
collect_existing_paths: bool = False,
progress: _ScanProgress | None = None,
) -> set[str] | None:
return sync_prefixes_with_filesystem(
session,
get_scan_prefixes_for_root(root),
collect_existing_paths=collect_existing_paths,
progress=progress,
)
@@ -159,6 +170,7 @@ def sync_prefixes_with_filesystem(
session: Session,
prefixes: list[str],
collect_existing_paths: bool = False,
progress: _ScanProgress | None = None,
) -> set[str] | None:
if not prefixes:
return set() if collect_existing_paths else None
@@ -171,6 +183,8 @@ def sync_prefixes_with_filesystem(
mark_content_missing(session, content.id)
except PermissionError as e:
_log_scan_error("reference_stat", e)
if progress is not None:
progress.permission_denied += 1
logging.debug("Permission denied accessing %s", content.path)
except OSError as e:
_log_scan_error("reference_stat", e)
@@ -192,7 +206,9 @@ def _is_under_prefixes(path: str, prefixes: list[str]) -> bool:
return is_path_under_prefixes(path, prefixes)
def sync_root_safely(root: RootType) -> set[str]:
def sync_root_safely(
root: RootType, progress: _ScanProgress | None = None
) -> set[str]:
"""Sync a single root's references with the filesystem.
Returns survivors (existing paths) or empty set on failure.
@@ -203,22 +219,39 @@ def sync_root_safely(root: RootType) -> set[str]:
sess,
root,
collect_existing_paths=True,
progress=progress,
)
sess.commit()
return survivors or set()
except Exception as e:
logging.exception("fast DB scan failed for %s: %s", root, e)
except Exception as exc:
logging.exception("fast DB scan failed for %s: %s", root, exc)
emit(
"scanner.fast_scan_failed",
root=root,
error_type=error_type(exc),
)
return set()
def sync_temp_references_safely() -> None:
def sync_temp_references_safely(
progress: _ScanProgress | None = None,
) -> None:
"""Retire temp references whose file is gone; temp is never scanned, so nothing else stats them."""
try:
with create_session() as sess:
sync_prefixes_with_filesystem(sess, get_temp_prefixes())
sync_prefixes_with_filesystem(
sess,
get_temp_prefixes(),
progress=progress,
)
sess.commit()
except Exception as e:
logging.exception("temp reference sync failed: %s", e)
except Exception as exc:
logging.exception("temp reference sync failed: %s", exc)
emit(
"scanner.temp_sync_failed",
root="temp",
error_type=error_type(exc),
)
def mark_missing_outside_prefixes_safely(prefixes: list[str]) -> int:
@@ -231,8 +264,12 @@ def mark_missing_outside_prefixes_safely(prefixes: list[str]) -> int:
count = mark_contents_missing_outside_prefixes(sess, prefixes)
sess.commit()
return count
except Exception as e:
logging.exception("marking missing assets failed: %s", e)
except Exception as exc:
logging.exception("marking missing assets failed: %s", exc)
emit(
"scanner.mark_missing_failed",
error_type=error_type(exc),
)
return 0
@@ -264,6 +301,7 @@ def build_asset_specs(
paths: list[str],
existing_paths: set[str],
enable_metadata_extraction: bool = True,
progress: _ScanProgress | None = None,
) -> tuple[list[SeedAssetSpec], set[str], int]:
"""Build asset specs from paths, returning (specs, tag_pool, skipped_count).
@@ -271,6 +309,7 @@ def build_asset_specs(
paths: List of file paths to process
existing_paths: Set of paths that already exist in the database
enable_metadata_extraction: If True, extract tier 1 & 2 metadata
progress: Optional per-scan state for emit-once bookkeeping
"""
specs: list[SeedAssetSpec] = []
tag_pool: set[str] = set()
@@ -291,6 +330,11 @@ def build_asset_specs(
continue
except OSError as e:
_log_scan_error("discovery_stat", e)
if progress is not None:
if isinstance(e, PermissionError):
progress.permission_denied += 1
if progress.mark_emitted("stat_failed:discovery"):
emit("scanner.stat_failed", site="discovery", error_type=error_type(e))
continue
if not stat_p.st_size:
continue
@@ -447,6 +491,7 @@ def enrich_asset(
record_id: str,
extract_metadata: bool = True,
compute_hash: bool = False,
progress: _ScanProgress | None = None,
) -> bool:
"""Enrich a single asset with metadata and/or hash.
@@ -467,6 +512,11 @@ def enrich_asset(
return False
except OSError as e:
_log_scan_error("enrichment_stat", e)
if progress is not None:
if isinstance(e, PermissionError):
progress.permission_denied += 1
if progress.mark_emitted("stat_failed:enrich"):
emit("scanner.stat_failed", site="enrich", error_type=error_type(e))
return False
initial_mtime_ns = get_mtime_ns(stat_p)
@@ -493,6 +543,8 @@ def enrich_asset(
try:
snapshot = snapshot_hash(file_path)
if snapshot is None:
if progress is None or progress.mark_emitted("hash_discarded_modified"):
emit("scanner.hash_discarded_modified")
logging.warning(
"File modified during hashing (snapshot unstable), discarding hash: %s",
file_path,
@@ -500,11 +552,17 @@ def enrich_asset(
return False
digest, verified_stat = snapshot
stored_hash = to_stored_hash(digest)
except Exception as e:
if isinstance(e, OSError):
_log_scan_error("hashing", e)
except Exception as exc:
emit_failure = progress is None
if progress is not None:
progress.hash_failed += 1
emit_failure = progress.mark_emitted("hash_failed")
if emit_failure:
emit("scanner.hash_failed", error_type=error_type(exc))
if isinstance(exc, OSError):
_log_scan_error("hashing", exc)
else:
logging.warning("Failed to hash %s: %s", file_path, e)
logging.warning("Failed to hash %s: %s", file_path, exc)
record = session.get(Asset, record_id)
if content is None or record is None or content.mtime_ns != initial_mtime_ns:
@@ -554,6 +612,7 @@ def enrich_assets_batch(
extract_metadata: bool = True,
compute_hash: bool = False,
interrupt_check: Callable[[], bool] | None = None,
progress: _ScanProgress | None = None,
) -> tuple[int, list[str]]:
"""Enrich a batch of assets.
@@ -587,13 +646,18 @@ def enrich_assets_batch(
record_id=row.record_id,
extract_metadata=extract_metadata,
compute_hash=compute_hash,
progress=progress,
)
if updated:
enriched += 1
else:
failed_ids.append(row.record_id)
except Exception as e:
logging.warning("Failed to enrich %s: %s", row.file_path, e)
except Exception as exc:
if progress is not None:
progress.enrich_failed += 1
if progress is None or progress.mark_emitted("enrich_failed"):
emit("scanner.enrich_failed", error_type=error_type(exc))
logging.warning("Failed to enrich %s: %s", row.file_path, exc)
sess.rollback()
failed_ids.append(row.record_id)
app/assets/seeder.py
+177 -64
View File
@@ -14,6 +14,7 @@ from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, TypedDict
from app.assets.event_log import emit, error_type
from app.assets.scanner import (
RootType,
build_asset_specs,
@@ -60,14 +61,25 @@ class PendingScan(TypedDict):
compute_hashes: bool
class _ScanStage(Enum):
MARK_MISSING = "mark_missing"
PRUNING = "pruning"
FAST_SCAN = "fast_scan"
ENRICH = "enrich"
FINALIZE = "finalize"
@dataclass
class Progress:
"""Progress information for a scan operation."""
"""Public snapshot of a scan's progress. Carries counters only."""
scanned: int = 0
total: int = 0
created: int = 0
skipped: int = 0
hash_failed: int = 0
enrich_failed: int = 0
permission_denied: int = 0
@dataclass
@@ -82,6 +94,46 @@ class ScanStatus:
ProgressCallback = Callable[[Progress], None]
@dataclass
class _ScanState:
"""Mutable in-flight state for one scan; never exposed outside this module.
Satisfies scanner.py's `_ScanProgress` Protocol. `cancel_stage` stores the
stage's string value (not the `_ScanStage` enum) so scanner.py never needs
to import `_ScanStage`. Take a `Progress` snapshot before exposing state.
"""
scanned: int = 0
total: int = 0
created: int = 0
skipped: int = 0
hash_failed: int = 0
enrich_failed: int = 0
permission_denied: int = 0
cancel_stage: str | None = None
_emitted_keys: set[str] = field(default_factory=set)
def mark_emitted(self, key: str) -> bool:
"""Return True the first call with `key` this scan, False every call after."""
if key in self._emitted_keys:
return False
self._emitted_keys.add(key)
return True
def _snapshot_progress(state: _ScanState) -> Progress:
"""Build the public counters-only `Progress` snapshot from live scan state."""
return Progress(
scanned=state.scanned,
total=state.total,
created=state.created,
skipped=state.skipped,
hash_failed=state.hash_failed,
enrich_failed=state.enrich_failed,
permission_denied=state.permission_denied,
)
class _AssetSeeder:
"""Background asset scanning manager.
@@ -95,7 +147,7 @@ class _AssetSeeder:
# holding _lock and re-enters start() which also acquires _lock.
self._lock = threading.RLock()
self._state = State.IDLE
self._progress: Progress | None = None
self._scan_state: _ScanState | None = None
self._last_progress: Progress | None = None
self._errors: list[str] = []
self._thread: threading.Thread | None = None
@@ -155,7 +207,7 @@ class _AssetSeeder:
logging.info("Asset seeder already running, skipping start")
return False
self._state = State.PAUSED if _start_paused else State.RUNNING
self._progress = Progress()
self._scan_state = _ScanState()
self._errors = []
self._roots = roots
self._phase = phase
@@ -355,17 +407,14 @@ class _AssetSeeder:
def get_status(self) -> ScanStatus:
"""Get the current status and progress of the seeder."""
with self._lock:
src = self._progress or self._last_progress
progress = (
_snapshot_progress(self._scan_state)
if self._scan_state is not None
else self._last_progress
)
return ScanStatus(
state=self._state,
progress=Progress(
scanned=src.scanned,
total=src.total,
created=src.created,
skipped=src.skipped,
)
if src
else None,
progress=progress,
errors=list(self._errors),
)
@@ -425,6 +474,11 @@ class _AssetSeeder:
all_prefixes = get_owned_prefixes()
marked = mark_missing_outside_prefixes_safely(all_prefixes)
emit(
"seeder.marked_missing",
count=marked,
stage=_ScanStage.MARK_MISSING.value,
)
if marked > 0:
logging.info("Marked %d references as missing", marked)
return marked
@@ -434,9 +488,10 @@ class _AssetSeeder:
def _reset_to_idle(self) -> None:
"""Reset state to IDLE, preserving last progress. Caller must hold _lock."""
self._last_progress = self._progress
if self._scan_state is not None:
self._last_progress = _snapshot_progress(self._scan_state)
self._state = State.IDLE
self._progress = None
self._scan_state = None
def _is_cancelled(self) -> bool:
"""Check if cancellation has been requested."""
@@ -450,9 +505,17 @@ class _AssetSeeder:
open while blocked. The caller is responsible for blocking on
_check_pause_and_cancel() afterward.
"""
return not self._run_gate.is_set() or self._cancel_event.is_set()
cancelled = self._cancel_event.is_set()
if cancelled:
self._record_cancel_stage(_ScanStage.ENRICH)
return not self._run_gate.is_set() or cancelled
def _check_pause_and_cancel(self) -> bool:
def _record_cancel_stage(self, stage: _ScanStage) -> None:
with self._lock:
if self._scan_state is not None and self._scan_state.cancel_stage is None:
self._scan_state.cancel_stage = stage.value
def _check_pause_and_cancel(self, stage: _ScanStage) -> bool:
"""Block while paused, then check if cancelled.
Call this at checkpoint locations in scan loops. It will:
@@ -465,7 +528,10 @@ class _AssetSeeder:
if not self._run_gate.is_set():
self._emit_event("assets.seed.paused", {})
self._run_gate.wait() # Blocks if paused
return self._is_cancelled()
cancelled = self._is_cancelled()
if cancelled:
self._record_cancel_stage(stage)
return cancelled
def _emit_event(self, event_type: str, data: dict[str, Any]) -> None:
"""Emit a WebSocket event if server is available."""
@@ -487,24 +553,19 @@ class _AssetSeeder:
progress: Progress | None = None
with self._lock:
if self._progress is None:
if self._scan_state is None:
return
if scanned is not None:
self._progress.scanned = scanned
self._scan_state.scanned = scanned
if total is not None:
self._progress.total = total
self._scan_state.total = total
if created is not None:
self._progress.created = created
self._scan_state.created = created
if skipped is not None:
self._progress.skipped = skipped
self._scan_state.skipped = skipped
if self._progress_callback:
callback = self._progress_callback
progress = Progress(
scanned=self._progress.scanned,
total=self._progress.total,
created=self._progress.created,
skipped=self._progress.skipped,
)
progress = _snapshot_progress(self._scan_state)
if callback and progress:
try:
@@ -540,6 +601,7 @@ class _AssetSeeder:
t_start = time.perf_counter()
roots = self._roots
phase = self._phase
root = roots[0] if len(roots) == 1 else None
cancelled = False
total_created = 0
total_enriched = 0
@@ -555,14 +617,23 @@ class _AssetSeeder:
)
return
emit("seeder.scan_started", phase=phase.value, root=root)
assert self._scan_state is not None
scan_state = self._scan_state
if self._prune_first:
all_prefixes = get_owned_prefixes()
marked = mark_missing_outside_prefixes_safely(all_prefixes)
emit(
"seeder.marked_missing",
count=marked,
stage=_ScanStage.PRUNING.value,
)
if marked > 0:
logging.info("Marked %d refs as missing before scan", marked)
sync_temp_references_safely()
sync_temp_references_safely(scan_state)
if self._check_pause_and_cancel():
if self._check_pause_and_cancel(_ScanStage.PRUNING):
logging.info("Asset scan cancelled after pruning phase")
cancelled = True
return
@@ -574,7 +645,7 @@ class _AssetSeeder:
created, skipped, paths = self._run_fast_phase(roots)
total_created, skipped_existing, total_paths = created, skipped, paths
if self._check_pause_and_cancel():
if self._check_pause_and_cancel(_ScanStage.FAST_SCAN):
cancelled = True
return
@@ -590,7 +661,7 @@ class _AssetSeeder:
# Phase 2: Enrichment scan (metadata + hashes)
if phase in (ScanPhase.ENRICH, ScanPhase.FULL):
if self._check_pause_and_cancel():
if self._check_pause_and_cancel(_ScanStage.ENRICH):
cancelled = True
return
@@ -608,6 +679,14 @@ class _AssetSeeder:
},
)
# Deliberately non-blocking, unlike every other checkpoint: no work
# remains, so pausing here would hold the scan open with nothing to
# do until main.py's next resume (a whole gc_collect_interval away).
if self._is_cancelled():
self._record_cancel_stage(_ScanStage.FINALIZE)
cancelled = True
return
elapsed = time.perf_counter() - t_start
logging.info(
"Scan(%s, %s) done %.3fs: created=%d enriched=%d skipped=%d",
@@ -618,6 +697,18 @@ class _AssetSeeder:
total_enriched,
skipped_existing,
)
emit(
"seeder.scan_completed",
phase=phase.value,
elapsed_ms=round(elapsed * 1000),
created=total_created,
enriched=total_enriched,
skipped=skipped_existing,
hash_failed=scan_state.hash_failed,
enrich_failed=scan_state.enrich_failed,
permission_denied=scan_state.permission_denied,
root=root,
)
self._emit_event(
"assets.seed.completed",
@@ -634,35 +725,51 @@ class _AssetSeeder:
except Exception as e:
self._add_error(f"Scan failed: {e}")
logging.exception("Asset scan failed")
emit(
"seeder.scan_failed",
phase=phase.value,
error_type=error_type(e),
root=root,
)
self._emit_event("assets.seed.error", {"message": str(e)})
finally:
if cancelled:
self._emit_event(
"assets.seed.cancelled",
{
"scanned": self._progress.scanned if self._progress else 0,
"total": total_paths,
"created": total_created,
},
)
with self._lock:
start_paused = self._state is State.PAUSED
self._reset_to_idle()
pending = self._pending_scan
if pending is not None:
self._pending_scan = None
if not self.start(
roots=pending["roots"],
phase=pending["phase"],
prune_first=False,
compute_hashes=pending["compute_hashes"],
_start_paused=start_paused,
):
logging.warning(
"Pending scan could not start (roots=%s, phase=%s)",
pending["roots"],
pending["phase"].value,
try:
if cancelled:
stage = self._scan_state.cancel_stage if self._scan_state else None
if stage is not None:
emit(
"seeder.scan_cancelled",
phase=phase.value,
stage=stage,
root=root,
)
self._emit_event(
"assets.seed.cancelled",
{
"scanned": self._scan_state.scanned if self._scan_state else 0,
"total": total_paths,
"created": total_created,
},
)
finally:
with self._lock:
start_paused = self._state is State.PAUSED
self._reset_to_idle()
pending = self._pending_scan
if pending is not None:
self._pending_scan = None
if not self.start(
roots=pending["roots"],
phase=pending["phase"],
prune_first=False,
compute_hashes=pending["compute_hashes"],
_start_paused=start_paused,
):
logging.warning(
"Pending scan could not start (roots=%s, phase=%s)",
pending["roots"],
pending["phase"].value,
)
def _run_fast_phase(self, roots: tuple[RootType, ...]) -> tuple[int, int, int]:
"""Run phase 1: fast scan to create stub records.
@@ -676,17 +783,19 @@ class _AssetSeeder:
existing_paths: set[str] = set()
t_sync = time.perf_counter()
assert self._scan_state is not None
scan_state = self._scan_state
for r in roots:
if self._check_pause_and_cancel():
if self._check_pause_and_cancel(_ScanStage.FAST_SCAN):
return total_created, skipped_existing, 0
existing_paths.update(sync_root_safely(r))
existing_paths.update(sync_root_safely(r, scan_state))
logging.debug(
"Fast scan: sync_root phase took %.3fs (%d existing paths)",
time.perf_counter() - t_sync,
len(existing_paths),
)
if self._check_pause_and_cancel():
if self._check_pause_and_cancel(_ScanStage.FAST_SCAN):
return total_created, skipped_existing, 0
t_collect = time.perf_counter()
@@ -710,6 +819,7 @@ class _AssetSeeder:
paths,
existing_paths,
enable_metadata_extraction=False,
progress=scan_state,
)
logging.debug(
"Fast scan: build_asset_specs took %.3fs (%d specs, %d skipped)",
@@ -719,7 +829,7 @@ class _AssetSeeder:
)
self._update_progress(skipped=skipped_existing)
if self._check_pause_and_cancel():
if self._check_pause_and_cancel(_ScanStage.FAST_SCAN):
return total_created, skipped_existing, total_paths
batch_size = 500
@@ -727,7 +837,7 @@ class _AssetSeeder:
progress_interval = 1.0
for i in range(0, len(specs), batch_size):
if self._check_pause_and_cancel():
if self._check_pause_and_cancel(_ScanStage.FAST_SCAN):
logging.info(
"Fast scan cancelled after %d/%d files (created=%d)",
i,
@@ -744,6 +854,7 @@ class _AssetSeeder:
except Exception as e:
self._add_error(f"Batch insert failed at offset {i}: {e}")
logging.exception("Batch insert failed at offset %d", i)
emit("seeder.batch_insert_failed", error_type=error_type(e))
scanned = i + len(batch)
now = time.perf_counter()
@@ -781,6 +892,7 @@ class _AssetSeeder:
Tuple of (cancelled, total_enriched)
"""
total_enriched = 0
scan_state = self._scan_state
with create_session() as session:
drain_pending_verifications(session)
tick_watch_list(session)
@@ -803,7 +915,7 @@ class _AssetSeeder:
max_consecutive_empty = 3
while True:
if self._check_pause_and_cancel():
if self._check_pause_and_cancel(_ScanStage.ENRICH):
logging.info("Enrich scan cancelled after %d assets", total_enriched)
return True, total_enriched
@@ -826,6 +938,7 @@ class _AssetSeeder:
extract_metadata=True,
compute_hash=self._compute_hashes,
interrupt_check=self._is_paused_or_cancelled,
progress=scan_state,
)
total_enriched += enriched
skip_ids.update(failed_ids)
server.py
+3
View File
@@ -45,6 +45,7 @@ from comfyui_version import __version__
from app.frontend_management import FrontendManager, parse_version
from comfy_api.internal import _ComfyNodeInternal
from app.assets.services.asset_management import resolve_hash_to_path
from app.assets.event_log import emit
from app.user_manager import UserManager
from app.model_manager import ModelFileManager
@@ -256,6 +257,8 @@ class PromptServer():
logging.info(f"[Prompt Server] web root: {self.web_root}")
self.asset_manager.register_routes(self.app, self.user_manager)
self.asset_manager.set_event_sink(self.send_sync)
if self.asset_manager.enabled:
emit("assets.enabled", hashing_enabled=args.enable_asset_hashing)
routes = web.RouteTableDef()
self.routes = routes
self.last_node_id = None
tests-unit/assets_test/fixtures/assets_event_lines.txt
+3
View File
@@ -0,0 +1,3 @@
[assets-event] seeder.scan_completed created=12 elapsed_ms=8123 enrich_failed=0 enriched=4 hash_failed=2 permission_denied=0 phase=fast root=models skipped=3
[assets-event] seeder.scan_started phase=enrich
[assets-event] scanner.stat_failed error_type=PermissionError site=discovery
tests-unit/assets_test/test_event_log.py
+280
View File
@@ -0,0 +1,280 @@
"""Tests for the structured assets event log lines (``app/assets/event_log.py``)."""
import logging
import re
from pathlib import Path
import pytest
from app.assets import event_log
from app.assets.event_log import ALLOWED_FIELDS, TAG, EventLogError, emit, error_type
# The line grammar below is the CONTRACT shared with the desktop launcher's log
# tap: Comfy-Org/Comfy-Desktop `src/main/lib/assetsTap.ts` holds the equivalent
# regex, and `tests-unit/assets_test/fixtures/assets_event_lines.txt` is a
# byte-identical copy of that repo's `src/main/lib/__fixtures__/assets-event-lines.txt`.
# Neither side may change without the other.
EVENT_LINE_PATTERN = re.compile(
r"^\[assets-event\] (?P<event>[a-z][a-z0-9_]*(?:\.[a-z][a-z0-9_]*)*)"
r"(?P<fields>(?: [a-z_]+=[^ =]+)*)$"
)
FIXTURE_PATH = Path(__file__).parent / "fixtures" / "assets_event_lines.txt"
# One valid value per allowed field, covering every enum member so the desktop
# tap's mirrored validator matrix has a counterpart on this side.
VALID_VALUES: dict[str, list[object]] = {
"root": ["models", "input", "output", "user", "temp"],
"phase": ["fast", "enrich", "full"],
"stage": ["mark_missing", "pruning", "fast_scan", "enrich", "finalize"],
"elapsed_ms": [0, 8123],
"created": [0, 12],
"enriched": [4],
"skipped": [3],
"hash_failed": [2],
"enrich_failed": [0],
"permission_denied": [0],
"count": [1],
"error_type": ["ValueError", "FileNotFoundError"],
"hashing_enabled": [True, False],
"site": ["discovery", "enrich"],
}
@pytest.fixture(autouse=True)
def autoclean_unit_test_assets():
"""Shadow the conftest fixture of the same name.
The conftest version reaches a running server to delete test-tagged assets,
which transitively boots ComfyUI for every test in this directory. Nothing
here touches a server or creates an asset, so the boot is pure cost.
"""
yield
def fixture_lines() -> list[str]:
return FIXTURE_PATH.read_text(encoding="utf-8").splitlines()
def parse_fields(raw: str) -> dict[str, bool | int | str]:
fields: dict[str, bool | int | str] = {}
for pair in raw.split():
name, value = pair.split("=", maxsplit=1)
if value == "true":
fields[name] = True
elif value == "false":
fields[name] = False
elif value.removeprefix("-").isdigit():
fields[name] = int(value)
else:
fields[name] = value
return fields
def emit_line(caplog: pytest.LogCaptureFixture, event: str, **fields: object) -> str:
"""Emit one event and return the single tagged line it produced."""
caplog.clear()
with caplog.at_level(logging.INFO):
emit(event, **fields)
tagged = [r.getMessage() for r in caplog.records if r.getMessage().startswith(TAG)]
assert len(tagged) == 1, tagged
return tagged[0]
def go_to_production_mode(monkeypatch: pytest.MonkeyPatch) -> None:
"""Leave strict mode so invalid calls warn-and-drop instead of raising."""
monkeypatch.delenv("PYTEST_CURRENT_TEST", raising=False)
monkeypatch.delenv("COMFYUI_ASSETS_EVENT_LOG_STRICT", raising=False)
event_log._warned_call_sites.clear()
# --- the shared cross-repo fixture -------------------------------------------------
def test_shared_fixture_file_holds_three_newline_terminated_lines():
raw = FIXTURE_PATH.read_text(encoding="utf-8")
assert raw.endswith("\n")
assert len(raw.splitlines()) == 3
@pytest.mark.parametrize("line", fixture_lines())
def test_emit_reproduces_each_shared_fixture_line_byte_for_byte(caplog, line):
"""Given a canonical line, When its fields are re-emitted, Then the bytes match."""
match = EVENT_LINE_PATTERN.match(line)
assert match is not None, line
fields = parse_fields(match.group("fields"))
assert emit_line(caplog, match.group("event"), **fields) == line
# --- line shape ---------------------------------------------------------------------
def test_fields_are_serialized_as_sorted_logfmt(caplog):
line = emit_line(caplog, "seeder.scan_started", root="models", phase="fast")
assert line == "[assets-event] seeder.scan_started phase=fast root=models"
def test_a_fieldless_event_still_matches_the_shared_pattern(caplog):
line = emit_line(caplog, "scanner.hash_discarded_modified")
assert line == "[assets-event] scanner.hash_discarded_modified"
assert EVENT_LINE_PATTERN.match(line) is not None
def test_the_emitted_record_is_a_single_line(caplog):
line = emit_line(caplog, "seeder.scan_failed", error_type="ValueError")
assert "\n" not in line
assert "\r" not in line
# --- error_type ---------------------------------------------------------------------
def test_error_type_is_the_class_name_and_the_path_never_reaches_the_line(caplog):
exc = FileNotFoundError("/home/x/model.safetensors")
assert error_type(exc) == "FileNotFoundError"
line = emit_line(caplog, "seeder.scan_failed", error_type=error_type(exc))
assert "/home/x/model.safetensors" not in line
assert "model.safetensors" not in line
# --- the closed vocabulary ----------------------------------------------------------
def test_the_valid_value_matrix_covers_every_allowed_field():
assert set(VALID_VALUES) == set(ALLOWED_FIELDS)
@pytest.mark.parametrize(
("field", "value"),
[(field, value) for field, values in VALID_VALUES.items() for value in values],
)
def test_every_allowed_field_value_round_trips(caplog, field, value):
line = emit_line(caplog, "seeder.scan_completed", **{field: value})
match = EVENT_LINE_PATTERN.match(line)
assert match is not None, line
assert parse_fields(match.group("fields")) == {field: value}
def test_unknown_field_raises_under_pytest():
with pytest.raises(EventLogError):
emit("seeder.scan_started", path="/home/x/models")
@pytest.mark.parametrize("value", ["a/b", "a\\b", "a:b", "a b", "a=b", 'a"b'])
def test_a_string_value_carrying_a_forbidden_character_raises(value):
with pytest.raises(EventLogError):
emit("seeder.scan_failed", error_type=value)
@pytest.mark.parametrize(
("field", "value"),
[
("root", "checkpoints"),
("root", 1),
("phase", "quick"),
("phase", None),
("stage", "scanning"),
("site", "reference"),
("error_type", "x" * 65),
("error_type", ""),
("error_type", 7),
("elapsed_ms", "8123"),
("count", 1.5),
("created", True),
("hashing_enabled", 1),
("hashing_enabled", "true"),
],
ids=[
"bad-root",
"non-string-root",
"bad-phase",
"none-phase",
"bad-stage",
"bad-site",
"oversized-string",
"empty-string",
"non-string-error-type",
"string-into-int-field",
"float-into-int-field",
"bool-into-int-field",
"int-into-bool-field",
"string-into-bool-field",
],
)
def test_every_validator_rejects_its_bad_value(field, value):
with pytest.raises(EventLogError):
emit("seeder.scan_completed", **{field: value})
@pytest.mark.parametrize(
"event",
["", "Seeder.scan_started", "seeder..scan", "9seeder.scan", "seeder.scan-started", "seeder scan", "seeder.", "scanner.made_up"],
)
def test_an_invalid_event_name_raises(event):
with pytest.raises(EventLogError):
emit(event)
# --- strict mode vs production mode -------------------------------------------------
def test_the_env_var_enables_strict_mode_without_pytest(monkeypatch):
go_to_production_mode(monkeypatch)
monkeypatch.setenv("COMFYUI_ASSETS_EVENT_LOG_STRICT", "1")
with pytest.raises(EventLogError):
emit("seeder.scan_started", path="/home/x")
def test_an_env_var_value_other_than_1_is_not_strict(caplog, monkeypatch):
go_to_production_mode(monkeypatch)
monkeypatch.setenv("COMFYUI_ASSETS_EVENT_LOG_STRICT", "true")
with caplog.at_level(logging.WARNING):
emit("seeder.scan_started", path="/home/x")
assert [r for r in caplog.records if r.levelno == logging.WARNING]
def test_production_mode_warns_once_for_repeated_calls_from_one_call_site(caplog, monkeypatch):
go_to_production_mode(monkeypatch)
caplog.clear()
with caplog.at_level(logging.WARNING):
for _ in range(3):
emit("seeder.scan_started", path="/home/x/models")
warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
assert len(warnings) == 1
assert not [r for r in caplog.records if r.getMessage().startswith(TAG)]
assert "/home/x/models" not in caplog.text
def test_production_mode_warns_once_per_distinct_call_site(caplog, monkeypatch):
go_to_production_mode(monkeypatch)
caplog.clear()
with caplog.at_level(logging.WARNING):
emit("seeder.scan_started", path="/home/x/models")
emit("seeder.scan_started", path="/home/x/models")
assert len([r for r in caplog.records if r.levelno == logging.WARNING]) == 2
def test_production_mode_still_emits_valid_events_after_a_dropped_one(caplog, monkeypatch):
go_to_production_mode(monkeypatch)
with caplog.at_level(logging.INFO):
emit("seeder.scan_started", path="/home/x/models")
emit("seeder.scan_started", phase="fast")
tagged = [r.getMessage() for r in caplog.records if r.getMessage().startswith(TAG)]
assert tagged == ["[assets-event] seeder.scan_started phase=fast"]
tests-unit/assets_test/test_event_log_sites.py
+66
View File
@@ -0,0 +1,66 @@
import subprocess
import sys
from pathlib import Path
import pytest
from app.assets.event_log import TAG
STARTUP_SCRIPT = (
"import runpy, comfy_kitchen; "
"comfy_kitchen.int8_attention_is_available=lambda: False; "
'runpy.run_path("main.py", run_name="__main__")'
)
@pytest.fixture(autouse=True)
def autoclean_unit_test_assets():
yield
def run_quick_startup(tmp_path: Path, *flags: str) -> str:
result = subprocess.run(
[
sys.executable,
"-c",
STARTUP_SCRIPT,
"--cpu",
"--quick-test-for-ci",
"--disable-all-custom-nodes",
"--disable-api-nodes",
f"--base-directory={tmp_path}",
f"--front-end-root={tmp_path}",
f"--database-url=sqlite:///{tmp_path / 'assets.sqlite3'}",
*flags,
],
cwd=Path(__file__).resolve().parents[2],
capture_output=True,
text=True,
timeout=120,
check=True,
)
return result.stdout + result.stderr
@pytest.mark.parametrize(
("hashing_flag", "expected"),
[
pytest.param((), False, id="hashing-disabled"),
pytest.param(("--enable-asset-hashing",), True, id="hashing-enabled"),
],
)
def test_enabled_assets_emits_once_with_the_hashing_flag(
tmp_path: Path, hashing_flag: tuple[str, ...], expected: bool
) -> None:
output = run_quick_startup(tmp_path, "--enable-assets", *hashing_flag)
lines = [line for line in output.splitlines() if f"{TAG} assets.enabled " in line]
assert len(lines) == 1
assert f"hashing_enabled={str(expected).lower()}" in lines[0]
def test_noassets_emits_no_enabled_event(tmp_path: Path) -> None:
output = run_quick_startup(tmp_path)
assert f"{TAG} assets.enabled " not in output
tests-unit/assets_test/test_scanner_ingest_event_log.py
+436
View File
@@ -0,0 +1,436 @@
import logging
import re
from contextlib import nullcontext
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import Mock
import pytest
from app.assets import scanner
from app.assets.event_log import TAG
from app.assets.scanner import UnenrichedContent
from app.assets.seeder import _ScanState
EVENT_LINE_PATTERN = re.compile(
rf"^{re.escape(TAG)} (?P<event>[a-z][a-z0-9_]*(?:\.[a-z][a-z0-9_]*)*)"
r"(?P<fields>(?: [a-z_]+=[^ =]+)*)$"
)
EventFields = dict[str, bool | int | str]
@pytest.fixture(autouse=True)
def autoclean_unit_test_assets():
yield
def parse_fields(raw: str) -> EventFields:
fields: EventFields = {}
for pair in raw.split():
name, value = pair.split("=", maxsplit=1)
if value == "true":
fields[name] = True
elif value == "false":
fields[name] = False
elif value.removeprefix("-").isdigit():
fields[name] = int(value)
else:
fields[name] = value
return fields
def tagged_events(caplog: pytest.LogCaptureFixture) -> list[tuple[str, EventFields]]:
events: list[tuple[str, EventFields]] = []
for record in caplog.records:
match = EVENT_LINE_PATTERN.match(record.getMessage())
if match is not None:
events.append((match.group("event"), parse_fields(match.group("fields"))))
return events
def events_named(
caplog: pytest.LogCaptureFixture, event_name: str
) -> list[EventFields]:
return [fields for event, fields in tagged_events(caplog) if event == event_name]
def tagged_lines(caplog: pytest.LogCaptureFixture) -> list[str]:
return [record.getMessage() for record in caplog.records if record.getMessage().startswith(TAG)]
def hash_session(path: Path) -> Mock:
stat_result = path.stat()
content = SimpleNamespace(hash=None, mtime_ns=stat_result.st_mtime_ns)
record = SimpleNamespace(system_metadata=None, mime_type=None)
session = Mock()
session.get.side_effect = lambda _model, row_id: content if row_id == "content" else record
return session
def run_hash_failure(session: Mock, path: Path, progress: _ScanState) -> bool:
return scanner.enrich_asset(
session,
file_path=str(path),
content_id="content",
record_id="record",
extract_metadata=False,
compute_hash=True,
progress=progress,
)
@pytest.mark.parametrize(
("operation", "event_name", "expected_fields", "expected_result"),
[
pytest.param(
lambda: scanner.sync_root_safely("models"),
"scanner.fast_scan_failed",
{"error_type": "FileNotFoundError", "root": "models"},
set(),
id="fast-scan",
),
],
)
def test_scanner_safe_failures_emit_exception_type_without_path(
operation,
event_name: str,
expected_fields: EventFields,
expected_result,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
secret_path = "/private/assets/secret.safetensors"
def fail_session():
raise FileNotFoundError(secret_path)
monkeypatch.setattr(scanner, "create_session", fail_session)
with caplog.at_level(logging.INFO):
result = operation()
assert result == expected_result
assert events_named(caplog, event_name) == [expected_fields]
assert all(secret_path not in line for line in tagged_lines(caplog))
def test_permission_error_in_reference_sync_increments_scan_counter(
monkeypatch: pytest.MonkeyPatch,
) -> None:
secret_path = "/private/assets/unreadable.safetensors"
content = SimpleNamespace(id="content", path=secret_path)
progress = _ScanState()
def deny_stat(*_args, **_kwargs):
raise PermissionError(secret_path)
monkeypatch.setattr(scanner, "os", SimpleNamespace(stat=deny_stat, path=scanner.os.path))
monkeypatch.setattr(scanner, "live_contents_under_prefixes", lambda _session, _prefixes: [content])
scanner.sync_prefixes_with_filesystem(Mock(), ["/private/assets"], progress=progress)
assert progress.permission_denied == 1
def test_hash_failures_emit_once_per_scan_and_count_every_failure(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
path = tmp_path / "model.safetensors"
path.write_bytes(b"model")
progress = _ScanState()
def fail_hash(_path: str):
raise FileNotFoundError(str(path))
monkeypatch.setattr(scanner, "snapshot_hash", fail_hash)
session = hash_session(path)
with caplog.at_level(logging.INFO):
assert run_hash_failure(session, path, progress) is False
assert run_hash_failure(session, path, progress) is False
assert events_named(caplog, "scanner.hash_failed") == [
{"error_type": "FileNotFoundError"}
]
assert progress.hash_failed == 2
def test_hash_failure_first_occurrence_resets_with_new_scan(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
path = tmp_path / "model.safetensors"
path.write_bytes(b"model")
def fail_hash(_path: str):
raise OSError("hash unavailable")
monkeypatch.setattr(scanner, "snapshot_hash", fail_hash)
session = hash_session(path)
with caplog.at_level(logging.INFO):
run_hash_failure(session, path, _ScanState())
run_hash_failure(session, path, _ScanState())
assert events_named(caplog, "scanner.hash_failed") == [
{"error_type": "OSError"},
{"error_type": "OSError"},
]
def test_hash_failure_no_log_line_leaks_exception_path(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
# Both the tagged [assets-event] line and the privacy-safe
# "Asset scan error" line must omit the file path.
path = tmp_path / "private-model.safetensors"
path.write_bytes(b"model")
def fail_hash(_path: str):
raise FileNotFoundError(str(path))
monkeypatch.setattr(scanner, "snapshot_hash", fail_hash)
with caplog.at_level(logging.INFO):
run_hash_failure(hash_session(path), path, _ScanState())
assert events_named(caplog, "scanner.hash_failed") == [
{"error_type": "FileNotFoundError"}
]
assert any("Asset scan error" in record.getMessage() for record in caplog.records)
assert all(str(path) not in record.getMessage() for record in caplog.records)
assert all(str(path) not in line for line in tagged_lines(caplog))
def test_modified_during_hash_emits_fieldless_discard_event(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
path = tmp_path / "changing.safetensors"
path.write_bytes(b"model")
monkeypatch.setattr(scanner, "snapshot_hash", lambda _path: None)
with caplog.at_level(logging.INFO):
updated = scanner.enrich_asset(
hash_session(path),
file_path=str(path),
content_id="content",
record_id="record",
extract_metadata=False,
compute_hash=True,
progress=_ScanState(),
)
assert updated is False
assert events_named(caplog, "scanner.hash_discarded_modified") == [{}]
def test_modified_hash_event_emits_once_per_scan_and_resets(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
path = tmp_path / "changing.safetensors"
path.write_bytes(b"model")
monkeypatch.setattr(scanner, "snapshot_hash", lambda _path: None)
progress = _ScanState()
session = hash_session(path)
with caplog.at_level(logging.INFO):
assert run_hash_failure(session, path, progress) is False
assert run_hash_failure(session, path, progress) is False
assert events_named(caplog, "scanner.hash_discarded_modified") == [{}]
assert run_hash_failure(session, path, _ScanState()) is False
assert events_named(caplog, "scanner.hash_discarded_modified") == [{}, {}]
assert progress.hash_failed == 0
def test_locked_files_during_discovery_emit_stat_failed_exactly_once(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
def deny_stat(*_args, **_kwargs):
raise PermissionError("/private/assets/secret.bin")
monkeypatch.setattr(scanner, "os", SimpleNamespace(stat=deny_stat, path=scanner.os.path))
progress = _ScanState()
paths = [f"/private/assets/locked-{i}.bin" for i in range(3)]
with caplog.at_level(logging.INFO):
specs, _tag_pool, _skipped = scanner.build_asset_specs(paths, set(), progress=progress)
assert specs == []
assert events_named(caplog, "scanner.stat_failed") == [
{"error_type": "PermissionError", "site": "discovery"}
]
assert progress.permission_denied == 3
def test_missing_files_during_discovery_emit_nothing(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
def vanish_stat(*_args, **_kwargs):
raise FileNotFoundError("/private/assets/gone.bin")
monkeypatch.setattr(scanner, "os", SimpleNamespace(stat=vanish_stat, path=scanner.os.path))
progress = _ScanState()
with caplog.at_level(logging.INFO):
specs, _tag_pool, _skipped = scanner.build_asset_specs(
["/private/assets/gone.bin"], set(), progress=progress
)
assert specs == []
assert events_named(caplog, "scanner.stat_failed") == []
def test_discovery_stat_failure_emits_nothing_when_progress_is_none(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
def deny_stat(*_args, **_kwargs):
raise PermissionError("/private/assets/secret.bin")
monkeypatch.setattr(scanner, "os", SimpleNamespace(stat=deny_stat, path=scanner.os.path))
with caplog.at_level(logging.INFO):
scanner.build_asset_specs(["/private/assets/locked.bin"], set(), progress=None)
assert events_named(caplog, "scanner.stat_failed") == []
def test_locked_files_during_enrichment_emit_stat_failed_exactly_once(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
def deny_stat(*_args, **_kwargs):
raise PermissionError("/private/assets/secret.bin")
monkeypatch.setattr(scanner, "os", SimpleNamespace(stat=deny_stat, path=scanner.os.path))
progress = _ScanState()
with caplog.at_level(logging.INFO):
first = scanner.enrich_asset(
Mock(),
file_path="/private/assets/locked-1.bin",
content_id="content-1",
record_id="record-1",
extract_metadata=False,
compute_hash=False,
progress=progress,
)
second = scanner.enrich_asset(
Mock(),
file_path="/private/assets/locked-2.bin",
content_id="content-2",
record_id="record-2",
extract_metadata=False,
compute_hash=False,
progress=progress,
)
assert first is False
assert second is False
assert events_named(caplog, "scanner.stat_failed") == [
{"error_type": "PermissionError", "site": "enrich"}
]
assert progress.permission_denied == 2
def test_missing_file_during_enrichment_returns_false_without_emitting(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
def vanish_stat(*_args, **_kwargs):
raise FileNotFoundError("/private/assets/gone.bin")
monkeypatch.setattr(scanner, "os", SimpleNamespace(stat=vanish_stat, path=scanner.os.path))
progress = _ScanState()
with caplog.at_level(logging.INFO):
updated = scanner.enrich_asset(
Mock(),
file_path="/private/assets/gone.bin",
content_id="content",
record_id="record",
extract_metadata=False,
compute_hash=False,
progress=progress,
)
assert updated is False
assert events_named(caplog, "scanner.stat_failed") == []
def test_enrich_failures_emit_once_per_scan_and_reset_with_new_scan(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
rows = [
UnenrichedContent("content-1", "record-1", "/private/assets/one.bin"),
UnenrichedContent("content-2", "record-2", "/private/assets/two.bin"),
]
monkeypatch.setattr(scanner, "create_session", lambda: nullcontext(Mock()))
def fail_enrich(*_args, **_kwargs):
raise FileNotFoundError("/private/assets/secret.bin")
monkeypatch.setattr(scanner, "enrich_asset", fail_enrich)
with caplog.at_level(logging.INFO):
first_result = scanner.enrich_assets_batch(rows, progress=_ScanState())
second_result = scanner.enrich_assets_batch(rows[:1], progress=_ScanState())
assert first_result == (0, ["record-1", "record-2"])
assert second_result == (0, ["record-1"])
assert events_named(caplog, "scanner.enrich_failed") == [
{"error_type": "FileNotFoundError"},
{"error_type": "FileNotFoundError"},
]
def test_enrich_exception_counts_one_failure_per_raising_row(
monkeypatch: pytest.MonkeyPatch,
) -> None:
rows = [
UnenrichedContent("content-1", "record-1", "/private/assets/one.bin"),
UnenrichedContent("content-2", "record-2", "/private/assets/two.bin"),
]
monkeypatch.setattr(scanner, "create_session", lambda: nullcontext(Mock()))
def fail_enrich(*_args, **_kwargs):
raise FileNotFoundError("/private/assets/secret.bin")
monkeypatch.setattr(scanner, "enrich_asset", fail_enrich)
progress = _ScanState()
enriched, failed_ids = scanner.enrich_assets_batch(rows, progress=progress)
assert enriched == 0
assert failed_ids == ["record-1", "record-2"]
assert progress.enrich_failed == 2
def test_benign_enrich_no_op_is_skipped_without_counting_a_failure(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
deleted = tmp_path / "gone.safetensors"
rows = [UnenrichedContent("content-1", "record-1", str(deleted))]
monkeypatch.setattr(scanner, "create_session", lambda: nullcontext(Mock()))
progress = _ScanState()
enriched, failed_ids = scanner.enrich_assets_batch(rows, progress=progress)
assert enriched == 0
assert failed_ids == ["record-1"]
assert progress.enrich_failed == 0
tests-unit/seeder_test/test_seeder.py
+435
View File
@@ -1,8 +1,70 @@
import logging
import re
import threading
from contextlib import nullcontext
from unittest.mock import Mock
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from app.assets import seeder as seeder_module
from app.assets.database.models import Base
from app.assets.database.queries import create_content, create_record, mark_content_missing
from app.assets.event_log import TAG
from app.assets.seeder import ScanPhase, State, _AssetSeeder, _ScanStage, _ScanState
EVENT_LINE_PATTERN = re.compile(
rf"^{re.escape(TAG)} (?P<event>[a-z][a-z0-9_]*(?:\.[a-z][a-z0-9_]*)*)"
r"(?P<fields>(?: [a-z_]+=[^ =]+)*)$"
)
EventFields = dict[str, bool | int | str]
# Hang detector: a checkpoint that parks the scan fails the test, not the suite.
SCAN_JOIN_TIMEOUT = 5.0
@pytest.fixture
def scan_seeder(monkeypatch: pytest.MonkeyPatch) -> _AssetSeeder:
instance = _AssetSeeder()
instance._state = State.RUNNING
instance._scan_state = _ScanState()
instance._roots = ("models", "input")
instance._phase = ScanPhase.FULL
monkeypatch.setattr(seeder_module, "dependencies_available", lambda: True)
monkeypatch.setattr(instance, "_log_scan_config", lambda roots: None)
return instance
def parse_fields(raw: str) -> EventFields:
fields: EventFields = {}
for pair in raw.split():
name, value = pair.split("=", maxsplit=1)
if value == "true":
fields[name] = True
elif value == "false":
fields[name] = False
elif value.removeprefix("-").isdigit():
fields[name] = int(value)
else:
fields[name] = value
return fields
def tagged_events(caplog: pytest.LogCaptureFixture) -> list[tuple[str, EventFields]]:
events: list[tuple[str, EventFields]] = []
for record in caplog.records:
match = EVENT_LINE_PATTERN.match(record.getMessage())
if match is not None:
events.append((match.group("event"), parse_fields(match.group("fields"))))
return events
def events_named(
caplog: pytest.LogCaptureFixture, event_name: str
) -> list[EventFields]:
return [fields for event, fields in tagged_events(caplog) if event == event_name]
def test_seeder_models_missing_as_content_state():
@@ -22,3 +84,376 @@ def test_seeder_models_missing_as_content_state():
assert content.is_missing is True
assert record.content_id == content.id
def test_multi_root_scan_emits_one_started_and_completed_without_root(
scan_seeder: _AssetSeeder,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
clock = iter((10.0, 10.8126))
monkeypatch.setattr(seeder_module.time, "perf_counter", lambda: next(clock))
monkeypatch.setattr(scan_seeder, "_run_fast_phase", lambda roots: (3, 2, 5))
monkeypatch.setattr(scan_seeder, "_run_enrich_phase", lambda roots: (False, 4))
with caplog.at_level(logging.INFO):
scan_seeder._run_scan()
assert events_named(caplog, "seeder.scan_started") == [{"phase": "full"}]
completed = events_named(caplog, "seeder.scan_completed")
assert len(completed) == 1
assert completed[0] == {
"created": 3,
"elapsed_ms": 813,
"enrich_failed": 0,
"enriched": 4,
"hash_failed": 0,
"permission_denied": 0,
"phase": "full",
"skipped": 2,
}
def test_scan_completed_reports_per_scan_failure_counts(
scan_seeder: _AssetSeeder,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
scan_seeder._scan_state = _ScanState(
hash_failed=2,
enrich_failed=3,
permission_denied=1,
)
clock = iter((10.0, 10.5))
monkeypatch.setattr(seeder_module.time, "perf_counter", lambda: next(clock))
monkeypatch.setattr(scan_seeder, "_run_fast_phase", lambda roots: (0, 0, 0))
monkeypatch.setattr(scan_seeder, "_run_enrich_phase", lambda roots: (False, 0))
with caplog.at_level(logging.INFO):
scan_seeder._run_scan()
completed = events_named(caplog, "seeder.scan_completed")
assert len(completed) == 1
assert completed[0]["hash_failed"] == 2
assert completed[0]["enrich_failed"] == 3
assert completed[0]["permission_denied"] == 1
def test_enrich_phase_does_not_count_returned_ids_as_failures(
scan_seeder: _AssetSeeder,
monkeypatch: pytest.MonkeyPatch,
) -> None:
session = Mock()
batches = iter(
(
[
Mock(record_id="record-1"),
Mock(record_id="record-2"),
],
[],
)
)
monkeypatch.setattr(seeder_module, "create_session", lambda: nullcontext(session))
monkeypatch.setattr(seeder_module, "drain_pending_verifications", lambda _session: None)
monkeypatch.setattr(seeder_module, "tick_watch_list", lambda _session: None)
monkeypatch.setattr(seeder_module, "drain_transition_queue", lambda _session: None)
monkeypatch.setattr(
seeder_module,
"get_unenriched_assets_for_roots",
lambda *_args, **_kwargs: next(batches),
)
monkeypatch.setattr(
seeder_module,
"enrich_assets_batch",
lambda *_args, **_kwargs: (0, ["record-1", "record-2"]),
)
monkeypatch.setattr(scan_seeder, "_check_pause_and_cancel", lambda _stage: False)
cancelled, enriched = scan_seeder._run_enrich_phase(("models",))
assert cancelled is False
assert enriched == 0
assert scan_seeder._scan_state is not None
assert scan_seeder._scan_state.enrich_failed == 0
def test_starting_a_scan_installs_fresh_per_scan_failure_state(
monkeypatch: pytest.MonkeyPatch,
) -> None:
instance = _AssetSeeder()
instance._scan_state = _ScanState(
hash_failed=7,
enrich_failed=4,
permission_denied=2,
)
instance._scan_state.mark_emitted("enrich_failed")
monkeypatch.setattr(instance, "_run_scan", lambda: None)
started = instance.start(roots=("models",), phase=ScanPhase.FAST)
assert started is True
assert instance._thread is not None
instance._thread.join(timeout=5)
assert instance._scan_state == _ScanState()
def test_single_root_scan_emits_root_and_phase(
scan_seeder: _AssetSeeder,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
scan_seeder._roots = ("output",)
scan_seeder._phase = ScanPhase.FAST
monkeypatch.setattr(scan_seeder, "_run_fast_phase", lambda roots: (0, 0, 0))
with caplog.at_level(logging.INFO):
scan_seeder._run_scan()
assert events_named(caplog, "seeder.scan_started") == [
{"phase": "fast", "root": "output"}
]
completed = events_named(caplog, "seeder.scan_completed")
assert len(completed) == 1
assert completed[0]["phase"] == "fast"
assert completed[0]["root"] == "output"
def test_dependency_failure_emits_no_tagged_scan_lifecycle_lines(
scan_seeder: _AssetSeeder,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
monkeypatch.setattr(seeder_module, "dependencies_available", lambda: False)
with caplog.at_level(logging.INFO):
scan_seeder._run_scan()
assert [
event for event, _fields in tagged_events(caplog) if event.startswith("seeder.scan_")
] == []
def test_scan_failure_emits_exception_type_without_message(
scan_seeder: _AssetSeeder,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
scan_seeder._roots = ("models",)
scan_seeder._phase = ScanPhase.ENRICH
def fail_scan(_roots: tuple[str, ...]) -> None:
raise FileNotFoundError("/private/models/secret.safetensors")
monkeypatch.setattr(scan_seeder, "_log_scan_config", fail_scan)
with caplog.at_level(logging.INFO):
scan_seeder._run_scan()
assert events_named(caplog, "seeder.scan_failed") == [
{"error_type": "FileNotFoundError", "phase": "enrich", "root": "models"}
]
tagged = "\n".join(record.getMessage() for record in caplog.records if TAG in record.getMessage())
assert "/private/models/secret.safetensors" not in tagged
@pytest.mark.parametrize(
("stage", "phase"),
[
pytest.param("pruning", ScanPhase.FAST, id="pruning"),
pytest.param("fast_scan", ScanPhase.FAST, id="fast-scan"),
pytest.param("enrich", ScanPhase.ENRICH, id="enrich"),
pytest.param("finalize", ScanPhase.ENRICH, id="finalize"),
],
)
def test_scan_cancellation_emits_the_checkpoint_stage(
stage: str,
phase: ScanPhase,
scan_seeder: _AssetSeeder,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
scan_seeder._phase = phase
original_check = scan_seeder._check_pause_and_cancel
def cancel_at_stage(checkpoint_stage) -> bool:
if checkpoint_stage.value == stage:
scan_seeder._cancel_event.set()
return original_check(checkpoint_stage)
def run_enrich(roots) -> tuple[bool, int]:
# The finalize checkpoint is non-blocking and never routes through
# _check_pause_and_cancel, so its cancel has to land before it.
if stage == "finalize":
scan_seeder._cancel_event.set()
return (False, 0)
monkeypatch.setattr(scan_seeder, "_check_pause_and_cancel", cancel_at_stage)
monkeypatch.setattr(scan_seeder, "_run_fast_phase", lambda roots: (0, 0, 0))
monkeypatch.setattr(scan_seeder, "_run_enrich_phase", run_enrich)
with caplog.at_level(logging.INFO):
scan_seeder._run_scan()
assert events_named(caplog, "seeder.scan_cancelled") == [
{"phase": phase.value, "stage": stage}
]
def test_idle_reset_survives_a_raising_cancellation_emit(
scan_seeder: _AssetSeeder,
monkeypatch: pytest.MonkeyPatch,
) -> None:
original_check = scan_seeder._check_pause_and_cancel
def cancel_at_pruning(stage) -> bool:
if stage == _ScanStage.PRUNING:
scan_seeder._cancel_event.set()
return original_check(stage)
monkeypatch.setattr(scan_seeder, "_check_pause_and_cancel", cancel_at_pruning)
monkeypatch.setattr(seeder_module, "get_owned_prefixes", lambda: ())
monkeypatch.setattr(
seeder_module, "mark_missing_outside_prefixes_safely", lambda prefixes: 0
)
original_emit = seeder_module.emit
def raise_on_scan_cancelled(event, **kwargs):
if event == "seeder.scan_cancelled":
raise RuntimeError("event bus down")
return original_emit(event, **kwargs)
monkeypatch.setattr(seeder_module, "emit", raise_on_scan_cancelled)
with pytest.raises(RuntimeError, match="event bus down"):
scan_seeder._run_scan()
assert scan_seeder._state is State.IDLE
assert scan_seeder._scan_state is None
assert scan_seeder.mark_missing_outside_prefixes() == 0
def test_scan_paused_after_its_last_phase_still_completes(
scan_seeder: _AssetSeeder,
monkeypatch: pytest.MonkeyPatch,
) -> None:
scan_seeder._phase = ScanPhase.ENRICH
monkeypatch.setattr(seeder_module, "get_owned_prefixes", lambda: ())
monkeypatch.setattr(
seeder_module, "mark_missing_outside_prefixes_safely", lambda prefixes: 0
)
def pause_while_finishing(roots) -> tuple[bool, int]:
scan_seeder.pause()
return (False, 0)
monkeypatch.setattr(scan_seeder, "_run_enrich_phase", pause_while_finishing)
events: list[str] = []
scan_seeder.set_event_sink(lambda event_type, data: events.append(event_type))
scan = threading.Thread(target=scan_seeder._run_scan, daemon=True)
scan.start()
try:
scan.join(timeout=SCAN_JOIN_TIMEOUT)
assert scan.is_alive() is False, "paused scan parked at the finalize checkpoint"
assert "assets.seed.completed" in events
assert "assets.seed.paused" not in events
assert scan_seeder.mark_missing_outside_prefixes() == 0
finally:
scan_seeder._run_gate.set()
scan.join(timeout=SCAN_JOIN_TIMEOUT)
def test_enrich_interrupt_records_the_enrich_cancellation_stage(
scan_seeder: _AssetSeeder,
) -> None:
scan_seeder._cancel_event.set()
assert scan_seeder._is_paused_or_cancelled() is True
assert scan_seeder._scan_state is not None
assert scan_seeder._scan_state.cancel_stage is not None
assert scan_seeder._scan_state.cancel_stage == "enrich"
def test_prune_before_scan_emits_marked_missing_with_pruning_stage(
scan_seeder: _AssetSeeder,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
scan_seeder._prune_first = True
scan_seeder._phase = ScanPhase.FAST
monkeypatch.setattr(seeder_module, "get_owned_prefixes", lambda: ())
monkeypatch.setattr(
seeder_module, "mark_missing_outside_prefixes_safely", lambda prefixes: 5
)
monkeypatch.setattr(
seeder_module, "sync_temp_references_safely", lambda _progress: None
)
monkeypatch.setattr(scan_seeder, "_run_fast_phase", lambda roots: (0, 0, 0))
with caplog.at_level(logging.INFO):
scan_seeder._run_scan()
assert events_named(caplog, "seeder.marked_missing") == [
{"count": 5, "stage": "pruning"}
]
def test_standalone_mark_missing_emits_count_with_mark_missing_stage(
scan_seeder: _AssetSeeder,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
scan_seeder._state = State.IDLE
monkeypatch.setattr(seeder_module, "get_owned_prefixes", lambda: ())
monkeypatch.setattr(
seeder_module, "mark_missing_outside_prefixes_safely", lambda prefixes: 7
)
with caplog.at_level(logging.INFO):
result = scan_seeder.mark_missing_outside_prefixes()
assert result == 7
assert events_named(caplog, "seeder.marked_missing") == [
{"count": 7, "stage": "mark_missing"}
]
def test_batch_insert_failure_emits_only_the_exception_type(
scan_seeder: _AssetSeeder,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
session = Mock()
monkeypatch.setattr(
seeder_module, "sync_root_safely", lambda _root, _progress: set()
)
monkeypatch.setattr(
seeder_module, "collect_paths_for_roots", lambda roots: ["asset.safetensors"]
)
monkeypatch.setattr(
seeder_module,
"build_asset_specs",
lambda paths, existing_paths, enable_metadata_extraction, progress=None: (
[{"tags": []}],
{},
0,
),
)
def fail_insert(batch, batch_tags) -> int:
raise PermissionError("/private/models/asset.safetensors")
monkeypatch.setattr(seeder_module, "insert_asset_specs", fail_insert)
monkeypatch.setattr(seeder_module, "create_session", lambda: nullcontext(session))
monkeypatch.setattr(seeder_module, "tick_watch_list", lambda current_session: None)
with caplog.at_level(logging.INFO):
scan_seeder._run_fast_phase(("models",))
assert events_named(caplog, "seeder.batch_insert_failed") == [
{"error_type": "PermissionError"}
]
tagged = "\n".join(record.getMessage() for record in caplog.records if TAG in record.getMessage())
assert "/private/models/asset.safetensors" not in tagged
tests-unit/test_assets_event_log_static.py
+260
View File
@@ -0,0 +1,260 @@
"""Static discipline check for the ``[assets-event]`` log lines.
Pure :mod:`ast` analysis: no module under ``app/assets`` is imported or
executed, so this check never needs a running ComfyUI. It deliberately lives at
the ``tests-unit`` root rather than under ``tests-unit/assets_test/``, whose
autouse fixture boots a ComfyUI subprocess for every test in that subtree.
Four rules are enforced over every emit call site:
a. keyword fields come from the closed vocabulary, the event is a string literal
b. no other log line anywhere carries the tag, so the tap only ever sees emits
c. the call sites present in the tree match an explicit manifest
d. ``error_type=`` values come from ``event_log.error_type()``, never a string
"""
from __future__ import annotations
import ast
from collections import Counter
from collections.abc import Iterator
from pathlib import Path
from typing import NamedTuple
from app.assets.event_log import ALLOWED_EVENTS, ALLOWED_FIELDS, TAG
REPO_ROOT = Path(__file__).resolve().parents[1]
MODULE_SCOPE = "<module>"
EVENT_LOG_NAME = "event_log"
LOG_METHODS = frozenset({"debug", "info", "warning", "error", "exception", "critical", "log"})
FUNCTION_NODES = (ast.FunctionDef, ast.AsyncFunctionDef)
class CallSite(NamedTuple):
"""The identity of one emit call: file, enclosing function, event."""
path: str
function: str
event: str
# The manifest of every tagged event this branch emits: (file, enclosing
# function, event) triples that must be present in the tree exactly as written.
EXPECTED_CALL_SITES: frozenset[CallSite] = frozenset(
{
# todo 10 - seeder lifecycle + the single assets.enabled site
CallSite("server.py", "__init__", "assets.enabled"),
CallSite("app/assets/seeder.py", "_run_scan", "seeder.scan_started"),
CallSite("app/assets/seeder.py", "_run_scan", "seeder.scan_completed"),
CallSite("app/assets/seeder.py", "_run_scan", "seeder.scan_failed"),
CallSite("app/assets/seeder.py", "_run_scan", "seeder.scan_cancelled"),
CallSite("app/assets/seeder.py", "_run_scan", "seeder.marked_missing"),
CallSite("app/assets/seeder.py", "mark_missing_outside_prefixes", "seeder.marked_missing"),
CallSite("app/assets/seeder.py", "_run_fast_phase", "seeder.batch_insert_failed"),
# todo 11 - scanner failure paths
CallSite("app/assets/scanner.py", "sync_root_safely", "scanner.fast_scan_failed"),
CallSite("app/assets/scanner.py", "sync_temp_references_safely", "scanner.temp_sync_failed"),
CallSite(
"app/assets/scanner.py", "mark_missing_outside_prefixes_safely", "scanner.mark_missing_failed"
),
CallSite("app/assets/scanner.py", "enrich_asset", "scanner.hash_failed"),
CallSite("app/assets/scanner.py", "enrich_asset", "scanner.hash_discarded_modified"),
CallSite("app/assets/scanner.py", "enrich_assets_batch", "scanner.enrich_failed"),
# todo 16 - discovery/enrich stat failures, emit-once per scan per site
CallSite("app/assets/scanner.py", "build_asset_specs", "scanner.stat_failed"),
CallSite("app/assets/scanner.py", "enrich_asset", "scanner.stat_failed"),
}
)
class Aliases(NamedTuple):
"""The names one module binds to the event_log module and its functions."""
module: frozenset[str]
emit: frozenset[str]
error_type: frozenset[str]
class Scan(NamedTuple):
"""Everything the AST walk learned about the tree."""
files: tuple[str, ...]
call_sites: Counter[CallSite]
vocabulary: tuple[str, ...]
event_names: tuple[str, ...]
error_types: tuple[str, ...]
tagged_logs: tuple[str, ...]
def _scanned_files(root: Path) -> tuple[str, ...]:
"""Every assets module, plus server.py for its single assets.enabled emit."""
assets = sorted(p.relative_to(root).as_posix() for p in root.glob("app/assets/**/*.py"))
return (*assets, "server.py")
def _scoped_nodes(tree: ast.Module) -> Iterator[tuple[ast.AST, str]]:
"""Yield every node paired with the name of its innermost enclosing function."""
def walk(node: ast.AST, scope: str) -> Iterator[tuple[ast.AST, str]]:
for child in ast.iter_child_nodes(node):
child_scope = child.name if isinstance(child, FUNCTION_NODES) else scope
yield child, child_scope
yield from walk(child, child_scope)
yield from walk(tree, MODULE_SCOPE)
def _resolve_aliases(tree: ast.Module) -> Aliases:
module: set[str] = set()
emit: set[str] = set()
error_type: set[str] = set()
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for alias in node.names:
if alias.name.split(".")[-1] == EVENT_LOG_NAME:
module.add(alias.asname or alias.name)
elif isinstance(node, ast.ImportFrom):
from_event_log = (node.module or "").rsplit(".", 1)[-1] == EVENT_LOG_NAME
for alias in node.names:
if from_event_log and alias.name == "emit":
emit.add(alias.asname or alias.name)
elif from_event_log and alias.name == "error_type":
error_type.add(alias.asname or alias.name)
elif not from_event_log and alias.name == EVENT_LOG_NAME:
module.add(alias.asname or alias.name)
return Aliases(frozenset(module), frozenset(emit), frozenset(error_type))
def _is_emit_call(func: ast.expr, aliases: Aliases) -> bool:
if isinstance(func, ast.Attribute) and func.attr == "emit" and isinstance(func.value, ast.Name):
return func.value.id in aliases.module
return isinstance(func, ast.Name) and func.id in aliases.emit
def _is_error_type_call(value: ast.expr, aliases: Aliases) -> bool:
"""True only for a call to the sanctioned event_log.error_type()."""
if not isinstance(value, ast.Call):
return False
func = value.func
if isinstance(func, ast.Attribute) and func.attr == "error_type" and isinstance(func.value, ast.Name):
return func.value.id in aliases.module
return isinstance(func, ast.Name) and func.id in aliases.error_type
def _carries_tag(node: ast.Call) -> bool:
return any(
isinstance(child, ast.Constant) and isinstance(child.value, str) and TAG in child.value
for child in ast.walk(node)
)
def _is_log_call(func: ast.expr) -> bool:
return isinstance(func, ast.Attribute) and func.attr in LOG_METHODS
def _event_of(call: ast.Call) -> str | None:
"""The literal event name, or None when it is not a plain string literal."""
if len(call.args) != 1:
return None
first = call.args[0]
if not isinstance(first, ast.Constant) or not isinstance(first.value, str):
return None
return first.value
def _field_faults(call: ast.Call, aliases: Aliases) -> Iterator[tuple[str, str]]:
"""(category, reason) for every keyword that breaks rule (a) or rule (d)."""
for keyword in call.keywords:
if keyword.arg is None:
yield "vocabulary", "**splat fields cannot be checked statically"
elif keyword.arg not in ALLOWED_FIELDS:
yield "vocabulary", f"field {keyword.arg!r} is not in ALLOWED_FIELDS"
elif keyword.arg == "error_type" and not _is_error_type_call(keyword.value, aliases):
yield (
"error_types",
"error_type= must be a call to event_log.error_type(), got "
f"{ast.unparse(keyword.value)!r}",
)
def _file_faults(call: ast.Call, aliases: Aliases) -> Iterator[tuple[str, str]]:
if _is_emit_call(call.func, aliases):
event = _event_of(call)
if event not in ALLOWED_EVENTS:
yield "event_names", "the event must be one string literal in the allowed vocabulary"
yield from _field_faults(call, aliases)
elif _is_log_call(call.func) and _carries_tag(call):
yield "tagged_logs", f"log line carries {TAG} outside event_log.emit()"
def _scan_file(root: Path, relative: str) -> tuple[Counter[CallSite], list[tuple[str, str]]]:
tree = ast.parse((root / relative).read_text(encoding="utf-8"), filename=relative)
aliases = _resolve_aliases(tree)
sites: Counter[CallSite] = Counter()
faults: list[tuple[str, str]] = []
for node, scope in _scoped_nodes(tree):
if not isinstance(node, ast.Call):
continue
if _is_emit_call(node.func, aliases):
event = _event_of(node)
if event is not None and event in ALLOWED_EVENTS:
sites[CallSite(relative, scope, event)] += 1
for category, reason in _file_faults(node, aliases):
faults.append((category, f"{relative}:{node.lineno}: {reason}"))
return sites, faults
def scan_repository(root: Path = REPO_ROOT) -> Scan:
files = _scanned_files(root)
sites: Counter[CallSite] = Counter()
found: dict[str, list[str]] = {"vocabulary": [], "event_names": [], "error_types": [], "tagged_logs": []}
for relative in files:
file_sites, faults = _scan_file(root, relative)
sites += file_sites
for category, message in faults:
found[category].append(message)
return Scan(
files=files,
call_sites=sites,
vocabulary=tuple(found["vocabulary"]),
event_names=tuple(found["event_names"]),
error_types=tuple(found["error_types"]),
tagged_logs=tuple(found["tagged_logs"]),
)
SCAN = scan_repository()
def test_the_walk_actually_covers_the_assets_tree() -> None:
"""Guards every other check: a broken glob would make them all vacuous."""
assert "app/assets/event_log.py" in SCAN.files
assert "app/assets/seeder.py" in SCAN.files
assert "server.py" in SCAN.files
assert len(SCAN.files) > 20
def test_emit_fields_stay_inside_the_closed_vocabulary() -> None:
assert SCAN.vocabulary == ()
def test_emit_events_are_literals_in_the_allowed_vocabulary() -> None:
assert SCAN.event_names == ()
def test_error_type_values_come_from_event_log_error_type() -> None:
assert SCAN.error_types == ()
def test_no_other_log_line_carries_the_event_tag() -> None:
assert SCAN.tagged_logs == ()
def test_call_sites_match_the_manifest() -> None:
manifest = Counter(EXPECTED_CALL_SITES)
unexpected = SCAN.call_sites - manifest
missing = manifest - SCAN.call_sites
assert not unexpected, (
f"emit call sites not in the manifest: {sorted(unexpected)} — add them to "
"EXPECTED_CALL_SITES"
)
assert not missing, f"manifest call sites absent from the tree: {sorted(missing)}"