ditto.cli¶

def gather_stats(
    entries: list[ManifestEntry], ext_map: Mapping[str, RecorderInfo]
) -> SnapshotStats:
    """Aggregate snapshot statistics from a list of ManifestEntry items."""
    total_size = SizeSummary()
    by_recorder: dict[str, RecorderStats] = {}
    oldest: tuple[float, str] | None = None
    newest: tuple[float, str] | None = None

    for entry in entries:
        total_size = _add_size(total_size, entry.size_bytes)
        _, _, ext = _parse_snapshot_name(entry.storage_key)
        recorder_name = _recorder_name(ext, ext_map)
        current = by_recorder.get(
            recorder_name,
            RecorderStats(count=0, size=SizeSummary()),
        )
        by_recorder[recorder_name] = RecorderStats(
            count=current.count + 1,
            size=_add_size(current.size, entry.size_bytes),
        )

        if entry.modified is not None:
            if oldest is None or entry.modified < oldest[0]:
                oldest = (entry.modified, entry.storage_key)
            if newest is None or entry.modified > newest[0]:
                newest = (entry.modified, entry.storage_key)

    return SnapshotStats(
        total_count=len(entries),
        total_size=total_size,
        by_recorder=by_recorder,
        oldest=oldest,
        newest=newest,
    )

def render_stats(stats: SnapshotStats, console: Console) -> None:
    """Render a SnapshotStats value as a Rich panel."""
    colour_map = _build_colour_map(stats.by_recorder.keys())

    lines = Text()
    lines.append("  Total snapshots  ", style=MUTED)
    lines.append(f"{stats.total_count}\n", style=f"bold {TEXT}")
    lines.append("  Total size       ", style=MUTED)
    lines.append(f"{_format_size_summary(stats.total_size)}\n", style=f"bold {TEXT}")
    lines.append("\n")
    lines.append("  By recorder:\n", style=f"bold {HEADER}")
    name_w = max(len(name) for name in stats.by_recorder.keys())
    count_w = max(len(str(recorder.count)) for recorder in stats.by_recorder.values())
    size_w = max(
        len(_format_size_summary(recorder.size))
        for recorder in stats.by_recorder.values()
    )
    for name, recorder in sorted(stats.by_recorder.items()):
        lines.append(f"    {name:<{name_w}}", style=colour_map.get(name, MUTED))
        lines.append(f"  {recorder.count:>{count_w}}  ", style=TEXT)
        lines.append(
            f"{_format_size_summary(recorder.size):>{size_w}}\n",
            style=MUTED,
        )

    if stats.oldest and stats.newest:
        lines.append("\n")
        oldest_date = datetime.fromtimestamp(stats.oldest[0]).strftime("%Y-%m-%d")
        newest_date = datetime.fromtimestamp(stats.newest[0]).strftime("%Y-%m-%d")
        lines.append("  Oldest  ", style=MUTED)
        lines.append(f"{stats.oldest[1]}  ", style=PATH)
        lines.append(f"{oldest_date}\n", style=MUTED)
        lines.append("  Newest  ", style=MUTED)
        lines.append(f"{stats.newest[1]}  ", style=PATH)
        lines.append(f"{newest_date}", style=MUTED)

    console.print(
        Panel(
            lines,
            title=f"[bold {TITLE}]ditto status[/bold {TITLE}]",
            border_style=TITLE,
            expand=False,
        )
    )

@click.group()
@click.version_option(package_name="pytest-ditto", message="%(package)s %(version)s")
def cli():
    """pytest-ditto snapshot management."""

@cli.command(
    name="run",
    context_settings={"ignore_unknown_options": True, "allow_extra_args": True},
)
@click.argument("pytest_args", nargs=-1, type=click.UNPROCESSED)
def cmd_run(pytest_args):
    """Run pytest, reporting any snapshot activity at the end.

    Any extra arguments are passed directly to pytest.

    \b
    Examples:
      ditto run
      ditto run tests/ci/
      ditto run tests/ci/ -k test_foo
    """
    result = subprocess.run(
        [sys.executable, "-m", "pytest", *pytest_args],
        check=False,
    )
    sys.exit(result.returncode)

@cli.command(
    name="update",
    context_settings={"ignore_unknown_options": True, "allow_extra_args": True},
)
@click.argument("pytest_args", nargs=-1, type=click.UNPROCESSED)
def cmd_update(pytest_args):
    """Re-run pytest with --ditto-update to regenerate snapshots.

    Any extra arguments are passed directly to pytest.

    \b
    Examples:
      ditto update
      ditto update tests/ci/
      ditto update tests/ci/ -k test_foo
    """
    result = subprocess.run(
        [sys.executable, "-m", "pytest", "--ditto-update", *pytest_args],
        check=False,
    )
    sys.exit(result.returncode)

@cli.command(
    name="prune",
    context_settings={"ignore_unknown_options": True, "allow_extra_args": True},
)
@click.option(
    "--check",
    is_flag=True,
    default=False,
    help="Dry run: report what would be pruned, without deleting.",
)
@click.argument("pytest_args", nargs=-1, type=click.UNPROCESSED)
def cmd_prune(check, pytest_args):
    """Re-run pytest to delete snapshots not in ditto.lock.

    With --check, report what would be pruned without deleting anything. Any
    extra arguments are passed directly to pytest.

    \b
    Examples:
      ditto prune
      ditto prune --check
      ditto prune tests/ci/
    """
    flag = "--ditto-prune-dry-run" if check else "--ditto-prune"
    result = subprocess.run(
        [sys.executable, "-m", "pytest", flag, *pytest_args],
        check=False,
    )
    sys.exit(result.returncode)

@cli.command(
    name="lock",
    context_settings={"ignore_unknown_options": True, "allow_extra_args": True},
)
@click.argument("pytest_args", nargs=-1, type=click.UNPROCESSED)
def cmd_lock(pytest_args):
    """Rebuild ditto.lock from current snapshots (full run; values unchanged).

    Must run the whole suite: passing positional path/nodeid args narrows the run
    and is refused, because a narrowed rebuild can truncate entries for files it
    did not collect. Configure the suite's scope via `testpaths` in pyproject/ini
    instead.

    \b
    Examples:
      ditto lock
    """
    result = subprocess.run(
        [sys.executable, "-m", "pytest", "--ditto-lock", *pytest_args],
        check=False,
    )
    sys.exit(result.returncode)

@cli.command(
    name="verify",
    context_settings={"ignore_unknown_options": True, "allow_extra_args": True},
)
@click.argument("pytest_args", nargs=-1, type=click.UNPROCESSED)
def cmd_verify(pytest_args):
    """Fail if the backend has drifted from ditto.lock (read-only).

    \b
    Examples:
      ditto verify
      ditto verify tests/ci/
    """
    result = subprocess.run(
        [sys.executable, "-m", "pytest", "--ditto-verify", *pytest_args],
        check=False,
    )
    sys.exit(result.returncode)

@cli.command(name="list")
@_live_option
@click.argument(
    "path", default=".", type=click.Path(exists=True, file_okay=False, path_type=Path)
)
def cmd_list(path: Path, live: bool):
    """List all snapshot files under PATH (default: current directory).

    By default reads local snapshots from disk and remote snapshots from
    ditto.lock (credential-free); pass --live to read live backends.

    \b
    Examples:
      ditto list
      ditto list tests/ci/
    """
    manifest = _inventory_or_exit(path, live=live)
    entries = _entries(manifest)
    if not entries:
        console.print(f"[{MUTED}]No snapshot files found.[/{MUTED}]")
        _print_inventory_notes(path, entries, live=live)
        sys.exit(1)

    infos = _load_recorder_infos()
    em = _ext_map(infos)
    colour_map = _build_colour_map(info.name for info in infos)

    table = Table(
        title=f"[bold {TITLE}]ditto snapshots[/bold {TITLE}]",
        border_style=MUTED,
        header_style=f"bold {HEADER}",
        show_header=True,
    )
    table.add_column("Test", style=TEXT)
    table.add_column("Key", style=SUBTEXT1)
    table.add_column("Recorder")
    table.add_column("Size", justify="right", style=MUTED)
    table.add_column("Modified", style=MUTED)

    for entry in entries:
        group, key, ext = _parse_snapshot_name(entry.storage_key)
        recorder_name = _recorder_name(ext, em)
        modified = (
            datetime.fromtimestamp(entry.modified).strftime("%Y-%m-%d")
            if entry.modified is not None
            else "—"
        )
        table.add_row(
            group,
            key,
            Text(recorder_name, style=colour_map.get(recorder_name, MUTED)),
            _human_size(entry.size_bytes),
            modified,
        )

    console.print(table)
    _print_inventory_notes(path, entries, live=live)

@cli.command(name="clean")
@click.argument(
    "path", default=".", type=click.Path(exists=True, file_okay=False, path_type=Path)
)
@click.option("--yes", is_flag=True, default=False, help="Skip confirmation prompt.")
def cmd_clean(path: Path, yes: bool):
    """Delete all .ditto/ directories under PATH.

    Shows a preview of what will be deleted and requires confirmation
    unless --yes is passed.

    \b
    Examples:
      ditto clean
      ditto clean --yes
      ditto clean tests/ci/ --yes
    """
    dirs = _find_ditto_dirs(path)
    if not dirs:
        console.print(f"[{MUTED}]No .ditto/ directories found.[/{MUTED}]")
        sys.exit(1)

    preview = Text()
    preview.append("Will delete:\n\n", style=f"bold {TEXT}")
    for d in dirs:
        preview.append(f"  {d}\n", style=PATH)

    console.print(Panel(preview, border_style=PRUNED, expand=False))

    if not yes:
        click.confirm(click.style("\nProceed?", fg="bright_white"), abort=True)

    for d in dirs:
        shutil.rmtree(d)
        t = Text()
        t.append("  deleted  ", style=f"bold {PRUNED}")
        t.append(str(d), style=PATH)
        console.print(t)

    n = len(dirs)
    console.print(
        f"\n[bold {CREATED}]Removed {n} "
        f".ditto/ director{'y' if n == 1 else 'ies'}.[/bold {CREATED}]"
    )

@cli.command(name="status")
@_live_option
@click.argument(
    "path", default=".", type=click.Path(exists=True, file_okay=False, path_type=Path)
)
def cmd_status(path: Path, live: bool):
    """Show aggregate statistics for snapshots under PATH.

    By default aggregates local snapshots from disk and remote snapshots from
    ditto.lock (credential-free); pass --live to read live backends.

    \b
    Examples:
      ditto status
      ditto status tests/ci/
    """
    manifest = _inventory_or_exit(path, live=live)
    entries = _entries(manifest)
    if not entries:
        console.print(f"[{MUTED}]No snapshot files found.[/{MUTED}]")
        _print_inventory_notes(path, entries, live=live)
        sys.exit(1)

    render_stats(gather_stats(entries, _ext_map(_load_recorder_infos())), console)
    _print_inventory_notes(path, entries, live=live)

@cli.command(name="recorders")
def cmd_recorders():
    """List all registered recorder plugins.

    \b
    Examples:
      ditto recorders
    """
    infos = _load_recorder_infos()
    if not infos:
        console.print(f"[{MUTED}]No recorders registered.[/{MUTED}]")
        sys.exit(1)
    _render_recorders(infos, console)
    problems = RecorderRegistry().problems
    if problems:
        console.print(
            f"[{PRUNED}]{len(problems)} plugin contract problem(s); run "
            f"`ditto doctor` for details.[/{PRUNED}]"
        )

@cli.command(name="doctor")
def cmd_doctor():
    """Run health checks: plugin loading, pytest availability.

    \b
    Examples:
      ditto doctor
    """
    checks = _doctor_checks()
    _render_doctor(checks, console)
    if not all(c.ok for c in checks):
        sys.exit(1)

@cli.command(name="lint")
@_live_option
@click.argument(
    "path", default=".", type=click.Path(exists=True, file_okay=False, path_type=Path)
)
def cmd_lint(path: Path, live: bool):
    """Check snapshot files for naming issues, unknown formats, and empty files.

    By default lints local snapshots from disk and remote snapshots from
    ditto.lock (credential-free); pass --live to read live backends.

    \b
    Examples:
      ditto lint
      ditto lint tests/ci/
    """
    manifest = _inventory_or_exit(path, live=live)
    entries = _entries(manifest)
    issues = _find_lint_issues(entries, _ext_map(_load_recorder_infos()))
    if issues:
        _render_lint_issues(issues, console)
    else:
        console.print(f"[{MUTED}]All snapshots are valid.[/{MUTED}]")
    _print_inventory_notes(path, entries, live=live)
    if issues:
        sys.exit(1)

@cli.command(name="stats")
@_live_option
@click.argument(
    "path", default=".", type=click.Path(exists=True, file_okay=False, path_type=Path)
)
def cmd_stats(path: Path, live: bool):
    """Show per-directory snapshot usage breakdown.

    By default breaks down local snapshots from disk and remote snapshots from
    ditto.lock (credential-free); pass --live to read live backends.

    \b
    Examples:
      ditto stats
      ditto stats tests/ci/
    """
    manifest = _inventory_or_exit(path, live=live)
    if not manifest:
        console.print(f"[{MUTED}]No snapshot files found.[/{MUTED}]")
        _print_inventory_notes(path, [], live=live)
        sys.exit(1)
    em = _ext_map(_load_recorder_infos())
    dir_stats = [(b.location, gather_stats(b.entries, em)) for b in manifest]
    _render_stats_table(dir_stats, console)
    _print_inventory_notes(path, _entries(manifest), live=live)