diff --git a/INSTALLATION_GUIDE.md b/INSTALLATION_GUIDE.md index a35ba255..27b4f59b 100644 --- a/INSTALLATION_GUIDE.md +++ b/INSTALLATION_GUIDE.md @@ -245,6 +245,31 @@ vidxp search speech "the bread just came out of the oven" Add `--media-id ` to a search command to restrict results to one video. Without it, VidXP searches all indexed videos in the active repository. +### Index more than one video + +Index every registered video that the active index does not already cover: + +```bash +vidxp index bulk --all --modality scene +``` + +Videos with matching content and indexing settings are skipped. To select specific +videos, pass their media IDs after `bulk` instead of `--all`. Add `--plan-only` +to preview the decisions without indexing, or `--reindex` to rebuild matching +videos. Bulk indexing accepts the same sampling and capability options as +`index create`. + +Each video uses its own indexing job. If one fails, the others continue and +successful results remain available. Rerun the same command to retry missing +results, or use `vidxp jobs retry ` with the failed job ID from the +summary. `--detach` returns after submission and reports job IDs; it does not +wait for indexing to succeed. + +The existing repository rule still applies: videos in one active index must +use the same indexing settings. Bulk indexing does not migrate an existing +multi-video index to a different profile or remove old results to make room +for one. Such jobs report the existing profile-compatibility error. + ### Start an installed interface | Interface | Command | diff --git a/src/vidxp/application.py b/src/vidxp/application.py index b8462ac5..71b40d2d 100644 --- a/src/vidxp/application.py +++ b/src/vidxp/application.py @@ -1,5 +1,7 @@ from __future__ import annotations +from dataclasses import replace + from contextlib import contextmanager from pathlib import Path from shutil import which @@ -235,36 +237,11 @@ def create_index( execution: ExecutionContext | None = None, ) -> IndexResult: active_execution = execution_context(execution) - selected = self.registry.validate_names(command.modalities) - non_indexable = [ - name for name in selected if self.registry.get(name).collection_name is None - ] - if non_indexable: - raise CapabilityRequestError( - "One or more selected capabilities do not support indexing." - ) + config = replace(self._index_config(command, media_id=command.media_id), device=self.device) + selected = config.enabled_modalities media = self.media.require_record(command.media_id) content = self.media.content(command.media_id) self.layout.ensure_local_directories() - capability_options = { - name: dict(options) for name, options in command.capability_options.items() - } - if command.scene_sample_fps is not None: - capability_options.setdefault("scene", {})["sample_fps"] = ( - command.scene_sample_fps - ) - config = IndexConfig.local( - video_id=command.media_id, - enabled_modalities=selected, - frame_stride=command.frame_stride, - storage_directory=self.index_directory, - collection_names=self.registry.collection_names(selected), - capability_options=self.registry.validate_options( - selected, - capability_options, - ), - device=self.device, - ) with self.runtime.scheduler.indexing(): with self._capability_dependencies(selected): result = self.index_backend.create( diff --git a/src/vidxp/application_models.py b/src/vidxp/application_models.py index f029bfae..257edcf8 100644 --- a/src/vidxp/application_models.py +++ b/src/vidxp/application_models.py @@ -562,13 +562,7 @@ class MediaUploadSessionStatus(ApplicationModel): next_action: str = Field(min_length=1, max_length=1024) -class CreateIndexCommand(ApplicationModel): - media_id: MediaId = Field( - description=( - "Stable identifier returned by list_media, get_media, or a " - "completed upload." - ) - ) +class IndexOptions(ApplicationModel): modalities: tuple[str, ...] frame_stride: int = Field( default=1, @@ -628,12 +622,21 @@ def _canonicalize_scene_sampling(cls, value: Any) -> Any: return payload @model_validator(mode="after") - def _scene_sampling_requires_scene(self) -> "CreateIndexCommand": + def _scene_sampling_requires_scene(self) -> "IndexOptions": if self.scene_sample_fps is not None and "scene" not in self.modalities: raise ValueError("scene_sample_fps requires the scene modality.") return self +class CreateIndexCommand(IndexOptions): + media_id: MediaId = Field( + description=( + "Stable identifier returned by list_media, get_media, or a " + "completed upload." + ) + ) + + class IndexResult(ApplicationModel): media_id: MediaId generation_id: IndexGenerationId @@ -642,6 +645,86 @@ class IndexResult(ApplicationModel): record_counts: dict[str, NonNegativeInt] = Field(default_factory=dict) +class BulkIndexTargetState(StrEnum): + pending = "pending" + skipped = "skipped" + + +class BulkIndexSkipReason(StrEnum): + already_indexed = "already_indexed" + media_not_ready = "media_not_ready" + + +class BulkIndexTarget(ApplicationModel): + media_id: MediaId + original_filename: str = Field(min_length=1) + state: BulkIndexTargetState + reason: BulkIndexSkipReason | None = Field( + default=None, + description="Why the media was skipped. Absent for pending targets.", + ) + generation_id: IndexGenerationId | None = Field( + default=None, + description=( + "Generation already covering this media in the active snapshot." + ), + ) + + @model_validator(mode="after") + def _reason_matches_state(self) -> "BulkIndexTarget": + if self.state == BulkIndexTargetState.skipped and self.reason is None: + raise ValueError("A skipped target requires a reason.") + if self.state == BulkIndexTargetState.pending and self.reason is not None: + raise ValueError("A pending target cannot carry a skip reason.") + return self + + +class PlanBulkIndexCommand(IndexOptions): + media_ids: tuple[MediaId, ...] = Field( + default=(), + description=( + "Registered media to consider. Empty selects every registered " + "media item in the repository." + ), + ) + modalities: tuple[str, ...] + reindex: bool = Field( + default=False, + description=( + "Plan already-indexed media for indexing instead of skipping it." + ), + ) + + @field_validator("media_ids") + @classmethod + def _unique_media_ids(cls, value: tuple[str, ...]) -> tuple[str, ...]: + if len(set(value)) != len(value): + raise ValueError("media_ids must not repeat a media identifier.") + return value + + +class BulkIndexPlan(ApplicationModel): + options: IndexOptions + targets: tuple[BulkIndexTarget, ...] = () + modalities: tuple[str, ...] + + @property + def pending(self) -> tuple[BulkIndexTarget, ...]: + return tuple( + target + for target in self.targets + if target.state == BulkIndexTargetState.pending + ) + + @property + def skipped(self) -> tuple[BulkIndexTarget, ...]: + return tuple( + target + for target in self.targets + if target.state == BulkIndexTargetState.skipped + ) + + class RemoveIndexCommand(ApplicationModel): media_id: MediaId diff --git a/src/vidxp/bulk_indexing.py b/src/vidxp/bulk_indexing.py new file mode 100644 index 00000000..db363fcb --- /dev/null +++ b/src/vidxp/bulk_indexing.py @@ -0,0 +1,182 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Callable, Sequence + +from vidxp.application_models import ( + ApplicationError, + CreateIndexCommand, + ListMediaCommand, + MediaAsset, + PlanBulkIndexCommand, + BulkIndexPlan, + BulkIndexTargetState, + MediaState, +) + +if TYPE_CHECKING: + from vidxp.application import VidXPApplication + from vidxp.control_plane import ControlPlaneApplication + from vidxp.job_service import JobService + + +@dataclass(frozen=True) +class BulkIndexItemResult: + media_id: str + filename: str + status: str + job_id: str | None = None + error_code: str | None = None + error_message: str | None = None + + +@dataclass(frozen=True) +class BulkIndexSummary: + total: int + indexed: int + skipped: int + failed: int + queued: int = 0 + results: tuple[BulkIndexItemResult, ...] = () + + +def _resolve_all_media( + application: VidXPApplication | ControlPlaneApplication, +) -> list[MediaAsset]: + media_list: list[MediaAsset] = [] + cursor: str | None = None + while True: + page = application.list_media( + ListMediaCommand( + page_size=100, + cursor=cursor, + state=MediaState.ready, + ) + ) + media_list.extend(page.items) + if not page.next_cursor or not page.items: + break + cursor = page.next_cursor + return media_list + + +def run_bulk_index( + application: VidXPApplication | ControlPlaneApplication, + jobs: JobService, + media_ids: Sequence[str] | None = None, + *, + all_eligible: bool = False, + skip_indexed: bool = True, + detach: bool = False, + plan: BulkIndexPlan | None = None, + modalities: Sequence[str] | None = None, + frame_stride: int = 1, + scene_sample_fps: float | None = None, + capability_options: dict[str, dict] | None = None, + on_item_start: Callable[[str, str], None] | None = None, + on_item_progress: Callable[[str, Any], None] | None = None, + on_item_complete: Callable[[BulkIndexItemResult], None] | None = None, +) -> BulkIndexSummary: + if plan is None: + if bool(media_ids) == all_eligible: + raise ValueError("Provide media IDs or all_eligible, not both.") + selected = application.select_index_modalities( + tuple(modalities) if modalities is not None else None + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand( + media_ids=tuple(media_ids or ()), + modalities=selected, + reindex=not skip_indexed, + frame_stride=frame_stride, + scene_sample_fps=scene_sample_fps, + capability_options=capability_options or {}, + ) + ) + + results: list[BulkIndexItemResult] = [] + + for target in plan.targets: + media_id, filename = target.media_id, target.original_filename + if target.state == BulkIndexTargetState.skipped: + item_result = BulkIndexItemResult( + media_id=media_id, + filename=filename, + status="skipped", + error_message=target.reason.value if target.reason else None, + ) + results.append(item_result) + if on_item_complete is not None: + on_item_complete(item_result) + continue + + if on_item_start is not None: + on_item_start(media_id, filename) + + command = CreateIndexCommand( + media_id=media_id, **plan.options.model_dump(mode="python") + ) + job_id = None + + try: + job = jobs.submit_index(command) + job_id = job.job_id + if detach: + item_result = BulkIndexItemResult( + media_id=media_id, + filename=filename, + status="queued", + job_id=job.job_id, + ) + results.append(item_result) + if on_item_complete is not None: + on_item_complete(item_result) + else: + + def _progress(current: Any) -> None: + if on_item_progress is not None: + on_item_progress(media_id, current) + + job = jobs.wait(job.job_id, progress=_progress) + item_result = BulkIndexItemResult( + media_id=media_id, + filename=filename, + status="indexed", + job_id=job.job_id, + ) + results.append(item_result) + if on_item_complete is not None: + on_item_complete(item_result) + except ApplicationError as exc: + item_result = BulkIndexItemResult( + media_id=media_id, + filename=filename, + status="failed", + job_id=job_id, + error_code=exc.code, + error_message=str(exc), + ) + results.append(item_result) + if on_item_complete is not None: + on_item_complete(item_result) + except Exception as exc: + item_result = BulkIndexItemResult( + media_id=media_id, + filename=filename, + status="failed", + job_id=job_id, + error_code="unexpected_error", + error_message=str(exc), + ) + results.append(item_result) + if on_item_complete is not None: + on_item_complete(item_result) + + return BulkIndexSummary( + total=len(results), + indexed=sum(1 for r in results if r.status == "indexed"), + skipped=sum(1 for r in results if r.status == "skipped"), + failed=sum(1 for r in results if r.status == "failed"), + queued=sum(1 for r in results if r.status == "queued"), + results=tuple(results), + ) diff --git a/src/vidxp/cli_commands/index.py b/src/vidxp/cli_commands/index.py index 14eb192a..d51a1593 100644 --- a/src/vidxp/cli_commands/index.py +++ b/src/vidxp/cli_commands/index.py @@ -1,15 +1,18 @@ from __future__ import annotations -from typing import Annotated, Iterable +from typing import Annotated, Any, Iterable import typer from rich.console import Console +from rich.markup import escape from rich.table import Table from vidxp.application_models import ( CreateIndexCommand, + PlanBulkIndexCommand, RemoveIndexCommand, ) +from vidxp.bulk_indexing import run_bulk_index from vidxp.cli_support import ( CLIState, IndexProgress, @@ -36,9 +39,7 @@ def create_index( capability_options: dict[str, dict], detach: bool = False, ) -> dict: - show_progress = ( - not state.quiet and state.output_format == OutputFormat.rich - ) + show_progress = not state.quiet and state.output_format == OutputFormat.rich selected = tuple(modalities) with IndexProgress(show_progress) as progress: job = state.jobs.submit_index( @@ -54,9 +55,7 @@ def create_index( job = state.jobs.wait( job.job_id, progress=lambda current: ( - progress.update( - current.progress.model_dump(mode="python") - ) + progress.update(current.progress.model_dump(mode="python")) if current.progress is not None else None ), @@ -97,10 +96,7 @@ def index_create( typer.Option( "--frame-stride", min=1, - help=( - "Materialize every Nth frame for actor and legacy visual " - "indexing." - ), + help=("Materialize every Nth frame for actor and legacy visual indexing."), ), ] = 1, scene_sample_fps: Annotated[ @@ -154,6 +150,217 @@ def index_create( ) +@app.command("bulk") +def index_bulk( + ctx: typer.Context, + media_ids: Annotated[ + list[str] | None, + typer.Argument( + help="Registered media identifiers to index.", + ), + ] = None, + all_eligible: Annotated[ + bool, + typer.Option( + "--all", + help="Index all eligible registered media in the catalog.", + ), + ] = False, + plan_only: Annotated[ + bool, typer.Option("--plan-only", help="Preview indexing and skip decisions.") + ] = False, + reindex: Annotated[ + bool, + typer.Option( + "--reindex", + help="Reindex media even if already present in the active index.", + ), + ] = False, + modalities: Annotated[ + list[str] | None, + typer.Option( + "--modality", + "-m", + help="Modality to index; repeat to select more than one.", + ), + ] = None, + frame_stride: Annotated[ + int, + typer.Option( + "--frame-stride", + min=1, + help=("Materialize every Nth frame for actor and legacy visual indexing."), + ), + ] = 1, + scene_sample_fps: Annotated[ + float | None, + typer.Option( + "--scene-sample-fps", + min=0.01, + help=( + "Target scene samples per second; lower-FPS media uses every " + "available frame." + ), + ), + ] = None, + capability_options: Annotated[ + list[str] | None, + typer.Option( + "--option", + help=( + "Capability setting as CAPABILITY.KEY=VALUE; " + "repeat for multiple settings." + ), + ), + ] = None, + detach: Annotated[ + bool, + typer.Option( + "--detach", + help="Return after the durable job is queued.", + ), + ] = False, + json_output: Annotated[ + bool, + typer.Option("--json", help="Emit machine-readable JSON."), + ] = False, +) -> None: + """Index multiple media items or all eligible media in the catalog.""" + + if not media_ids and not all_eligible: + raise typer.BadParameter("Provide either media IDs or pass --all.") + if media_ids and all_eligible: + raise typer.BadParameter("Pass media IDs or --all, not both.") + + state = state_from_context(ctx) + indexable = tuple( + capability.name + for capability in state.service.list_capabilities() + if capability.supports_indexing + ) + selected = selected_modalities(modalities, indexable) + parsed_options = parse_capability_options(capability_options) + output_fmt = effective_output_format(state, json_output) + show_progress = not state.quiet and output_fmt == OutputFormat.rich + + if plan_only and detach: + raise typer.BadParameter("--plan-only cannot be combined with --detach.") + plan = state.service.plan_bulk_index( + PlanBulkIndexCommand( + media_ids=tuple(media_ids or ()), + modalities=selected, + reindex=reindex, + frame_stride=frame_stride, + scene_sample_fps=scene_sample_fps, + capability_options=parsed_options, + ) + ) + if plan_only: + if output_fmt == OutputFormat.json: + emit_json(plan.model_dump(mode="json")) + else: + for target in plan.targets: + detail = target.reason.value if target.reason else "would index" + typer.echo(f"{target.original_filename}: {detail}") + typer.echo(f"{len(plan.pending)} to index, {len(plan.skipped)} skipped.") + return + + completed = 0 + + def on_item_complete(result) -> None: + nonlocal completed + completed += 1 + if show_progress: + typer.echo( + f"[{completed}/{len(plan.targets)}] {result.filename}: {result.status}" + ) + + with IndexProgress(show_progress) as progress: + + def on_item_start(media_id: str, filename: str) -> None: + if show_progress: + progress.update( + { + "stage": "indexing", + "message": f"Indexing {filename} ({media_id[:8]}...)", + } + ) + + def on_item_progress(media_id: str, current: Any) -> None: + if show_progress: + if hasattr(current, "progress") and current.progress is not None: + progress.update(current.progress.model_dump(mode="python")) + elif isinstance(current, dict): + progress.update(current) + + summary = run_bulk_index( + application=state.service, + plan=plan, + on_item_complete=on_item_complete, + jobs=state.jobs, + media_ids=media_ids, + all_eligible=all_eligible, + skip_indexed=not reindex, + detach=detach, + modalities=selected, + frame_stride=frame_stride, + scene_sample_fps=scene_sample_fps, + capability_options=parsed_options, + on_item_start=on_item_start, + on_item_progress=on_item_progress, + ) + + if output_fmt == OutputFormat.json: + payload = { + "total": summary.total, + "indexed": summary.indexed, + "skipped": summary.skipped, + "failed": summary.failed, + "queued": summary.queued, + "results": [ + { + "media_id": r.media_id, + "filename": r.filename, + "status": r.status, + "job_id": r.job_id, + "error_code": r.error_code, + "error_message": r.error_message, + } + for r in summary.results + ], + } + emit_json(payload) + else: + table = Table(title="Bulk indexing summary") + table.add_column("Media ID") + table.add_column("Filename") + table.add_column("Status") + table.add_column("Job ID") + table.add_column("Error") + for r in summary.results: + error_str = ( + f"[{r.error_code}] {r.error_message}" + if r.error_code + else (r.error_message or "—") + ) + table.add_row( + escape(r.media_id), + escape(r.filename), + escape(r.status), + escape(r.job_id or "—"), + escape(error_str), + ) + Console().print(table) + typer.echo( + f"Total: {summary.total}, Indexed: {summary.indexed}, " + f"Skipped: {summary.skipped}, Failed: {summary.failed}, " + f"Queued: {summary.queued}." + ) + + if summary.failed > 0: + raise typer.Exit(code=1) + + @app.command("remove") def index_remove( ctx: typer.Context, @@ -169,9 +376,7 @@ def index_remove( """Remove one media item from the active snapshot.""" state = state_from_context(ctx) - removed = state.service.remove_from_index( - RemoveIndexCommand(media_id=media_id) - ) + removed = state.service.remove_from_index(RemoveIndexCommand(media_id=media_id)) payload = {"removed": removed, "media_id": media_id} if effective_output_format(state, json_output) == OutputFormat.json: emit_json(payload) @@ -216,10 +421,7 @@ def index_list( assets = ( () if summary is None - else tuple( - state.service.get_media(media_id) - for media_id in summary.media_ids - ) + else tuple(state.service.get_media(media_id) for media_id in summary.media_ids) ) payload = { "state": status.state, diff --git a/src/vidxp/control_plane.py b/src/vidxp/control_plane.py index 7b84a08a..bc86329c 100644 --- a/src/vidxp/control_plane.py +++ b/src/vidxp/control_plane.py @@ -6,6 +6,12 @@ from vidxp.application_boundary import application_boundary from vidxp.application_models import ( Artifact, + IndexOptions, + PlanBulkIndexCommand, + BulkIndexPlan, + BulkIndexTarget, + BulkIndexTargetState, + BulkIndexSkipReason, CapabilityInfo, CapabilityRole, CapabilitySummary, @@ -30,7 +36,9 @@ from vidxp.capabilities.contracts import CapabilityRequestError from vidxp.capability_service import CapabilityService from vidxp.core.media import QuarantinedMedia -from vidxp.core.snapshots import IndexSnapshot +from vidxp.core.snapshots import GenerationReference, IndexSnapshot +from vidxp.core.contracts import IndexConfig +from vidxp.core.media import MediaState from vidxp.index_state import INDEX_STATUS_SCHEMA from vidxp.media_service import MediaService from vidxp.ports import LocalFileResource @@ -59,6 +67,112 @@ def __init__( self._read_active_snapshot = active_snapshot or (lambda: None) self.model_cache = model_cache + def _index_config( + self, command: IndexOptions, *, media_id: str | None = None + ) -> IndexConfig: + selected = self.select_index_modalities(command.modalities) + registry = self.capabilities.registry + options = { + name: dict(values) for name, values in command.capability_options.items() + } + if command.scene_sample_fps is not None: + options.setdefault("scene", {})["sample_fps"] = command.scene_sample_fps + return IndexConfig.local( + video_id=media_id, + enabled_modalities=selected, + frame_stride=command.frame_stride, + storage_directory=self.layout.indexes, + collection_names=registry.collection_names(selected), + capability_options=registry.validate_options(selected, options), + ) + + @application_boundary + def plan_bulk_index(self, command: PlanBulkIndexCommand) -> BulkIndexPlan: + """Decide which registered media still need indexing. + + The plan is a read-only decision. Callers submit one ordinary indexing + operation per pending target, so a failure isolates to its own media + and can be retried without disturbing the rest of the selection. + """ + + config = self._index_config(command) + selected = config.enabled_modalities + snapshot = self._read_active_snapshot() + generations = {} if snapshot is None else snapshot.generations + requested = frozenset(selected) + targets = tuple( + self._bulk_index_target( + asset, + generation=generations.get(asset.media_id), + requested=requested, + reindex=command.reindex, + config_fingerprint=config.fingerprint(), + ) + for asset in self._bulk_index_selection(command.media_ids) + ) + return BulkIndexPlan( + targets=targets, + modalities=selected, + options=IndexOptions( + modalities=selected, + frame_stride=config.frame_stride, + capability_options=config.capability_options, + ), + ) + + def _bulk_index_selection( + self, + media_ids: tuple[str, ...], + ) -> tuple[MediaAsset, ...]: + if media_ids: + return tuple(self.get_media(media_id) for media_id in media_ids) + assets: list[MediaAsset] = [] + cursor: str | None = None + while True: + page = self.list_media(ListMediaCommand(page_size=100, cursor=cursor)) + assets.extend(page.items) + cursor = page.next_cursor + if cursor is None: + break + return tuple(assets) + + @staticmethod + def _bulk_index_target( + asset: MediaAsset, + *, + generation: GenerationReference | None, + requested: frozenset[str], + reindex: bool, + config_fingerprint: str, + ) -> BulkIndexTarget: + if asset.state != MediaState.ready: + return BulkIndexTarget( + media_id=asset.media_id, + original_filename=asset.original_filename, + state=BulkIndexTargetState.skipped, + reason=BulkIndexSkipReason.media_not_ready, + ) + covered = ( + not reindex + and generation is not None + and requested <= frozenset(generation.modalities) + and generation.input_sha256 == asset.sha256 + and generation.config_fingerprint == config_fingerprint + ) + if covered and generation is not None: + return BulkIndexTarget( + media_id=asset.media_id, + original_filename=asset.original_filename, + state=BulkIndexTargetState.skipped, + reason=BulkIndexSkipReason.already_indexed, + generation_id=generation.generation_id, + ) + return BulkIndexTarget( + media_id=asset.media_id, + original_filename=asset.original_filename, + state=BulkIndexTargetState.pending, + ) + @application_boundary def import_uploaded_media( self, @@ -99,13 +213,9 @@ def select_index_modalities( registry = self.capabilities.registry indexable = registry.index_names() selected = ( - indexable - if requested is None - else registry.validate_names(requested) - ) - unsupported = tuple( - name for name in selected if name not in indexable + indexable if requested is None else registry.validate_names(requested) ) + unsupported = tuple(name for name in selected if name not in indexable) if unsupported: raise CapabilityRequestError( "Indexing does not support these capabilities: " diff --git a/tests/test_bulk_index.py b/tests/test_bulk_index.py new file mode 100644 index 00000000..ea6fce94 --- /dev/null +++ b/tests/test_bulk_index.py @@ -0,0 +1,306 @@ +import unittest +from datetime import datetime, timezone +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import Mock + +from vidxp.application import VidXPApplication +from vidxp.application_models import ( + BulkIndexSkipReason, + BulkIndexTargetState, + InvalidRequestError, + MediaAsset, + MediaPage, + PlanBulkIndexCommand, +) +from vidxp.capabilities.registry import create_capability_registry +from vidxp.core.contracts import IndexConfig +from vidxp.core.media import MediaState, MediaStream +from vidxp.core.snapshots import GenerationReference, IndexSnapshot +from vidxp.runtime import ModelRuntime +from vidxp.repository_layout import RepositoryLayout +from vidxp.settings import VidXPSettings + + +FIRST_MEDIA_ID = "123456781234423481234567890abcde" +SECOND_MEDIA_ID = "223456781234423481234567890abcde" +GENERATION_ID = "323456781234423481234567890abcde" +SNAPSHOT_ID = "423456781234423481234567890abcde" +FIRST_SHA256 = "a" * 64 +SECOND_SHA256 = "b" * 64 +_registry = create_capability_registry() +CONFIG_FINGERPRINT = IndexConfig.local( + enabled_modalities=("scene",), + collection_names=_registry.collection_names(("scene",)), + capability_options=_registry.validate_options(("scene",), {}), +).fingerprint() +MANIFEST_SHA256 = "d" * 64 + + +def media_asset( + media_id: str, + *, + sha256: str = FIRST_SHA256, + filename: str = "clip.mp4", + state: MediaState = MediaState.ready, +) -> MediaAsset: + return MediaAsset( + media_id=media_id, + video_id=media_id, + original_filename=filename, + sha256=sha256, + byte_size=1024, + detected_mime_type="video/mp4", + container="mov,mp4,m4a,3gp,3g2,mj2", + duration_seconds=3.0, + streams=(MediaStream(index=0, kind="video", codec="h264"),), + state=state, + created_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + ) + + +def generation( + media_id: str, + *, + input_sha256: str = FIRST_SHA256, + modalities: tuple[str, ...] = ("scene",), +) -> GenerationReference: + return GenerationReference( + generation_id=GENERATION_ID, + media_id=media_id, + manifest_sha256=MANIFEST_SHA256, + input_sha256=input_sha256, + config_fingerprint=CONFIG_FINGERPRINT, + modalities=modalities, + record_counts={name: 1 for name in modalities}, + store_size_bytes_at_commit=2048, + ) + + +def snapshot(*references: GenerationReference) -> IndexSnapshot: + return IndexSnapshot( + snapshot_id=SNAPSHOT_ID, + created_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + config_fingerprint=CONFIG_FINGERPRINT, + configuration={"enabled_modalities": ["scene"]}, + generations={reference.media_id: reference for reference in references}, + ) + + +class BulkIndexPlanTests(unittest.TestCase): + def application( + self, + root: str | Path, + *, + assets: tuple[MediaAsset, ...], + active_snapshot: IndexSnapshot | None = None, + page_size: int | None = None, + ) -> VidXPApplication: + settings = VidXPSettings( + repository_root=Path(root), + runtime_backend="cpu", + ) + media_service = Mock() + by_id = {asset.media_id: asset for asset in assets} + media_service.get.side_effect = lambda media_id: by_id[media_id] + + def list_media(command): + if page_size is None: + return MediaPage(items=assets, total=len(assets)) + start = 0 if command.cursor is None else int(command.cursor) + window = assets[start : start + page_size] + following = start + page_size + return MediaPage( + items=window, + total=len(assets), + next_cursor=(str(following) if following < len(assets) else None), + ) + + media_service.list.side_effect = list_media + return VidXPApplication( + settings=settings, + layout=RepositoryLayout(root=Path(root)), + registry=create_capability_registry(), + runtime=ModelRuntime(settings), + index_backend=Mock(), + media=media_service, + artifacts=Mock(), + index_status=lambda: None, + active_snapshot=lambda: active_snapshot, + ) + + def test_media_covered_by_the_active_snapshot_is_skipped(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID),), + active_snapshot=snapshot(generation(FIRST_MEDIA_ID)), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",)) + ) + self.assertEqual(len(plan.skipped), 1) + self.assertEqual(plan.pending, ()) + skipped = plan.skipped[0] + self.assertEqual(skipped.reason, BulkIndexSkipReason.already_indexed) + self.assertEqual(skipped.generation_id, GENERATION_ID) + + def test_changed_configuration_is_planned_and_matches_execution_options(self): + application = self.application( + "unused", + assets=(media_asset(FIRST_MEDIA_ID),), + active_snapshot=snapshot(generation(FIRST_MEDIA_ID)), + ) + for options in ( + {"frame_stride": 3}, + {"scene_sample_fps": 2}, + {"capability_options": {"scene": {"batch_size": 8}}}, + ): + with self.subTest(options=options): + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",), **options) + ) + self.assertEqual(len(plan.pending), 1) + self.assertNotEqual( + application._index_config(plan.options).fingerprint(), + CONFIG_FINGERPRINT, + ) + + def test_media_missing_a_requested_modality_is_planned(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID),), + active_snapshot=snapshot( + generation(FIRST_MEDIA_ID, modalities=("scene",)) + ), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene", "speech")) + ) + self.assertEqual(len(plan.pending), 1) + self.assertEqual(plan.skipped, ()) + + def test_replaced_media_content_is_planned_again(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID, sha256=SECOND_SHA256),), + active_snapshot=snapshot( + generation(FIRST_MEDIA_ID, input_sha256=FIRST_SHA256) + ), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",)) + ) + self.assertEqual(len(plan.pending), 1) + self.assertEqual(plan.skipped, ()) + + def test_reindex_plans_media_the_snapshot_already_covers(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID),), + active_snapshot=snapshot(generation(FIRST_MEDIA_ID)), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",), reindex=True) + ) + self.assertEqual(len(plan.pending), 1) + self.assertEqual(plan.skipped, ()) + + def test_media_that_is_not_ready_is_skipped(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID, state=MediaState.pending),), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",)) + ) + self.assertEqual(len(plan.skipped), 1) + self.assertEqual( + plan.skipped[0].reason, + BulkIndexSkipReason.media_not_ready, + ) + + def test_an_empty_selection_covers_every_registered_media_item(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=( + media_asset(FIRST_MEDIA_ID, filename="one.mp4"), + media_asset(SECOND_MEDIA_ID, filename="two.mp4"), + ), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",)) + ) + self.assertEqual(len(plan.targets), 2) + self.assertEqual( + [target.state for target in plan.targets], + [BulkIndexTargetState.pending] * 2, + ) + + def test_an_explicit_selection_ignores_other_media(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=( + media_asset(FIRST_MEDIA_ID, filename="one.mp4"), + media_asset(SECOND_MEDIA_ID, filename="two.mp4"), + ), + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand( + media_ids=(SECOND_MEDIA_ID,), + modalities=("scene",), + ) + ) + self.assertEqual(len(plan.targets), 1) + self.assertEqual(plan.targets[0].media_id, SECOND_MEDIA_ID) + + def test_every_page_of_registered_media_is_selected(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=( + media_asset(FIRST_MEDIA_ID, filename="one.mp4"), + media_asset(SECOND_MEDIA_ID, filename="two.mp4"), + ), + page_size=1, + ) + plan = application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("scene",)) + ) + self.assertEqual( + [target.media_id for target in plan.targets], + [FIRST_MEDIA_ID, SECOND_MEDIA_ID], + ) + + def test_an_unknown_capability_is_rejected(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID),), + ) + with self.assertRaises(InvalidRequestError): + application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("nonexistent",)) + ) + + def test_planning_reads_no_media_when_the_selection_is_rejected(self): + with TemporaryDirectory() as root: + application = self.application( + root, + assets=(media_asset(FIRST_MEDIA_ID),), + ) + with self.assertRaises(InvalidRequestError): + application.plan_bulk_index( + PlanBulkIndexCommand(modalities=("nonexistent",)) + ) + application.media.list.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_bulk_indexing.py b/tests/test_bulk_indexing.py new file mode 100644 index 00000000..c7ec5f37 --- /dev/null +++ b/tests/test_bulk_indexing.py @@ -0,0 +1,688 @@ +from __future__ import annotations + +import json +import unittest +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from types import MethodType +from vidxp.control_plane import ControlPlaneApplication +from vidxp.core.contracts import IndexConfig +from unittest.mock import Mock, patch + +from click import unstyle +from typer.testing import CliRunner + +from vidxp import cli +from vidxp.application_models import ( + ApplicationError, + CreateIndexCommand, + DependencyCheckResult, + ErrorCategory, + IndexJobResult, + IndexResult, + Job, + JobKind, + JobProgress, + JobQueue, + JobState, + ListMediaCommand, + MediaAsset, + MediaPage, + MediaState, + MediaStream, +) +from vidxp.bulk_indexing import ( + BulkIndexItemResult, + BulkIndexSummary, + _resolve_all_media, + run_bulk_index, +) +from vidxp.capabilities.registry import create_capability_registry +from vidxp.capability_service import CapabilityService +from vidxp.composition import LocalApplicationContext +from vidxp.core.snapshots import GenerationReference, IndexSnapshot +from vidxp.repositories import RepositoryConfig, RepositoryRegistry + +MEDIA_ID_1 = "123456781234423481234567890abcde" +MEDIA_ID_2 = "223456781234423481234567890abcde" +MEDIA_ID_3 = "323456781234423481234567890abcde" +JOB_ID_1 = "423456781234423481234567890abcde" +JOB_ID_2 = "523456781234423481234567890abcde" +SNAPSHOT_ID = "623456781234423481234567890abcde" +GENERATION_ID = "723456781234423481234567890abcde" + + +def make_media(media_id: str, filename: str = "video.mp4") -> MediaAsset: + return MediaAsset( + schema_version=1, + media_id=media_id, + video_id=media_id, + original_filename=filename, + sha256="1" * 64, + byte_size=1024, + detected_mime_type="video/mp4", + container="mp4", + duration_seconds=10.0, + streams=( + MediaStream( + index=0, + kind="video", + codec="h264", + width=640, + height=480, + ), + ), + state=MediaState.ready, + created_at=datetime.now(timezone.utc), + ) + + +def make_job( + job_id: str, + state: JobState = JobState.succeeded, + media_id: str = MEDIA_ID_1, +) -> Job: + result = None + if state == JobState.succeeded: + result = IndexJobResult( + result=IndexResult( + media_id=media_id, + generation_id=GENERATION_ID, + snapshot_id=SNAPSHOT_ID, + active_media_count=1, + record_counts={"scene": 1}, + ) + ) + return Job( + job_id=job_id, + kind=JobKind.index, + state=state, + queue=JobQueue.cpu, + result=result, + progress=JobProgress( + stage="indexing", + current=1, + total=1, + message="Done", + updated_at=datetime.now(timezone.utc), + ), + ) + + +def make_snapshot(generations: dict[str, tuple[str, ...]]) -> IndexSnapshot: + gen_refs: dict[str, GenerationReference] = {} + for mid, modalities in generations.items(): + gen_refs[mid] = GenerationReference( + generation_id=GENERATION_ID, + media_id=mid, + manifest_sha256="1" * 64, + input_sha256="1" * 64, + config_fingerprint=IndexConfig.local( + enabled_modalities=modalities, + collection_names=create_capability_registry().collection_names( + modalities + ), + capability_options=create_capability_registry().validate_options( + modalities, {} + ), + ).fingerprint(), + modalities=modalities, + record_counts={m: 1 for m in modalities}, + store_size_bytes_at_commit=1024, + ) + return IndexSnapshot( + schema_version=1, + snapshot_id=SNAPSHOT_ID, + created_at=datetime.now(timezone.utc), + config_fingerprint="0" * 64, + configuration={}, + generations=gen_refs, + ) + + +class BulkIndexingHelperTests(unittest.TestCase): + def test_resolve_all_media_paginates_cursor(self): + media_1 = make_media(MEDIA_ID_1, "first.mp4") + media_2 = make_media(MEDIA_ID_2, "second.mp4") + media_3 = make_media(MEDIA_ID_3, "third.mp4") + + app = Mock() + app.list_media.side_effect = [ + MediaPage(items=(media_1, media_2), total=3, next_cursor="c1"), + MediaPage(items=(media_3,), total=3, next_cursor=None), + ] + + result = _resolve_all_media(app) + + self.assertEqual(result, [media_1, media_2, media_3]) + self.assertEqual(app.list_media.call_count, 2) + app.list_media.assert_any_call( + ListMediaCommand( + page_size=100, + cursor=None, + state=MediaState.ready, + ) + ) + app.list_media.assert_any_call( + ListMediaCommand( + page_size=100, + cursor="c1", + state=MediaState.ready, + ) + ) + + def test_resolve_all_media_empty_catalog(self): + app = Mock() + app.list_media.return_value = MediaPage(items=(), total=0, next_cursor=None) + + result = _resolve_all_media(app) + + self.assertEqual(result, []) + self.assertEqual(app.list_media.call_count, 1) + + +def bind_planner(app): + app.capabilities = CapabilityService(create_capability_registry()) + app.layout.indexes = Path("unused/indexes") + for name in ( + "select_index_modalities", + "_index_config", + "plan_bulk_index", + "_bulk_index_selection", + ): + setattr(app, name, MethodType(getattr(ControlPlaneApplication, name), app)) + app._bulk_index_target = ControlPlaneApplication._bulk_index_target + + +class RunBulkIndexTests(unittest.TestCase): + def setUp(self): + self.app = Mock() + bind_planner(self.app) + self.jobs = Mock() + self.media_1 = make_media(MEDIA_ID_1, "one.mp4") + self.media_2 = make_media(MEDIA_ID_2, "two.mp4") + self.app.get_media.side_effect = lambda mid: ( + self.media_1 if mid == MEDIA_ID_1 else self.media_2 + ) + self.app._read_active_snapshot.return_value = None + + def test_bulk_index_multiple_media_ids_success(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.side_effect = [job_1, job_2] + self.jobs.wait.side_effect = [job_1, job_2] + + started: list[tuple[str, str]] = [] + completed: list[BulkIndexItemResult] = [] + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1, MEDIA_ID_2], + modalities=["scene"], + on_item_start=lambda mid, fn: started.append((mid, fn)), + on_item_complete=lambda r: completed.append(r), + ) + + self.assertEqual( + summary, + BulkIndexSummary( + total=2, + indexed=2, + skipped=0, + failed=0, + queued=0, + results=( + BulkIndexItemResult( + media_id=MEDIA_ID_1, + filename="one.mp4", + status="indexed", + job_id=JOB_ID_1, + ), + BulkIndexItemResult( + media_id=MEDIA_ID_2, + filename="two.mp4", + status="indexed", + job_id=JOB_ID_2, + ), + ), + ), + ) + self.assertEqual( + started, + [(MEDIA_ID_1, "one.mp4"), (MEDIA_ID_2, "two.mp4")], + ) + self.assertEqual(len(completed), 2) + self.assertEqual(self.jobs.submit_index.call_count, 2) + self.assertEqual(self.jobs.wait.call_count, 2) + + def test_bulk_index_all_eligible_paginates(self): + self.app.list_media.side_effect = [ + MediaPage(items=(self.media_1,), total=2, next_cursor="c1"), + MediaPage(items=(self.media_2,), total=2, next_cursor=None), + ] + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.side_effect = [job_1, job_2] + self.jobs.wait.side_effect = [job_1, job_2] + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + all_eligible=True, + modalities=["scene"], + ) + + self.assertEqual(summary.total, 2) + self.assertEqual(summary.indexed, 2) + self.assertEqual(summary.skipped, 0) + self.assertEqual(summary.failed, 0) + + def test_bulk_index_skips_already_indexed(self): + snapshot = make_snapshot({MEDIA_ID_1: ("scene",)}) + self.app._read_active_snapshot.return_value = snapshot + + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.return_value = job_2 + self.jobs.wait.return_value = job_2 + + started: list[tuple[str, str]] = [] + completed: list[BulkIndexItemResult] = [] + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1, MEDIA_ID_2], + skip_indexed=True, + modalities=["scene"], + on_item_start=lambda mid, fn: started.append((mid, fn)), + on_item_complete=lambda r: completed.append(r), + ) + + self.assertEqual(summary.total, 2) + self.assertEqual(summary.indexed, 1) + self.assertEqual(summary.skipped, 1) + self.assertEqual(summary.failed, 0) + self.assertEqual(summary.results[0].status, "skipped") + self.assertEqual(summary.results[0].media_id, MEDIA_ID_1) + self.assertEqual(summary.results[1].status, "indexed") + self.assertEqual(summary.results[1].media_id, MEDIA_ID_2) + # started should NOT include skipped item + self.assertEqual(started, [(MEDIA_ID_2, "two.mp4")]) + self.assertEqual(len(completed), 2) + self.jobs.submit_index.assert_called_once() + + def test_bulk_index_reindex_flag_overrides_skip(self): + snapshot = make_snapshot({MEDIA_ID_1: ("scene",)}) + self.app._read_active_snapshot.return_value = snapshot + + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + self.jobs.wait.return_value = job_1 + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1], + skip_indexed=False, + modalities=["scene"], + ) + + self.assertEqual(summary.total, 1) + self.assertEqual(summary.indexed, 1) + self.assertEqual(summary.skipped, 0) + self.assertEqual(summary.results[0].status, "indexed") + + def test_bulk_index_detach_queues_jobs(self): + job_1 = make_job(JOB_ID_1, state=JobState.queued, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1], + detach=True, + modalities=["scene"], + ) + + self.assertEqual(summary.total, 1) + self.assertEqual(summary.queued, 1) + self.assertEqual(summary.indexed, 0) + self.assertEqual(summary.results[0].status, "queued") + self.assertEqual(summary.results[0].job_id, JOB_ID_1) + self.jobs.wait.assert_not_called() + + def test_bulk_index_error_resilience(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.side_effect = [job_1, job_2] + self.jobs.wait.side_effect = [ + ApplicationError( + "transcription_failed", + ErrorCategory.unavailable, + "Model crashed during transcription.", + ), + job_2, + ] + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1, MEDIA_ID_2], + modalities=["scene"], + ) + + self.assertEqual(summary.total, 2) + self.assertEqual(summary.indexed, 1) + self.assertEqual(summary.failed, 1) + self.assertEqual(summary.results[0].job_id, JOB_ID_1) + self.assertEqual(summary.skipped, 0) + self.assertEqual(summary.results[0].status, "failed") + self.assertEqual(summary.results[0].error_code, "transcription_failed") + self.assertEqual( + summary.results[0].error_message, + "Model crashed during transcription.", + ) + self.assertEqual(summary.results[1].status, "indexed") + self.assertEqual(summary.results[1].job_id, JOB_ID_2) + + def test_bulk_index_generic_exception_resilience(self): + self.jobs.submit_index.side_effect = RuntimeError("Disk IO error") + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1], + modalities=["scene"], + ) + + self.assertEqual(summary.total, 1) + self.assertEqual(summary.failed, 1) + self.assertEqual(summary.results[0].status, "failed") + self.assertEqual(summary.results[0].error_code, "unexpected_error") + self.assertIn("Disk IO error", summary.results[0].error_message or "") + + def test_bulk_index_forwards_options_and_command_fields(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + self.jobs.wait.return_value = job_1 + + progress_events: list[tuple[str, Any]] = [] + + def fake_wait(job_id, progress=None): + if progress: + progress(job_1) + return job_1 + + self.jobs.wait.side_effect = fake_wait + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1], + modalities=["scene"], + frame_stride=3, + scene_sample_fps=1.5, + capability_options={"scene": {"batch_size": 8}}, + on_item_progress=lambda mid, curr: progress_events.append((mid, curr)), + ) + + self.assertEqual(summary.total, 1) + self.assertEqual(summary.indexed, 1) + submitted_command = self.jobs.submit_index.call_args.args[0] + self.assertIsInstance(submitted_command, CreateIndexCommand) + self.assertEqual(submitted_command.media_id, MEDIA_ID_1) + self.assertEqual(submitted_command.modalities, ("scene",)) + self.assertEqual(submitted_command.frame_stride, 3) + self.assertEqual(submitted_command.scene_sample_fps, 1.5) + self.assertEqual( + submitted_command.capability_options, + {"scene": {"batch_size": 8}}, + ) + self.assertEqual(len(progress_events), 1) + self.assertEqual(progress_events[0][0], MEDIA_ID_1) + + def test_bulk_index_default_modalities_resolves_from_application(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + self.jobs.wait.return_value = job_1 + self.app.select_index_modalities = lambda requested: ( + requested or ("scene", "speech") + ) + + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + media_ids=[MEDIA_ID_1], + modalities=None, + ) + + self.assertEqual(summary.total, 1) + self.assertEqual(summary.indexed, 1) + submitted_command = self.jobs.submit_index.call_args.args[0] + self.assertEqual(submitted_command.modalities, ("scene", "speech")) + + def test_bulk_index_empty_items(self): + self.app.list_media.return_value = MediaPage(items=(), total=0) + summary = run_bulk_index( + application=self.app, + jobs=self.jobs, + all_eligible=True, + modalities=["scene"], + ) + self.assertEqual(summary.total, 0) + self.assertEqual(summary.indexed, 0) + self.assertEqual(summary.skipped, 0) + self.assertEqual(summary.failed, 0) + self.assertEqual(summary.queued, 0) + self.assertEqual(summary.results, ()) + + +class CliBulkIndexTests(unittest.TestCase): + def setUp(self): + self.runner = CliRunner() + self.service = Mock() + bind_planner(self.service) + self.service.registry = create_capability_registry() + self.service.list_capabilities.return_value = CapabilityService( + self.service.registry + ).list() + self.service.index_directory = Path("repo/indexes") + self.service.layout.root = Path("repo") + self.service.model_cache = Path("model-cache") + self.service.runtime.backends.requested = "cpu" + self.service.model_readiness.return_value = DependencyCheckResult( + ok=True, + modalities=(), + checks=(), + ) + self.service._read_active_snapshot.return_value = None + + self.media_1 = make_media(MEDIA_ID_1, "one.mp4") + self.media_2 = make_media(MEDIA_ID_2, "two.mp4") + self.service.get_media.side_effect = lambda mid: ( + self.media_1 if mid == MEDIA_ID_1 else self.media_2 + ) + self.service.list_media.return_value = MediaPage( + items=(self.media_1, self.media_2), total=2, next_cursor=None + ) + + self.jobs = Mock() + self.registry = Mock(spec=RepositoryRegistry) + self.registry.path = Path("repositories.json") + self.repository = RepositoryConfig( + "default", + Path("repo"), + device="cpu", + configured=False, + ) + + def invoke(self, arguments, *, media_runtime_initialized=True): + with ( + patch.object( + cli, + "create_local_application", + return_value=LocalApplicationContext( + application=self.service, + jobs=self.jobs, + repositories=self.registry, + repository=self.repository, + ), + ) as create_local_application, + patch( + "vidxp.cli_support.media_runtime_is_initialized", + return_value=media_runtime_initialized, + ), + ): + result = self.runner.invoke(cli.app, arguments) + self.create_local_application = create_local_application + return result + + def test_cli_preview_does_not_submit_jobs(self): + result = self.invoke( + ["index", "bulk", "--all", "--modality", "scene", "--plan-only", "--json"] + ) + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.output) + self.assertEqual(len(payload["targets"]), 2) + self.jobs.submit_index.assert_not_called() + + def test_cli_requires_media_ids_or_all(self): + result = self.invoke(["index", "bulk"]) + self.assertEqual(result.exit_code, 2, result.output) + self.assertIn("Provide either media IDs or pass --all.", unstyle(result.output)) + + def test_cli_rejects_both_media_ids_and_all(self): + result = self.invoke(["index", "bulk", MEDIA_ID_1, "--all"]) + self.assertEqual(result.exit_code, 2, result.output) + self.assertIn("Pass media IDs or --all, not both.", unstyle(result.output)) + + def test_cli_bulk_index_success_json(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.side_effect = [job_1, job_2] + self.jobs.wait.side_effect = [job_1, job_2] + + result = self.invoke( + [ + "index", + "bulk", + MEDIA_ID_1, + MEDIA_ID_2, + "--modality", + "scene", + "--json", + ] + ) + + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.output) + self.assertEqual(payload["total"], 2) + self.assertEqual(payload["indexed"], 2) + self.assertEqual(payload["failed"], 0) + self.assertEqual(payload["skipped"], 0) + self.assertEqual(len(payload["results"]), 2) + self.assertEqual(payload["results"][0]["media_id"], MEDIA_ID_1) + self.assertEqual(payload["results"][0]["status"], "indexed") + self.assertEqual(payload["results"][0]["job_id"], JOB_ID_1) + + def test_cli_bulk_index_all_success_table(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.side_effect = [job_1, job_2] + self.jobs.wait.side_effect = [job_1, job_2] + + result = self.invoke(["index", "bulk", "--all", "--modality", "scene"]) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertIn("Bulk indexing summary", result.output) + self.assertIn( + "Total: 2, Indexed: 2, Skipped: 0, Failed: 0, Queued: 0.", result.output + ) + + def test_cli_bulk_index_forwards_options_and_flags(self): + job_1 = make_job(JOB_ID_1, state=JobState.queued, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + + result = self.invoke( + [ + "index", + "bulk", + MEDIA_ID_1, + "--modality", + "scene", + "--frame-stride", + "4", + "--scene-sample-fps", + "2.5", + "--option", + "scene.batch_size=8", + "--detach", + "--reindex", + "--json", + ] + ) + + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.output) + self.assertEqual(payload["total"], 1) + self.assertEqual(payload["queued"], 1) + submitted_command = self.jobs.submit_index.call_args.args[0] + self.assertEqual(submitted_command.frame_stride, 4) + self.assertEqual(submitted_command.scene_sample_fps, 2.5) + self.assertEqual( + submitted_command.capability_options, + {"scene": {"batch_size": 8}}, + ) + + def test_cli_bulk_index_with_failure_exits_code_1(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + self.jobs.wait.side_effect = ApplicationError( + "model_error", + ErrorCategory.unavailable, + "Failed to load model weights.", + ) + + result = self.invoke( + ["index", "bulk", MEDIA_ID_1, "--modality", "scene", "--json"] + ) + + self.assertEqual(result.exit_code, 1, result.output) + payload = json.loads(result.output) + self.assertEqual(payload["total"], 1) + self.assertEqual(payload["failed"], 1) + self.assertEqual(payload["results"][0]["status"], "failed") + self.assertEqual(payload["results"][0]["error_code"], "model_error") + + def test_cli_bulk_index_all_json(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + job_2 = make_job(JOB_ID_2, media_id=MEDIA_ID_2) + self.jobs.submit_index.side_effect = [job_1, job_2] + self.jobs.wait.side_effect = [job_1, job_2] + + result = self.invoke( + ["index", "bulk", "--all", "--modality", "scene", "--json"] + ) + + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.output) + self.assertEqual(payload["total"], 2) + self.assertEqual(payload["indexed"], 2) + + def test_cli_bulk_index_with_failure_table_output(self): + job_1 = make_job(JOB_ID_1, media_id=MEDIA_ID_1) + self.jobs.submit_index.return_value = job_1 + self.jobs.wait.side_effect = ApplicationError( + "model_error", + ErrorCategory.unavailable, + "Failed to load model weights.", + ) + + result = self.invoke(["index", "bulk", MEDIA_ID_1, "--modality", "scene"]) + + self.assertEqual(result.exit_code, 1, result.output) + self.assertIn("Bulk indexing summary", result.output) + self.assertIn("model_error", result.output) + self.assertIn("Failed: 1", result.output)