Fleet

Run one command across every submodule of a superproject

Author

Benedict Thekkel

Discovery and reporting

Every verb walks the same list and reports the same way.

format_summary

def format_summary(
    rep, verb:str='ran'
):

The one line that replaces a message per submodule. Quiet submodules are counted, not narrated

Exported source

import json
import re
import shutil
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from pathlib import Path
from urllib.parse import unquote

from fastcore.script import *

# The rendering layer lives in `automate`, the git/discovery layer in `github`. This
# module is the only one that knows about MANY repos; those two stay single-repo.
from nbdevAuto.automate import StepFailed, _push_summary, _run, _step, console, escape
from nbdevAuto.github import find_root, git, local_state, parse_gitmodules

Exported source

class Skip(Exception):
    "Raised by a walked callable that has decided this submodule has nothing to do"

@dataclass
class Outcome:
    "What happened in one submodule"
    path:   str
    state:  str                                  # ran | skipped | failed
    reason: str  = ""                            # why it was skipped, or how it failed
    detail: list = field(default_factory=list)   # short lines to show under the path

@dataclass
class Report:
    "Everything a `walk` did, and the exit code it implies"
    outcomes: list = field(default_factory=list)

    @property
    def ran(self):    return [o for o in self.outcomes if o.state == "ran"]
    @property
    def failed(self): return [o for o in self.outcomes if o.state == "failed"]
    @property
    def code(self):   return 1 if self.failed else 0

    def skip_counts(self):
        "Skip reason -> count, in the order the reasons were first seen"
        counts = {}
        for o in self.outcomes:
            if o.state == "skipped": counts[o.reason] = counts.get(o.reason, 0) + 1
        return counts

def format_summary(rep, verb="ran"):
    "The one line that replaces a message per submodule. Quiet submodules are counted, not narrated"
    n = len(rep.outcomes)
    parts = [f"{len(rep.ran)} {verb}"]
    parts += [f"{c} {reason}" for reason, c in rep.skip_counts().items()]
    if rep.failed: parts.append(f"{len(rep.failed)} FAILED")
    return f"{n} submodule{'' if n == 1 else 's'}: " + ", ".join(parts)

source

Report

def Report(
    outcomes:list=<factory>
)->None:

Everything a walk did, and the exit code it implies

source

Outcome

def Outcome(
    path:str, state:str, reason:str='', detail:list=<factory>
)->None:

What happened in one submodule

source

Skip

def Skip(
    *args, **kwargs
):

Raised by a walked callable that has decided this submodule has nothing to do

source

finish

def finish(
    rep, verb
):

Print the counted summary, name any failures, and return the exit code

Exported source

def _classify(base, sub, need_nbs, only_dirty):
    "The reason to skip this submodule before running anything, or None to run it"
    p = base/sub.path
    # An uninitialised submodule is an empty directory: present, but with no .git
    if not (p/".git").exists():             return "not initialised"
    if need_nbs and not (p/"nbs").is_dir(): return "without nbs"
    if only_dirty and not git(p, "status", "--porcelain"): return "clean"
    return None

def _fail_parts(e):
    """(headline, remaining lines) for a callable that did not finish.

    A `StepFailed` carries the command's own output, and its first line is the message
    worth putting on the summary line - "uncommitted changes" says far more than
    "update exited 1". The rest is kept for the block renderer."""
    if isinstance(e, StepFailed):
        out = (e.output or "").strip().splitlines()
        if out: return out[0], out[1:]
        return f"{e.cmd} exited {e.code}", []
    return f"{type(e).__name__}: {e}", []

def _render(o, style="block", width=0, header=True):
    "One submodule's result. A skip prints nothing at all; it is counted instead"
    if o.state == "skipped": return
    if style == "line":
        pad = f"{o.path:<{width}}"
        if o.state == "failed":
            console.print(f"[bold red]{escape(pad)}[/bold red] [red]{escape(o.reason)}[/red]")
            return
        console.print(f"[cyan]{escape(pad)}[/cyan] [dim]{escape(o.detail[0] if o.detail else '')}[/dim]")
        for line in o.detail[1:]: console.print(f"{' ' * width} [dim]{escape(line)}[/dim]")
        return
    if header: console.print(f"\n[bold cyan]==> {escape(o.path)}[/bold cyan]")
    if o.state == "failed":
        console.print(f"    [bold red]FAIL[/bold red] [red]{escape(o.reason)}[/red]")
    for line in o.detail: console.print(f"    [dim]{escape(line)}[/dim]")

def walk(fn, *, root=None, only_dirty=False, need_nbs=False, workers=1, style="block"):
    """Run `fn(path, sub)` in each submodule, returning a `Report`.

    `fn` returns a list of short strings to show under the submodule's name, or raises
    `Skip` once it knows there is nothing to do. Only submodules that do something print;
    the rest are counted by `format_summary`.

    `style="block"` gives each submodule a `==> path` heading with its own output beneath,
    which is what a long command like `upload` needs. `style="line"` gives one padded line
    per submodule, which suits verbs that report a single fact.

    Sequential block runs print the heading BEFORE calling `fn`, so its output streams
    underneath. Everything else prints after, in submodule order, so parallel work cannot
    interleave and two runs of the same fleet read the same way."""
    base = Path(root) if root else find_root()
    rep, todo = Report(), []
    for s in parse_gitmodules(base):
        why = _classify(base, s, need_nbs, only_dirty)
        if why: rep.outcomes.append(Outcome(s.path, "skipped", why))
        else:   todo.append(s)
    width = max((len(s.path) for s in todo), default=0) + 2

    def _one(s):
        try:                       return Outcome(s.path, "ran", detail=list(fn(base/s.path, s) or []))
        except Skip as e:          return Outcome(s.path, "skipped", str(e))
        except BaseException as e:
            head, rest = _fail_parts(e)
            return Outcome(s.path, "failed", head, rest)

    if workers > 1 and len(todo) > 1:
        with ThreadPoolExecutor(max_workers=workers) as ex: done = list(ex.map(_one, todo))
        for o in done: _render(o, style, width)
    else:
        done = []
        live = style == "block"
        for s in todo:
            if live: console.print(f"\n[bold cyan]==> {escape(s.path)}[/bold cyan]")
            o = _one(s)
            done.append(o)
            _render(o, style, width, header=not live)
    rep.outcomes += done
    rep.outcomes.sort(key=lambda o: o.path)
    return rep

def finish(rep, verb):
    "Print the counted summary, name any failures, and return the exit code"
    console.print(f"\n[bold]{escape(format_summary(rep, verb))}[/bold]")
    for o in rep.failed:
        console.print(f"[bold red]FAILED[/bold red] {escape(o.path)}: [dim]{escape(o.reason)}[/dim]")
    return rep.code

source

walk

def walk(
    fn, *, root:NoneType=None, only_dirty:bool=False, need_nbs:bool=False, workers:int=1, style:str='block'
):

Run fn(path, sub) in each submodule, returning a Report.

fn returns a list of short strings to show under the submodule’s name, or raises Skip once it knows there is nothing to do. Only submodules that do something print; the rest are counted by format_summary.

style="block" gives each submodule a ==> path heading with its own output beneath, which is what a long command like upload needs. style="line" gives one padded line per submodule, which suits verbs that report a single fact.

Sequential block runs print the heading BEFORE calling fn, so its output streams underneath. Everything else prints after, in submodule order, so parallel work cannot interleave and two runs of the same fleet read the same way.

Running a console script per submodule

source

upload_one

def upload_one(
    path, sub:NoneType=None, args:tuple=()
):

Run the single-repo upload in path, letting its own rich output through.

A subprocess rather than an in-process call: prep reads nbdev’s config from the cwd and caches it, and nbdev_test spawns its own process pool, so nineteen repos in one interpreter is a trap. The subprocess keeps this tty, so colour survives.

Exported source

def script_path(name):
    """The console script `name` from the environment this is running in.

    Taking it from `sys.executable`'s directory is what makes this work under a
    non-interactive shell that never put the venv on PATH."""
    cand = Path(sys.executable).parent/name
    if cand.is_file(): return str(cand)
    found = shutil.which(name)
    if found: return found
    raise RuntimeError(f"Cannot find the '{name}' console script. Run `uv sync` first")

def upload_one(path, sub=None, args=()):
    """Run the single-repo `upload` in `path`, letting its own rich output through.

    A subprocess rather than an in-process call: `prep` reads nbdev's config from the cwd
    and caches it, and `nbdev_test` spawns its own process pool, so nineteen repos in one
    interpreter is a trap. The subprocess keeps this tty, so colour survives."""
    r = subprocess.run([script_path("upload"), *args], cwd=str(path), check=False)
    if r.returncode: raise StepFailed("upload", r.returncode)
    return []

source

script_path

def script_path(
    name
):

The console script name from the environment this is running in.

Taking it from sys.executable’s directory is what makes this work under a non-interactive shell that never put the venv on PATH.

Verbs

source

fleet_upload

def fleet_upload(
    all:bool=False, # Upload every submodule with an nbs/ folder, not just the changed ones
    root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):

Run upload in each submodule that has local changes

Exported source

def fleet_upload(
    all:bool = False,    # Upload every submodule with an nbs/ folder, not just the changed ones
    root:str = None,     # Superproject root (default: nearest ancestor with .gitmodules)
):
    "Run `upload` in each submodule that has local changes"
    # Sequential on purpose: `upload` runs nbdev_test with 8 workers, and knowledge-lab
    # has 4 vCPU. Parallel uploads would oversubscribe the box and interleave the output.
    rep = walk(upload_one, root=root, need_nbs=True, only_dirty=not all, workers=1)
    sys.exit(finish(rep, "uploaded"))

push and copy. Both report one line per submodule, and both take --dry-run.

source

copy_one

def copy_one(
    path, sub:NoneType=None, files:tuple=(), dest:str='nbs', dry_run:bool=False
):

Copy files into path/dest.

The file list belongs to the caller, so this stays a general-purpose copy rather than learning about any particular superproject’s shared chrome.

Exported source

def ahead_of_remote(path):
    "(commits origin has not seen, branch name). Branch is None when HEAD is detached"
    br = git(path, "rev-parse", "--abbrev-ref", "HEAD")
    if not br or br == "HEAD": return 0, None
    n = git(path, "rev-list", "--count", f"origin/{br}..HEAD")
    return (int(n) if n.isdigit() else 0), br

def push_one(path, sub=None, dry_run=False):
    "Push `path` when it has local commits origin has not seen"
    n, br = ahead_of_remote(path)
    if br is None:              raise Skip("detached HEAD")
    if not git(path, "remote"): raise Skip("no remote")
    if not n:                   raise Skip("nothing to push")
    if dry_run: return [f"would push {n} commit{'' if n == 1 else 's'} on {br}"]
    return _push_summary(_run("git", "-C", str(path), "push"))

def copy_one(path, sub=None, files=(), dest="nbs", dry_run=False):
    """Copy `files` into `path`/`dest`.

    The file list belongs to the caller, so this stays a general-purpose copy rather than
    learning about any particular superproject's shared chrome."""
    out = Path(path)/dest
    if not out.is_dir(): raise Skip(f"no {dest}/")
    names = []
    for f in files:
        src = Path(f)
        if not src.is_file(): raise StepFailed(f"copy {src.name}", 1, f"no such file: {src}")
        if not dry_run: shutil.copy2(src, out/src.name)
        names.append(src.name)
    return [("would copy " if dry_run else "copied ") + ", ".join(names) + f" -> {dest}/"]

source

push_one

def push_one(
    path, sub:NoneType=None, dry_run:bool=False
):

Push path when it has local commits origin has not seen

source

ahead_of_remote

def ahead_of_remote(
    path
):

(commits origin has not seen, branch name). Branch is None when HEAD is detached

source

fleet_copy

def fleet_copy(
    files:str, # Files to copy into each submodule, comma or space separated
    dest:str='nbs', # Directory inside each submodule to copy into
    dry_run:bool=False, # Report what would be copied without copying it
    root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):

Copy shared files into every submodule that has the destination directory

Exported source

def fleet_push(
    dry_run:bool = False,   # Report what would be pushed without pushing it
    root:str     = None,    # Superproject root (default: nearest ancestor with .gitmodules)
):
    "Push every submodule that has local commits not yet on its remote"
    # Network-bound and independent per remote, so this one does parallelise.
    rep = walk(lambda p, s: push_one(p, s, dry_run=dry_run),
               root=root, workers=8, style="line")
    sys.exit(finish(rep, "would push" if dry_run else "pushed"))

def fleet_copy(
    files:str,              # Files to copy into each submodule, comma or space separated
    dest:str     = "nbs",   # Directory inside each submodule to copy into
    dry_run:bool = False,   # Report what would be copied without copying it
    root:str     = None,    # Superproject root (default: nearest ancestor with .gitmodules)
):
    "Copy shared files into every submodule that has the destination directory"
    fs = [f for f in re.split(r"[,\s]+", files) if f]
    if not fs: sys.exit("fleet copy: no files given")
    rep = walk(lambda p, s: copy_one(p, s, files=fs, dest=dest, dry_run=dry_run),
               root=root, workers=8, style="line")
    sys.exit(finish(rep, "would update" if dry_run else "updated"))

source

fleet_push

def fleet_push(
    dry_run:bool=False, # Report what would be pushed without pushing it
    root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):

Push every submodule that has local commits not yet on its remote

status replaces both the git submodule status loop and the gss recipe.

source

fleet_status

def fleet_status(
    short:bool=False, # Only the submodules with changes, listing the changed paths
    root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):

Branch, dirty flag and HEAD for every submodule

Exported source

def _porcelain(path):
    """`git status -s` for `path`, NOT stripped.

    `github.git` strips its output, and `git status -s` pads the code to two columns, so
    stripping eats the leading space of the FIRST line and shifts that one path by a
    character. `automate.status` documents the same trap."""
    return subprocess.run(["git", "-C", str(path), "status", "-s"],
                          capture_output=True, text=True, check=False).stdout

def status_one(path, sub=None, short=False):
    """Branch, dirty flag and short HEAD for `path`.

    With `short`, clean submodules are skipped and the dirty ones list their changed
    paths - the fleet-wide "what have I got outstanding" view."""
    st = local_state(path)
    br = st["branch"] or "?"
    if br == "HEAD": br = "DETACHED"
    line = f"{br:<10} {'dirty' if st['dirty'] else 'clean':<6} {st['head'] or '-'}"
    if not short: return [line]
    if not st["dirty"]: raise Skip("clean")
    return [line] + [f"  {l}" for l in _porcelain(path).splitlines()]

def fleet_status(
    short:bool = False,   # Only the submodules with changes, listing the changed paths
    root:str   = None,    # Superproject root (default: nearest ancestor with .gitmodules)
):
    "Branch, dirty flag and HEAD for every submodule"
    rep = walk(lambda p, s: status_one(p, s, short=short), root=root, workers=8, style="line")
    sys.exit(finish(rep, "changed" if short else "reported"))

source

status_one

def status_one(
    path, sub:NoneType=None, short:bool=False
):

Branch, dirty flag and short HEAD for path.

With short, clean submodules are skipped and the dirty ones list their changed paths - the fleet-wide “what have I got outstanding” view.

update and sync, the destructive pair. Fetching parallelises; checkout and merge do not.

source

pull_root

def pull_root(
    base
):

Fast-forward the superproject itself, with the advice its failure actually needs

Exported source

def _git_ok(path, *args):
    "True when git succeeded in `path`. `git()` hides the exit code, and here it matters"
    return subprocess.run(["git", "-C", str(path), *args],
                          capture_output=True, check=False).returncode == 0

def _count(path, rng):
    "Commit count for an `a..b` range, 0 when the range does not resolve"
    n = git(path, "rev-list", "--count", rng)
    return int(n) if n.isdigit() else 0

def fetch_all(root=None, workers=8):
    "Fetch every initialised submodule concurrently. This is the slow half of an update"
    base = Path(root) if root else find_root()
    subs = [s for s in parse_gitmodules(base) if (base/s.path/".git").exists()]
    with ThreadPoolExecutor(max_workers=workers) as ex:
        list(ex.map(lambda s: git(base/s.path, "fetch", "-q", "origin"), subs))
    return len(subs)

def update_one(path, sub=None, dry_run=False, branch="main"):
    """Reattach `path` to `branch` and fast-forward it to origin/<branch>.

    Assumes `fetch_all` already ran: fetching is the half that parallelises, and this is
    the half that must not, because a checkout rewrites the worktree."""
    if git(path, "status", "--porcelain"):
        raise StepFailed("update", 1, "uncommitted changes; commit or stash first")
    if not git(path, "remote"): raise Skip("no remote")
    target = f"origin/{branch}"
    if not git(path, "rev-parse", "--verify", "--quiet", target): raise Skip(f"no {target}")

    br   = git(path, "rev-parse", "--abbrev-ref", "HEAD")
    move = br != branch
    # Compare the branch we are ABOUT to be on, not the one we happen to be on now
    ref    = branch if _git_ok(path, "rev-parse", "--verify", "--quiet", branch) else "HEAD"
    behind = _count(path, f"{ref}..{target}")
    ahead  = _count(path, f"{target}..{ref}")
    # Diverged needs a human. The bash version echoed a line and moved on, which is easy
    # to scroll past; this fails the run so the exit code carries it.
    if ahead and behind:
        raise StepFailed("merge --ff-only", 1,
                         f"diverged from {target}: {ahead} ahead, {behind} behind - resolve by hand")
    was = "DETACHED" if br == "HEAD" else br
    if not move and not behind: raise Skip("up to date")
    if dry_run:
        bits = ([f"checkout {branch} (from {was})"] if move else []) + \
               ([f"fast-forward {behind}"] if behind else [])
        return ["would " + ", ".join(bits)]

    done = []
    if move:
        if not _git_ok(path, "checkout", "-q", branch):
            _run("git", "-C", str(path), "checkout", "-q", "-b", branch, target)
        done.append(f"checked out {branch} (was {was})")
    behind = _count(path, f"HEAD..{target}")   # the checkout may already have moved us
    if behind:
        _run("git", "-C", str(path), "merge", "--ff-only", target)
        done.append(f"fast-forwarded {behind} commit{'' if behind == 1 else 's'}")
    return done

def dirty_submodules(root=None):
    "Submodule paths holding uncommitted changes"
    base = Path(root) if root else find_root()
    return [s.path for s in parse_gitmodules(base)
            if (base/s.path/".git").exists() and git(base/s.path, "status", "--porcelain")]

def init_new(root=None, dry_run=False):
    """Initialise only the submodules git reports as uninitialised.

    A blanket `submodule update --init` would reset every initialised submodule back to
    its recorded gitlink and detach its HEAD, which is the one thing this setup avoids."""
    base = Path(root) if root else find_root()
    new = [l.split()[1] for l in git(base, "submodule", "status").splitlines() if l.startswith("-")]
    if new and not dry_run:
        _run("git", "-C", str(base), "submodule", "update", "--init", "--", *new)
    return new

def pull_root(base):
    "Fast-forward the superproject itself, with the advice its failure actually needs"
    try: return _run("git", "-C", str(base), "pull", "--ff-only").strip().splitlines()[-1:]
    except StepFailed as e:
        # Two failures, two fixes. Discarding gitlinks does nothing for a diverged branch,
        # and a rebase does nothing for drifted gitlinks.
        if "Not possible to fast-forward" in e.output:
            hint = ("Local main and origin/main both have new commits.\n"
                    "Replay yours on top of the remote, then re-run:\n"
                    "  git pull --rebase && just sync")
        else:
            hint = ("If it complains about submodule paths, the recorded gitlinks drifted.\n"
                    "Discard them (the update step re-advances them anyway):\n"
                    "  git checkout -- $(git diff --name-only)")
        raise StepFailed(e.cmd, e.code, f"{e.output}\n{hint}") from None

source

init_new

def init_new(
    root:NoneType=None, dry_run:bool=False
):

Initialise only the submodules git reports as uninitialised.

A blanket submodule update --init would reset every initialised submodule back to its recorded gitlink and detach its HEAD, which is the one thing this setup avoids.

source

dirty_submodules

def dirty_submodules(
    root:NoneType=None
):

Submodule paths holding uncommitted changes

source

update_one

def update_one(
    path, sub:NoneType=None, dry_run:bool=False, branch:str='main'
):

Reattach path to branch and fast-forward it to origin/.

Assumes fetch_all already ran: fetching is the half that parallelises, and this is the half that must not, because a checkout rewrites the worktree.

source

fetch_all

def fetch_all(
    root:NoneType=None, workers:int=8
):

Fetch every initialised submodule concurrently. This is the slow half of an update

source

fleet_sync

def fleet_sync(
    dry_run:bool=False, # Report what would move without moving it
    root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):

Bring this checkout level with every remote: pull, init new submodules, fast-forward each

Exported source

def fleet_update(
    dry_run:bool = False,   # Report what would move without moving it
    root:str     = None,    # Superproject root (default: nearest ancestor with .gitmodules)
):
    "Fetch every submodule, then fast-forward each one's main to origin/main"
    base = Path(root) if root else find_root()
    with _step(1, 2, "fetch every submodule") as d:
        d += [f"{fetch_all(base)} fetched"]
    console.print("[dim]2/2[/dim] [bold]fast-forward each main[/bold]")
    rep = walk(lambda p, s: update_one(p, s, dry_run=dry_run), root=base, workers=1, style="line")
    sys.exit(finish(rep, "would update" if dry_run else "updated"))

def fleet_sync(
    dry_run:bool = False,   # Report what would move without moving it
    root:str     = None,    # Superproject root (default: nearest ancestor with .gitmodules)
):
    "Bring this checkout level with every remote: pull, init new submodules, fast-forward each"
    base = Path(root) if root else find_root()
    # Pre-flight FIRST: a dirty submodule turns the checkout/merge below into a conflicted
    # tree, so nothing at all should happen until this passes.
    dirty = dirty_submodules(base)
    if dirty:
        console.print("[bold red]Uncommitted changes in:[/bold red]")
        for d in dirty: console.print(f"  {escape(d)}")
        console.print("Commit or stash them first ([bold]just upload[/bold]), then re-run.")
        sys.exit(1)

    with _step(1, 3, "pull the superproject") as d:
        d += ["skipped (dry run)"] if dry_run else pull_root(base)
    with _step(2, 3, "initialise new submodules") as d:
        new = init_new(base, dry_run=dry_run)
        d += [f"{'would init ' if dry_run else 'initialised '}{', '.join(new)}"] if new else ["none"]
    with _step(3, 3, "fetch every submodule") as d:
        d += [f"{fetch_all(base)} fetched"]
    rep = walk(lambda p, s: update_one(p, s, dry_run=dry_run), root=base, workers=1, style="line")
    sys.exit(finish(rep, "would update" if dry_run else "updated"))

source

fleet_update

def fleet_update(
    dry_run:bool=False, # Report what would move without moving it
    root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):

Fetch every submodule, then fast-forward each one’s main to origin/main

Scanning for personal data

scan is the guardrail for issue #18: personal data reaching a published site. It is the one verb that reads file contents rather than asking git a question, so it is deliberately narrow.

Two things it knows that a plain grep does not. A notebook’s stored outputs are published too, so it scans outputs as well as source - a cleaned-up code cell whose stale output still holds the old figures is exactly the leak that prompted the issue. And it separates published from merely tracked, because the project.render exclude list in _quarto.yml drops p_* notebooks from the site but not from the repo.

source

scan_file

def scan_file(
    path
):

Pattern names found in path, or [] for nothing. Non-text files are never scanned

Exported source

# A finding's severity is which of these two lists a file falls in. `_quarto.yml`
# renders every notebook EXCEPT `p_*`, so "published" is a filename test, not a
# guess: a hit in a published file is live on the web, a hit in a `p_` one is
# only in git.
RAW_DOC_SUFFIXES = {".pdf", ".csv", ".ods", ".xlsx", ".xls", ".db", ".sqlite3"}

# Vendor prefixes that are a secret by construction, whatever they are assigned to.
KNOWN_TOKEN = re.compile(
    r"\b(?:AKIA[0-9A-Z]{16}|gh[pousr]_[A-Za-z0-9]{20,}|glpat-[A-Za-z0-9_-]{20,}"
    r"|sk-[A-Za-z0-9]{20,}|xox[baprs]-[A-Za-z0-9-]{10,})\b")

SECRET_PATTERNS = [
    # The header alone is not a key. Every VPN and SSH page in the fleet shows the
    # armour wrapped around a "(Insert Client Key)" placeholder, so real base64
    # key material has to follow before this counts.
    ("private key",   re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----\s*[\\n\s]*"
                                 r"[A-Za-z0-9+/]{40,}")),
    ("known token",   KNOWN_TOKEN),
    ("password hash", re.compile(r"\b(?:pbkdf2_sha256|argon2[a-z]*)\$"
                                 r"[^\s\"']*\$[A-Za-z0-9+/]{16,}")),
    ("tax file no",   re.compile(r"(?i)\btax file number\b[^\n]{0,20}?\b\d{8,9}\b")),
    ("passport no",   re.compile(r"(?i)\bpassport\b[^\n]{0,30}?\b[A-Z]{1,2}\d{7}\b")),
    ("hex secret",    re.compile(r"(?i)\b(?:secret|token|api[_ -]?key|client[_ -]?id)\b"
                                 r"\W{0,4}\b([0-9a-f]{32,})\b")),
    ("credential",    re.compile(r"(?i)\b(?:password|passwd|api[_ -]?key|secret|token)\b"
                                 r"\s*[:=]\s*[\"']?([^\s\"',}{]{8,})")),
]

def looks_secret(v):
    """Is this captured value a literal secret, or just code that mentions one?

    The credential regex cannot tell `token = 'b4bTbQ...'` from
    `password = forms.CharField(...)`, and the second is what every Django,
    Stripe and Grafana page in the fleet is FULL of. Ruling those out is the
    whole difference between a scan worth running and one everybody ignores.

    So the value has to survive four filters: be a plain literal rather than an
    expression, not be an identifier or env-var name, not already be masked, and
    mix character classes the way a generated secret does and a word does not."""
    if len(v) < 16:                                     return False
    if not re.fullmatch(r"[A-Za-z0-9_\-+/=.~]+", v):    return False  # an expression
    if re.fullmatch(r"[A-Z0-9_]+", v):                  return False  # ENV_VAR_NAME
    if re.fullmatch(r"[A-Za-z_][\w.]*", v) and "." in v:return False  # settings.API_KEY
    if re.search(r"[*x]{4,}|\.{3,}", v):                return False  # already masked
    classes = sum(bool(re.search(c, v)) for c in (r"[a-z]", r"[A-Z]", r"\d"))
    return classes >= 3

# An escape hatch for the deliberately secret-shaped: this module's own test
# vectors, and a doc that has to show what a real token looks like. Put the
# marker on the same line. It is a line comment, so it survives `nbdev-export`.
ALLOW = "scan:allow"

def _line_of(text, pos):
    "The whole line containing offset `pos`, so an inline marker can be seen"
    return text[text.rfind("\n", 0, pos) + 1 : (text.find("\n", pos) + 1 or None)]

def _hits(text):
    "Every secret-pattern name matching `text`, in pattern order, without duplicates"
    out = []
    for name, rx in SECRET_PATTERNS:
        for m in rx.finditer(text):
            if ALLOW in _line_of(text, m.start()): continue
            # A pattern with a group captures the VALUE, so it can be judged.
            # A pattern without one matched something already unambiguous.
            if m.groups() and name == "credential" and not looks_secret(m.group(1)): continue
            out.append(name); break
    return out

def notebook_text(path):
    """Cell sources plus TEXT outputs, with image payloads dropped.

    Scanning outputs is the point, not an extra: a code cell rewritten to drop its
    figures still carries the OLD numbers in its stored output, and Quarto
    publishes that output verbatim. That is the exact leak issue #18 found.

    Skipping `image/*` is not just speed. A notebook of plots is mostly base64,
    and feeding megabytes of it to these regexes manufactures false positives out
    of pure entropy."""
    try: nb = json.loads(path.read_text(encoding="utf-8", errors="replace"))
    except Exception: return ""
    parts = []
    for c in nb.get("cells", []):
        parts.append("".join(c.get("source", [])))
        for o in c.get("outputs", []):
            parts.append("".join(o.get("text", [])))
            for mime, val in (o.get("data") or {}).items():
                if not mime.startswith("image/"):
                    parts.append("".join(val) if isinstance(val, list) else str(val))
    return "\n".join(parts)

# Rendered by `_quarto.yml`, so a hit here is on the public web. Everything else
# tracked is only in git, which is a different and smaller problem.
RENDERED_SUFFIXES = {".ipynb", ".md"}
# Read for content. A binary document cannot be scanned without dragging in a
# parser, so it gets counted instead of judged.
TEXT_SUFFIXES = RENDERED_SUFFIXES | {".py", ".yml", ".yaml", ".txt", ".json", ".csv",
                                     ".ini", ".cfg", ".toml", ".sh", ".env"}

EMAIL = re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b")
# RFC 2606 reserves these for documentation, so a page full of them is a tutorial,
# not a contact list. Without this every mail and auth page in the fleet reports.
RESERVED_DOMAINS = re.compile(r"(?i)@(?:[\w.-]+\.)?(?:example\.(?:com|org|net)|"
                              r"test|localhost|invalid|example)$")
# One address is a contact detail; a column of them is somebody's user table.
BULK_EMAIL_MIN = 5

def is_published(path):
    "Does the site render this file? `p_*` is excluded by `project.render` in `_quarto.yml`"
    return path.suffix in RENDERED_SUFFIXES and not path.name.startswith("p_")

def scan_file(path):
    "Pattern names found in `path`, or [] for nothing. Non-text files are never scanned"
    if path.suffix.lower() not in TEXT_SUFFIXES: return []
    text = notebook_text(path) if path.suffix == ".ipynb" else \
           path.read_text(encoding="utf-8", errors="replace")
    names = _hits(text)
    addrs = {m.group(0).lower() for m in EMAIL.finditer(text)
             if not RESERVED_DOMAINS.search(m.group(0))}
    if len(addrs) >= BULK_EMAIL_MIN: names.append(f"{len(addrs)} email addresses")
    return names

source

is_published

def is_published(
    path
):

Does the site render this file? p_* is excluded by project.render in _quarto.yml

source

notebook_text

def notebook_text(
    path
):

Cell sources plus TEXT outputs, with image payloads dropped.

Scanning outputs is the point, not an extra: a code cell rewritten to drop its figures still carries the OLD numbers in its stored output, and Quarto publishes that output verbatim. That is the exact leak issue #18 found.

Skipping image/* is not just speed. A notebook of plots is mostly base64, and feeding megabytes of it to these regexes manufactures false positives out of pure entropy.

source

looks_secret

def looks_secret(
    v
):

Is this captured value a literal secret, or just code that mentions one?

The credential regex cannot tell token = 'b4bTbQ...' from password = forms.CharField(...), and the second is what every Django, Stripe and Grafana page in the fleet is FULL of. Ruling those out is the whole difference between a scan worth running and one everybody ignores.

So the value has to survive four filters: be a plain literal rather than an expression, not be an identifier or env-var name, not already be masked, and mix character classes the way a generated secret does and a word does not.

source

fleet_scan

def fleet_scan(
    detail:bool=False, # List every finding, not just the per-submodule counts
    strict:bool=False, # Exit 1 on any finding, not only on a published one
    root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):

Find personal data and credentials in tracked files, published ones first

Exported source

def scan_one(path, sub=None, detail=False, nbs="nbs"):
    """Secret and personal-data findings for one submodule.

    Only TRACKED files are scanned. An untracked file is on this disk and nowhere
    else, which is exactly where `.gitignore` is supposed to put a payslip, so
    flagging it would punish the fix.

    Binary documents are counted rather than read. The count is informational and
    never fails the scan: a tracked PDF may be a payslip or may be a datasheet,
    and this cannot tell without a parser it has no business importing."""
    root = path/nbs
    if not root.is_dir(): raise Skip("without nbs")
    tracked = [l for l in git(path, "ls-files", "-z", nbs).split("\0") if l]
    pub, tracked_hits, docs = [], [], 0
    for rel in sorted(tracked):
        f = path/rel
        if not f.is_file(): continue
        if f.suffix.lower() in RAW_DOC_SUFFIXES and f.suffix.lower() != ".csv":
            docs += 1; continue
        names = scan_file(f)
        if not names: continue
        (pub if is_published(f) else tracked_hits).append(f"{rel}: {', '.join(names)}")
    if not (pub or tracked_hits): raise Skip("clean")
    line = f"{len(pub):>3} published {len(tracked_hits):>3} tracked {docs:>4} binary docs"
    if not detail: return [line]
    return [line] + [f"PUBLISHED {h}" for h in pub] + [f"tracked   {h}" for h in tracked_hits]

def fleet_scan(
    detail:bool = False,  # List every finding, not just the per-submodule counts
    strict:bool = False,  # Exit 1 on any finding, not only on a published one
    root:str    = None,   # Superproject root (default: nearest ancestor with .gitmodules)
):
    "Find personal data and credentials in tracked files, published ones first"
    style = "block" if detail else "line"
    rep = walk(lambda p, s: scan_one(p, s, detail=detail), root=root,
               workers=8, style=style)
    code = finish(rep, "with findings")
    if code: sys.exit(code)
    # A published hit is live on the web and always fails. A tracked-but-unpublished
    # one is a cleanup job, so it only fails under --strict.
    published = sum(int(o.detail[0].split()[0]) for o in rep.ran if o.detail)
    sys.exit(1 if published or (strict and rep.ran) else 0)

source

scan_one

def scan_one(
    path, sub:NoneType=None, detail:bool=False, nbs:str='nbs'
):

Secret and personal-data findings for one submodule.

Only TRACKED files are scanned. An untracked file is on this disk and nowhere else, which is exactly where .gitignore is supposed to put a payslip, so flagging it would punish the fix.

Binary documents are counted rather than read. The count is informational and never fails the scan: a tracked PDF may be a payslip or may be a datasheet, and this cannot tell without a parser it has no business importing.

Linting the notebook standard

nblint measures how far the public notebooks are from the standard in the Knowledge repo’s AGENTS.md and .github/instructions/, one row per rule. It is the scoreboard for issue #24: each rule carries the phase of that plan that drives it to 0.

It reads every public notebook on disk (not only tracked ones, because a new notebook is exactly what should be checked before it is committed), skips p_* notebooks and build directories, and never looks at Personal/bthek1_career, which has no site. It always exits 0 unless --strict is given, so it is a measurement, not a gate.

source

fleet_nblint

def fleet_nblint(
    detail:bool=False, # List the offending notebooks under each rule
    phase:int=None, # Only the rules for this phase of issue #24
    strict:bool=False, # Exit 1 when any rule is above 0
    root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):

Count public notebooks that break the notebook standard, one row per rule

Exported source

# (key, label, issue #24 phase). The phase is what a count is FOR: each phase of the
# cleanup drives its own rows to 0, so the table doubles as the plan's progress bar.
NB_RULES = [
    ("mermaid_plain",     "Plain ```mermaid fence (raw source on the site)", 1),
    ("title_leading_rule","Title cell starts with --- (breaks nbdev-test)",   1),
    ("title_no_h1",       "Title cell has no # H1",                          1),
    ("index_unlinked",    "Public notebook missing from index.ipynb",        1),
    ("mermaid_braced",    "Braced {mermaid} not yet an .svg file",           2),
    ("svg_in_index",      "Inline <svg> in index.ipynb (README strips it)",  2),
    ("title_code",        "Title cell is a code cell",                       3),
    ("title_no_desc",     "Title cell has no > description",                 3),
    ("no_skip_showdoc",   "Concept notebook without skip_showdoc",           3),
    ("non_ascii",         "Non-ASCII characters in code cells",              4),
    ("naive_datetime",    "Naive now() or utcnow(), no timezone",            4),
    ("bad_filename",      "Filename with a space or &",                      4),
    ("dl_matplotlib",     "DL_methods chart in matplotlib",                  5),
    ("dltask_sections",   "DL_tasks notebook missing a required section",    5),
    ("dltask_no_require", "DL_tasks live-capture demo without require(...)", 5),
    ("stale_outputs",     "skip_exec notebook carrying outputs",             6),
    ("large_file",        "Notebook over 3 MB",                              6),
    ("multi_h2",          "Markdown cell with several ## sections",          7),
    ("no_trailing_rule",  "## cell without a trailing ---",                  7),
]
# Matched case-insensitively against the `##` headings, as substrings, so "Datasets"
# also accepts "Datasets and splits". From dl-task-notebook-format.instructions.md.
DLTASK_SECTIONS = ["What is", "Real-world use cases", "Evaluation", "Datasets",
                   "Setup", "benchmark", "Common frameworks", "Going further"]
DLTASK_NOT_TASKS = ("Theory", "Frameworks")
LARGE_NB_BYTES = 3_000_000
# Never a site, so never held to the publishing standard (see special-submodules).
NOT_PUBLISHED = ("Personal/bthek1_career",)

def _src(c):
    s = c.get("source", "")
    return "".join(s) if isinstance(s, list) else s

def lint_notebook(path, sub=""):
    """{rule key: count} for one notebook. A count is the number of offending items
    (diagrams, characters, cells); 1 for a yes/no rule. Empty means it meets the standard."""
    out = {}
    name = path.name
    if " " in name or "&" in name: out["bad_filename"] = 1
    try: nb = json.loads(path.read_text(encoding="utf-8"))
    except Exception: return out
    cells = nb.get("cells", [])
    if not cells: return out
    code = [_src(c) for c in cells if c.get("cell_type") == "code"]
    md   = [_src(c) for c in cells if c.get("cell_type") == "markdown"]
    allcode, allmd = "\n".join(code), "\n".join(md)
    exports = re.search(r"^#\|\s*(export|exporti|exports|default_exp)\b", allcode, re.M)
    first = _src(cells[0]).strip()

    if cells[0].get("cell_type") != "markdown": out["title_code"] = 1
    else:
        if first.startswith("---"): out["title_leading_rule"] = 1
        if not re.match(r"^(---[\s\S]*?---\s*)?#\s+\S", first): out["title_no_h1"] = 1
        if not re.search(r"^>\s*\S", first, re.M): out["title_no_desc"] = 1
        if not exports and "skip_showdoc: true" not in first: out["no_skip_showdoc"] = 1

    # The title cell is checked above; a `##` packed into it is a title problem first.
    multi = sum(len(re.findall(r"^##\s", m, re.M)) > 1 for m in md[1:])
    trail = sum(bool(re.search(r"^##\s", m, re.M)) and not m.rstrip().endswith("---")
                for m in md[1:])
    if multi: out["multi_h2"] = multi
    if trail: out["no_trailing_rule"] = trail

    plain = len(re.findall(r"^```mermaid\s*$", allmd, re.M))
    brace = len(re.findall(r"^```\{mermaid\}", allmd, re.M))
    if plain: out["mermaid_plain"] = plain
    if brace: out["mermaid_braced"] = brace
    if name == "index.ipynb" and re.search(r"<svg[\s>]", allmd): out["svg_in_index"] = 1

    nonascii = sum(ord(ch) > 127 for ch in allcode)
    if nonascii: out["non_ascii"] = nonascii
    if re.search(r"datetime\.utcnow\(|\b(?:dt|datetime)\.now\(\s*\)", allcode):
        out["naive_datetime"] = 1

    if sub == "DL/DL_methods" and re.search(
            r"^\s*(import|from)\s+(matplotlib|seaborn|plotly)", allcode, re.M):
        out["dl_matplotlib"] = 1
    # Only a real task notebook: not the index, a library notebook, a stub, or the
    # Theory/ and Frameworks/ reference pages, which have no task to benchmark.
    if (sub == "DL/DL_tasks" and name != "index.ipynb" and not exports and len(cells) > 3
            and path.parent.name not in DLTASK_NOT_TASKS):
        heads = "\n".join(re.findall(r"^##\s+.*$", allmd, re.M)).lower()
        missing = [s for s in DLTASK_SECTIONS if s.lower() not in heads]
        if missing: out["dltask_sections"] = len(missing)
        if re.search(r"cv2\.VideoCapture|sounddevice|pyaudio", allcode) and "require(" not in allcode:
            out["dltask_no_require"] = 1

    # DL notebooks keep outputs on purpose (benchmarks, demos), so only elsewhere is a
    # skip_exec notebook with outputs a sign of stale results.
    outs = sum(len(c.get("outputs", [])) for c in cells if c.get("cell_type") == "code")
    if outs and "skip_exec: true" in first and not sub.startswith("DL/"):
        out["stale_outputs"] = outs
    if path.stat().st_size > LARGE_NB_BYTES: out["large_file"] = 1
    return out

def _public_notebooks(nbs):
    "Every published notebook under `nbs`, relative paths. `_`/`.` dirs are build output"
    found = []
    for p in sorted(nbs.rglob("*.ipynb")):
        rel = p.relative_to(nbs)
        if any(part.startswith((".", "_")) for part in rel.parts[:-1]): continue
        if p.name.lower().startswith("p_"): continue
        found.append(rel)
    return found

def unlinked_from_index(nbs, public=None):
    "Public notebooks (as posix paths) that `nbs/index.ipynb` does not link to"
    idx = nbs/"index.ipynb"
    if not idx.exists(): return []
    public = public if public is not None else _public_notebooks(nbs)
    body = "".join(_src(c) for c in json.loads(idx.read_text(encoding="utf-8")).get("cells", []))
    links = {unquote(l).split("#")[0] for l in re.findall(r"\]\(([^)\s]+)\)", body)
             if not l.startswith(("http", "#"))}
    return [r.as_posix() for r in public
            if r.name != "index.ipynb" and r.as_posix() not in links]

def lint_one(path, sub=None, nbs="nbs"):
    "{notebook path relative to the superproject: {rule key: count}} for one submodule"
    name = sub.path if sub else path.name
    root = path/nbs
    if name in NOT_PUBLISHED: raise Skip("not published")
    if not root.is_dir(): raise Skip("without nbs")
    public = _public_notebooks(root)
    found = {}
    for rel in public:
        hits = lint_notebook(root/rel, name)
        if hits: found[f"{name}/{nbs}/{rel.as_posix()}"] = hits
    for rel in unlinked_from_index(root, public):
        found.setdefault(f"{name}/{nbs}/{rel}", {})["index_unlinked"] = 1
    return found

def lint_totals(found):
    "{rule key: (notebooks, items)} over the output of `lint_one`, every rule present"
    tot = {k: [0, 0] for k, _, _ in NB_RULES}
    for hits in found.values():
        for k, n in hits.items(): tot[k][0] += 1; tot[k][1] += n
    return {k: tuple(v) for k, v in tot.items()}

def fleet_nblint(
    detail:bool = False,  # List the offending notebooks under each rule
    phase:int   = None,   # Only the rules for this phase of issue #24
    strict:bool = False,  # Exit 1 when any rule is above 0
    root:str    = None,   # Superproject root (default: nearest ancestor with .gitmodules)
):
    "Count public notebooks that break the notebook standard, one row per rule"
    from rich.table import Table
    found = {}
    def _collect(p, s):
        found.update(lint_one(p, s))
        raise Skip("linted")   # counted below in one table, not narrated per submodule
    rep = walk(_collect, root=root, workers=8, style="line")
    if rep.failed: sys.exit(finish(rep, "linted"))
    rules = [r for r in NB_RULES if phase is None or r[2] == phase]
    tot = lint_totals(found)
    t = Table(title="Notebook standard (issue #24)", title_justify="left")
    for col, kw in (("Phase", {"justify": "right"}), ("Rule", {}),
                    ("Notebooks", {"justify": "right"}), ("Items", {"justify": "right"})):
        t.add_column(col, **kw)
    for key, label, ph in rules:
        nbks, items = tot[key]
        style = "green" if nbks == 0 else None
        t.add_row(str(ph), escape(label), str(nbks), str(items), style=style)
    console.print(t)
    left = sum(tot[k][0] for k, _, _ in rules)
    console.print(f"[bold]{len(found)} notebooks flagged, {left} rule hits in view[/bold]")
    if detail:
        for key, label, _ in rules:
            hits = sorted((p, h[key]) for p, h in found.items() if key in h)
            if not hits: continue
            console.print(f"\n[bold cyan]{escape(label)}[/bold cyan]")
            for p, n in hits: console.print(f"  {escape(p)} [dim]({n})[/dim]")
    sys.exit(1 if strict and left else 0)

source

lint_totals

def lint_totals(
    found
):

{rule key: (notebooks, items)} over the output of lint_one, every rule present

source

lint_one

def lint_one(
    path, sub:NoneType=None, nbs:str='nbs'
):

{notebook path relative to the superproject: {rule key: count}} for one submodule

source

unlinked_from_index

def unlinked_from_index(
    nbs, public:NoneType=None
):

Public notebooks (as posix paths) that nbs/index.ipynb does not link to

source

lint_notebook

def lint_notebook(
    path, sub:str=''
):

{rule key: count} for one notebook. A count is the number of offending items (diagrams, characters, cells); 1 for a yes/no rule. Empty means it meets the standard.

The `fleet` entry point

source

fleet

def fleet():

Dispatch fleet <verb> to the matching command.

upload and status are already console scripts bound to the single-repo versions in automate, so the fleet-wide ones cannot claim those names. One script with verbs sidesteps the collision and keeps --help working per verb.

Exported source

VERBS = {"upload": fleet_upload, "push": fleet_push, "copy": fleet_copy,
         "status": fleet_status, "update": fleet_update, "sync": fleet_sync,
         "scan": fleet_scan, "nblint": fleet_nblint}

def parse_verb(argv):
    "(verb, remaining args) for `fleet ...`, or None when `argv` does not name a verb"
    if not argv or argv[0].startswith("-"): return None
    return (argv[0], argv[1:]) if argv[0] in VERBS else None

def usage():
    "The verb list, for --help and for an unknown verb"
    lines = ["Usage: fleet <verb> [options]", "", "Verbs:"]
    lines += [f"  {name:<8} {(fn.__doc__ or '').strip()}" for name, fn in sorted(VERBS.items())]
    lines += ["", "Any verb takes --help for its own options."]
    return "\n".join(lines)

def run_verb(name, argv):
    """Parse `argv` against the verb's own annotations, then call it.

    Deliberately NOT `@call_parse`: fastcore runs the first decorated function in a module
    the moment that module is `__main__`, which is exactly what makes `python -m
    nbdevAuto.github` work with its single verb. With several verbs it would run whichever
    happened to be defined first, so the verbs stay plain functions and the parser is built
    here. `anno_parser` is the same one `call_parse` uses, so `--help` is unchanged."""
    fn = VERBS[name]
    args = vars(anno_parser(fn, prog=f"fleet {name}").parse_args(argv))
    for k in ("pdb", "xtra"): args.pop(k, None)
    return fn(**args)

def fleet():
    """Dispatch `fleet <verb>` to the matching command.

    `upload` and `status` are already console scripts bound to the single-repo versions in
    `automate`, so the fleet-wide ones cannot claim those names. One script with verbs
    sidesteps the collision and keeps `--help` working per verb."""
    picked = parse_verb(sys.argv[1:])
    if not picked:
        print(usage())
        sys.exit(0 if sys.argv[1:2] in ([], ["-h"], ["--help"]) else 2)
    # `_step` has already printed the failing command's output, so a traceback on top of
    # it only buries the hint. One line and the exit code are all that is left to say.
    try: return run_verb(*picked)
    except StepFailed as e:
        console.print(f"[bold red]FAILED[/bold red] [dim]{escape(str(e))}[/dim]")
        sys.exit(1)

source

run_verb

def run_verb(
    name, argv
):

Parse argv against the verb’s own annotations, then call it.

Deliberately NOT @call_parse: fastcore runs the first decorated function in a module the moment that module is __main__, which is exactly what makes python -m nbdevAuto.github work with its single verb. With several verbs it would run whichever happened to be defined first, so the verbs stay plain functions and the parser is built here. anno_parser is the same one call_parse uses, so --help is unchanged.

source

usage

def usage():

The verb list, for –help and for an unknown verb

source

parse_verb

def parse_verb(
    argv
):

(verb, remaining args) for fleet ..., or None when argv does not name a verb

python -m nbdevAuto.fleet is the other way in, and the one the justfile falls back to when the installed nbdevAuto predates this module. #|export puts the guard in the library; #|eval: false keeps nbdev_test from running the CLI while testing the notebook.

Tests

A throwaway superproject of real git repos, rebuilt per run.

from fastcore.test import *
import tempfile

def _fixture(specs=(("a_dirty", True, True), ("b_clean", True, False), ("c_nonbs", False, False))):
    "A throwaway superproject: (name, has nbs/, is dirty) per submodule"
    d = Path(tempfile.mkdtemp())
    lines = []
    for name, nbs, dirty in specs:
        p = d/name; p.mkdir()
        subprocess.run(["git", "init", "-q", "-b", "main", str(p)], check=True)
        for k, v in (("user.email", "t@example.com"), ("user.name", "t")):
            subprocess.run(["git", "-C", str(p), "config", k, v], check=True)
        (p/"README.md").write_text("x")
        if nbs: (p/"nbs").mkdir()
        subprocess.run(["git", "-C", str(p), "add", "."], check=True)
        subprocess.run(["git", "-C", str(p), "commit", "-q", "-m", "init"], check=True)
        if dirty: (p/"README.md").write_text("changed")
        lines.append(f'[submodule "{name}"]\n\tpath = {name}\n\turl = git@github.com:me/{name}.git\n')
    (d/".gitmodules").write_text("".join(lines))
    return d

def _with_origin(p, extra=1):
    "Give the repo at `p` a real bare origin, then leave it `extra` commits ahead of it"
    bare = Path(f"{p}.origin.git")
    subprocess.run(["git", "init", "-q", "--bare", str(bare)], check=True)
    subprocess.run(["git", "-C", str(p), "remote", "add", "origin", str(bare)], check=True)
    subprocess.run(["git", "-C", str(p), "push", "-q", "-u", "origin", "main"], check=True)
    for i in range(extra):
        (p/f"extra{i}.txt").write_text("x")
        subprocess.run(["git", "-C", str(p), "add", f"extra{i}.txt"], check=True)
        subprocess.run(["git", "-C", str(p), "commit", "-q", "-m", f"extra{i}"], check=True)
    return bare

_root = _fixture()

def scan_file_text(text):
    "Run the content scanners over a string, so a test need not make a file"
    names = _hits(text)
    addrs = {m.group(0).lower() for m in EMAIL.finditer(text)
             if not RESERVED_DOMAINS.search(m.group(0))}
    if len(addrs) >= BULK_EMAIL_MIN: names.append(f"{len(addrs)} email addresses")
    return names

# Only the dirty submodule with an nbs/ folder runs; the other two are counted, not narrated.
seen = []
rep = walk(lambda p, s: seen.append(s.path), root=_root, need_nbs=True, only_dirty=True, style="line")
test_eq(seen, ["a_dirty"])
test_eq(len(rep.ran), 1)
test_eq(rep.skip_counts(), {"without nbs": 1, "clean": 1})
test_eq(format_summary(rep, "uploaded"), "3 submodules: 1 uploaded, 1 clean, 1 without nbs")
test_eq(rep.code, 0)

# Dropping the dirty filter picks up both nbs/ submodules.
rep = walk(lambda p, s: None, root=_root, need_nbs=True, style="line")
test_eq([o.path for o in rep.ran], ["a_dirty", "b_clean"])
test_eq(format_summary(rep, "uploaded"), "3 submodules: 2 uploaded, 1 without nbs")

# A failure is named and counted, does not abort the others, and sets the exit code.
def _boom(p, s):
    if s.path == "a_dirty": raise StepFailed("upload", 2, "fatal: nope")
    return ["fine"]
rep = walk(_boom, root=_root, need_nbs=True, style="line")
test_eq([o.path for o in rep.failed], ["a_dirty"])
# A StepFailed's own message is the headline, because "upload exited 2" says nothing
test_eq(rep.failed[0].reason, "fatal: nope")
test_eq([o.path for o in rep.ran], ["b_clean"])
test_eq(rep.code, 1)
test_eq(format_summary(rep, "uploaded"), "3 submodules: 1 uploaded, 1 without nbs, 1 FAILED")

# With no output to quote, the command and exit code are the fallback headline.
def _silent(p, s): raise StepFailed("upload", 2)
test_eq(walk(_silent, root=_root, need_nbs=True, style="line").failed[0].reason, "upload exited 2")

# An uninitialised submodule is a directory with no .git, and is reported rather than run.
_bare = _fixture()
import shutil as _sh; _sh.rmtree(_bare/"b_clean"/".git")
test_eq(walk(lambda p, s: None, root=_bare, style="line").skip_counts(), {"not initialised": 1})

# Parallel and sequential runs agree, and both report in submodule order.
rep_p = walk(lambda p, s: [s.path], root=_root, workers=4, style="line")
rep_s = walk(lambda p, s: [s.path], root=_root, workers=1, style="line")
test_eq([o.path for o in rep_p.outcomes], [o.path for o in rep_s.outcomes])
test_eq([o.detail for o in rep_p.outcomes], [o.detail for o in rep_s.outcomes])

a_dirty   
a_dirty   
b_clean   
a_dirty   fatal: nope
b_clean   fine
a_dirty   upload exited 2
b_clean   upload exited 2
a_dirty   
c_nonbs   
a_dirty   a_dirty
b_clean   b_clean
c_nonbs   c_nonbs
a_dirty   a_dirty
b_clean   b_clean
c_nonbs   c_nonbs

# Verb dispatch: a verb, its args, and everything that is not one.
test_eq(parse_verb(["upload", "--all"]), ("upload", ["--all"]))
test_eq(parse_verb(["upload"]), ("upload", []))
test_eq(parse_verb([]), None)
test_eq(parse_verb(["--help"]), None)
test_eq(parse_verb(["nosuchverb"]), None)
assert "upload" in usage() and "Usage: fleet" in usage()

# `run_verb` parses against the verb's own signature. Checked on a throwaway verb so the
# test does not have to run a real upload.
def _echo(all:bool=False, root:str=None):
    "Echo verb, for tests"
    return (all, root)
VERBS["_echo"] = _echo
try:
    test_eq(run_verb("_echo", []), (False, None))
    test_eq(run_verb("_echo", ["--all"]), (True, None))
    test_eq(run_verb("_echo", ["--root", "/tmp/x"]), (False, "/tmp/x"))
    test_eq(parse_verb(["_echo", "--all"]), ("_echo", ["--all"]))
finally:
    del VERBS["_echo"]

# `script_path` says what to do rather than failing obscurely.
try:
    script_path("definitely-not-a-console-script-xyz")
    assert False, "a missing script must raise"
except RuntimeError as e:
    assert "uv sync" in str(e)

Phase 2: push and copy

# --- push -------------------------------------------------------------------------
_r2 = _fixture()
_with_origin(_r2/"b_clean", extra=2)
test_eq(ahead_of_remote(_r2/"b_clean"), (2, "main"))
test_eq(push_one(_r2/"b_clean", dry_run=True), ["would push 2 commits on main"])
# dry_run must not move the remote
test_eq(ahead_of_remote(_r2/"b_clean"), (2, "main"))
push_one(_r2/"b_clean")
test_eq(ahead_of_remote(_r2/"b_clean"), (0, "main"))

# The three reasons a submodule is passed over, each named rather than silently ignored.
def _skip_reason(fn, *a, **kw):
    try: fn(*a, **kw); return None
    except Skip as e: return str(e)
test_eq(_skip_reason(push_one, _r2/"b_clean"), "nothing to push")
test_eq(_skip_reason(push_one, _r2/"a_dirty"), "no remote")
subprocess.run(["git", "-C", str(_r2/"c_nonbs"), "checkout", "-q", "--detach"], check=True)
test_eq(_skip_reason(push_one, _r2/"c_nonbs"), "detached HEAD")

# A runtime Skip is counted, not printed, and never a failure.
rep = walk(lambda p, s: push_one(p, s, dry_run=True), root=_r2, style="line")
test_eq(rep.ran, [])
test_eq(rep.skip_counts(), {"no remote": 1, "nothing to push": 1, "detached HEAD": 1})
test_eq(rep.code, 0)

# --- copy -------------------------------------------------------------------------
_r3 = _fixture()
_shared = _r3/"shared.yml"; _shared.write_text("a: 1")
test_eq(copy_one(_r3/"a_dirty", files=[_shared], dry_run=True), ["would copy shared.yml -> nbs/"])
assert not (_r3/"a_dirty"/"nbs"/"shared.yml").exists(), "dry_run wrote a file"
test_eq(copy_one(_r3/"a_dirty", files=[_shared]), ["copied shared.yml -> nbs/"])
test_eq((_r3/"a_dirty"/"nbs"/"shared.yml").read_text(), "a: 1")

# No destination directory is a skip; a missing SOURCE is a failure, not a quiet no-op.
test_eq(_skip_reason(copy_one, _r3/"c_nonbs", files=[_shared]), "no nbs/")
try:
    copy_one(_r3/"a_dirty", files=["/nope/missing.yml"])
    assert False, "a missing source must fail"
except StepFailed as e:
    assert "no such file" in e.output

rep = walk(lambda p, s: copy_one(p, s, files=[_shared]), root=_r3, workers=4, style="line")
test_eq(len(rep.ran), 2)
test_eq(format_summary(rep, "updated"), "3 submodules: 2 updated, 1 no nbs/")

Phases 4 and 5: status, update, sync

# --- status -----------------------------------------------------------------------
_r4 = _fixture()
_line = status_one(_r4/"a_dirty")[0]
assert _line.startswith("main") and "dirty" in _line
assert "clean" in status_one(_r4/"b_clean")[0]
# --short lists the changed paths, and passes over the clean ones
_out = status_one(_r4/"a_dirty", short=True)
assert any("README.md" in l for l in _out[1:]), _out
# The FIRST porcelain line must not be shifted: `git status -s` pads the code to two
# columns and a strip would eat that leading space on line one only.
test_eq([l[:5] for l in _out[1:]], ["   M "] * len(_out[1:]))
test_eq(_skip_reason(status_one, _r4/"b_clean", short=True), "clean")
rep = walk(lambda p, s: status_one(p, s, short=True), root=_r4, style="line")
test_eq(format_summary(rep, "changed"), "3 submodules: 1 changed, 2 clean")

# A detached HEAD is named, not shown as a branch called HEAD
subprocess.run(["git", "-C", str(_r4/"c_nonbs"), "checkout", "-q", "--detach"], check=True)
assert status_one(_r4/"c_nonbs")[0].startswith("DETACHED")

# --- update -----------------------------------------------------------------------
_r5 = _fixture()
_p = _r5/"b_clean"
_with_origin(_p, extra=2)
subprocess.run(["git", "-C", str(_p), "push", "-q"], check=True)          # remote now level
subprocess.run(["git", "-C", str(_p), "reset", "-q", "--hard", "HEAD~2"], check=True)
test_eq(_count(_p, "HEAD..origin/main"), 2)                                # 2 behind

test_eq(update_one(_p, dry_run=True), ["would fast-forward 2"])
test_eq(_count(_p, "HEAD..origin/main"), 2)                                # dry run moved nothing
test_eq(update_one(_p), ["fast-forwarded 2 commits"])
test_eq(_count(_p, "HEAD..origin/main"), 0)
test_eq(_skip_reason(update_one, _p), "up to date")

# Diverged is a FAILURE, not a quiet skip: it needs a human and the exit code must say so.
subprocess.run(["git", "-C", str(_p), "reset", "-q", "--hard", "HEAD~1"], check=True)
(_p/"mine.txt").write_text("mine")
subprocess.run(["git", "-C", str(_p), "add", "mine.txt"], check=True)
subprocess.run(["git", "-C", str(_p), "commit", "-q", "-m", "mine"], check=True)
try:
    update_one(_p)
    assert False, "a diverged submodule must fail"
except StepFailed as e:
    assert "diverged" in e.output and "resolve by hand" in e.output

# The same divergence on the superproject gets the rebase advice, not the gitlink advice.
try:
    pull_root(_p)
    assert False, "a diverged superproject must fail"
except StepFailed as e:
    assert "git pull --rebase" in e.output and "checkout --" not in e.output, e.output

# A dirty worktree is refused before anything is checked out.
try:
    update_one(_r5/"a_dirty")
    assert False, "a dirty submodule must fail"
except StepFailed as e:
    assert "uncommitted changes" in e.output
test_eq(_skip_reason(update_one, _r5/"c_nonbs"), "no remote")

# A detached HEAD is reattached to main.
_q = _fixture()/"a_dirty"
subprocess.run(["git", "-C", str(_q), "checkout", "-q", "--", "README.md"], check=True)
_with_origin(_q, extra=0)
subprocess.run(["git", "-C", str(_q), "checkout", "-q", "--detach"], check=True)
test_eq(update_one(_q, dry_run=True), ["would checkout main (from DETACHED)"])
test_eq(update_one(_q), ["checked out main (was DETACHED)"])
test_eq(git(_q, "rev-parse", "--abbrev-ref", "HEAD"), "main")

# --- sync helpers -----------------------------------------------------------------
test_eq(dirty_submodules(_r4), ["a_dirty"])
# init_new only ever touches submodules git itself calls uninitialised
test_eq(init_new(_r4, dry_run=True), [])

The `scan` verb

# --- looks_secret: the filter that decides whether the scan is worth running ----
# Every one of these false positives is a REAL line from the fleet's published
# teaching notebooks. They are the regression test, not hypotheticals.
for expr in ["forms.CharField(widget=forms.PasswordInput)",  # Django forms page
             "settings.STRIPE_SECRET_KEY",                   # Stripe page
             "cleaned_data.get(",                            # Django forms page
             "&&(i[0]+=t._mapboxAccessToken",                # minified JS in a Plotly output
             "ghp_************************************",     # already masked
             "YOUR_API_KEY", "changeme", "short"]:
    test_eq((expr, looks_secret(expr)), (expr, False))

# A generated secret mixes case and digits over a decent length; a word does not.
for secret in ["b4bTbQgLdRCvoBZQSjaRlUh8lnslIflVDFet6VeToen4GWu-MYwHilUddyWZorKPFCSFiu1KL",  # scan:allow
               "EAAYp9C2YZAYYBO93tDbfKZAa5vp3ZB0fCh8yZCVv6VuWof1GxfZArv83OUlWoGI4jRYn1T6"]:  # scan:allow
    test_eq(looks_secret(secret), True)

# --- is_published mirrors the project.render exclude list in _quarto.yml -------
test_eq(is_published(Path("nbs/01_Tax.ipynb")),   True)
test_eq(is_published(Path("nbs/index.ipynb")),    True)
test_eq(is_published(Path("nbs/p_04_Pay.ipynb")), False)  # excluded by project.render
test_eq(is_published(Path("nbs/data.csv")),       False)  # not a rendered suffix

# --- scan_file reads a notebook's OUTPUTS, not just its source ----------------
# The leak that prompted issue #18: the code cell was rewritten, the stale output
# holding the old figures was not, and Quarto publishes the output verbatim.
_d = Path(tempfile.mkdtemp())
_leaky = {"cells": [{"cell_type": "code", "source": ["print(summary)\n"], "outputs": [  # scan:allow
    {"output_type": "stream", "text": ["token = 'aB3dEfGh1jKlMn0pQrStUvWxYz456789'\n"]}]}]}  # scan:allow
(_d/"01_stale.ipynb").write_text(json.dumps(_leaky))
test_eq(scan_file(_d/"01_stale.ipynb"), ["credential"])

# An image payload is skipped: megabytes of base64 manufacture entropy hits.
_plot = {"cells": [{"cell_type": "code", "source": [], "outputs": [
    {"output_type": "display_data", "data": {"image/png": "iVBORw0KGgoAAAANSUhEUg" * 500}}]}]}
(_d/"02_plot.ipynb").write_text(json.dumps(_plot))
test_eq(scan_file(_d/"02_plot.ipynb"), [])

# A column of addresses is someone's user table; one address is a contact detail.
# A real domain: the reserved ones are excluded, which the next cell tests.
(_d/"users.csv").write_text("email\n" + "".join(f"u{i}@clinic.com.au\n" for i in range(9)))
test_eq(scan_file(_d/"users.csv"), ["9 email addresses"])
(_d/"one.csv").write_text("owner\nben@clinic.com.au\n")
test_eq(scan_file(_d/"one.csv"), [])

# A binary document is never read, so it can never be a content finding.
(_d/"payslip.pdf").write_bytes(b"%PDF-1.4 token = aB3dEfGh1jKlMn0pQrStUvWxYz456789")  # scan:allow
test_eq(scan_file(_d/"payslip.pdf"), [])

# --- the three filters that stopped the fleet's own docs from reporting -------
# RFC 2606 addresses are a mail tutorial, not a contact list.
test_eq(scan_file_text("".join(f"u{i}@example.com\n" for i in range(9))), [])
# A placeholder hash is hash-shaped but carries no material.
test_eq(scan_file_text("password: argon2:$argon2id$v=19$m=10240,t=10,p=8$<salt>$<hash>"), [])
test_eq(scan_file_text("password: argon2:$argon2id$v=19$m=1$75Q7uCcjzIK2HqrZ$"  # scan:allow
                       "y8eYdSDFeAEAZW0S74AqnSMq3CAIl4hSBpdgkSVg"), ["password hash"])
# The armour without key material is every VPN page in the fleet.
test_eq(scan_file_text("-----BEGIN PRIVATE KEY-----\n(Insert Client Key)"), [])
# An inline marker is how this notebook stops flagging its own test vectors.
test_eq(scan_file_text("api_key = 'aB3dEfGh1jKlMn0pQrStUvWxYz456789'"), ["credential"])  # scan:allow
test_eq(scan_file_text("api_key = 'aB3dEfGh1jKlMn0pQrStUvWxYz456789'  # scan:allow"), [])

# --- scan_one: only TRACKED files count, and only submodules with nbs/ ---------
_r6 = _fixture(specs=(("leaky", True, False), ("clean", True, False), ("nonbs", False, False)))
_leak = _r6/"leaky"/"nbs"
(_leak/"01_public.ipynb").write_text(json.dumps(  # scan:allow
    {"cells": [{"cell_type": "code", "source": ["api_key = 'aB3dEfGh1jKlMn0pQrStUvWxYz456789'\n"],  # scan:allow
                "outputs": []}]}))
(_leak/"p_02_private.ipynb").write_text(json.dumps(
    {"cells": [{"cell_type": "code", "source": ["api_key = 'zY9xWv8uTs7rQp6oNm5lKj4iHg3fEd2c'\n"],  # scan:allow
                "outputs": []}]}))
(_leak/"receipts.pdf").write_bytes(b"%PDF-1.4")
subprocess.run(["git", "-C", str(_r6/"leaky"), "add", "-A"], check=True)
subprocess.run(["git", "-C", str(_r6/"leaky"), "commit", "-q", "-m", "data"], check=True)

# Untracked: on this disk and nowhere else, which is where .gitignore is meant to
# put a payslip. Flagging it would punish the fix.
(_leak/"99_untracked.ipynb").write_text(json.dumps(
    {"cells": [{"cell_type": "code", "source": ["api_key = 'Qq1Ww2Ee3Rr4Tt5Yy6Uu7Ii8Oo9Pp0Aa'\n"],  # scan:allow
                "outputs": []}]}))

_out = scan_one(_r6/"leaky", detail=True)
test_eq(_out[0], f"{1:>3} published {1:>3} tracked {1:>4} binary docs")
test_eq(_out[1], "PUBLISHED nbs/01_public.ipynb: credential")
test_eq(_out[2], "tracked   nbs/p_02_private.ipynb: credential")
test_eq(len(_out), 3)                                  # the untracked file is absent

# A submodule with nothing to report skips, so the fleet line never appears for it.
test_fail(lambda: scan_one(_r6/"clean"), contains="clean")
test_fail(lambda: scan_one(_r6/"nonbs"), contains="without nbs")

# The walk reports only the submodule that found something; the rest are counted.
_rep6 = walk(lambda p, s: scan_one(p, s), root=_r6, style="line")
test_eq([o.path for o in _rep6.ran], ["leaky"])
test_eq(_rep6.skip_counts(), {"clean": 1, "without nbs": 1})
test_eq(format_summary(_rep6, "with findings"),
        "3 submodules: 1 with findings, 1 clean, 1 without nbs")

The `nblint` verb

# --- nblint: one fixture per rule, so a rule that stops firing fails here ---------
def _nb(*cells, outputs=0):
    "A notebook from (cell_type, source) pairs; `outputs` stream outputs on the last code cell"
    cs = [{"cell_type": t, "metadata": {}, "source": s,
           **({"outputs": [], "execution_count": None} if t == "code" else {})} for t, s in cells]
    if outputs:
        code = [c for c in cs if c["cell_type"] == "code"][-1]
        code["outputs"] = [{"output_type": "stream", "name": "stdout", "text": "x"}] * outputs
    return {"cells": cs, "metadata": {}, "nbformat": 4, "nbformat_minor": 5}

_GOOD_TITLE = ("markdown", "# Title\n\n> What it is.\n\n- skip_showdoc: true\n- skip_exec: true\n")
_GOOD_SECTION = ("markdown", "## One\n\nText.\n\n---")

def _lint(nb, name="01_X.ipynb", sub="Web/Front_End"):
    d = Path(tempfile.mkdtemp()); p = d/name
    p.write_text(json.dumps(nb), encoding="utf-8")
    return lint_notebook(p, sub)

test_eq(_lint(_nb(_GOOD_TITLE, _GOOD_SECTION, ("code", "x = 1"))), {})

_cases = [
    ("mermaid_plain",      _nb(_GOOD_TITLE, ("markdown", "## D\n\n```mermaid\ngraph LR\n```\n\n---")), {}),
    ("mermaid_braced",     _nb(_GOOD_TITLE, ("markdown", "## D\n\n```{mermaid}\ngraph LR\n```\n\n---")), {}),
    ("title_leading_rule", _nb(("markdown", "---\ntitle: x\n---\n# T\n\n> d\n\n- skip_showdoc: true")), {}),
    ("title_no_h1",        _nb(("markdown", "Title\n\n> d\n\n- skip_showdoc: true")), {}),
    ("title_code",         _nb(("code", "import os")), {}),
    ("title_no_desc",      _nb(("markdown", "# T\n\n- skip_showdoc: true")), {}),
    ("no_skip_showdoc",    _nb(("markdown", "# T\n\n> d")), {}),
    ("non_ascii",          _nb(_GOOD_TITLE, ("code", "print('caf" + chr(0xe9) + "')")), {}),
    ("naive_datetime",     _nb(_GOOD_TITLE, ("code", "t = datetime" + ".now()")), {}),
    ("multi_h2",           _nb(_GOOD_TITLE, ("markdown", "## A\n\n---\n\n## B\n\n---")), {}),
    ("no_trailing_rule",   _nb(_GOOD_TITLE, ("markdown", "## A\n\ntext")), {}),
    ("stale_outputs",      _nb(_GOOD_TITLE, ("code", "x"), outputs=2), {}),
    ("dl_matplotlib",      _nb(_GOOD_TITLE, ("code", "import matplotlib.pyplot as plt")),
                           {"sub": "DL/DL_methods"}),
    ("dltask_no_require",  _nb(_GOOD_TITLE, *[("markdown", f"## {s}\n\n---") for s in DLTASK_SECTIONS],
                               ("code", "cap = cv2.VideoCapture(0)")), {"sub": "DL/DL_tasks"}),
    ("dltask_sections",    _nb(_GOOD_TITLE, ("markdown", "## What is X?\n\n---"),
                               ("markdown", "## Setup\n\n---"), ("code", "x")), {"sub": "DL/DL_tasks"}),
    ("bad_filename",       _nb(_GOOD_TITLE), {"name": "01_A & B.ipynb"}),
    ("svg_in_index",       _nb(_GOOD_TITLE, ("markdown", "<svg width='1'></svg>")), {"name": "index.ipynb"}),
]
for key, nb, kw in _cases:
    assert key in _lint(nb, **kw), (key, _lint(nb, **kw))

# Counts are items, not just yes/no: two plain fences, two non-ASCII characters
test_eq(_lint(_nb(_GOOD_TITLE, ("markdown", "```mermaid\na\n```\n\n```mermaid\nb\n```")))["mermaid_plain"], 2)
test_eq(_lint(_nb(_GOOD_TITLE, ("code", "s = '" + chr(0x3b1) + chr(0x3b2) + "'")))["non_ascii"], 2)
# A library notebook keeps its docs, so it needs no skip_showdoc
assert "no_skip_showdoc" not in _lint(_nb(("markdown", "# T\n\n> d"), ("code", "#| default_exp core")))
# Inline SVG is fine outside index.ipynb (the site renders it), and a timezone-aware now() is fine
assert "svg_in_index" not in _lint(_nb(_GOOD_TITLE, ("markdown", "<svg width='1'></svg>")))
assert "naive_datetime" not in _lint(_nb(_GOOD_TITLE, ("code", "t = datetime.now(UTC)")))
# DL notebooks keep outputs on purpose; matplotlib is only a DL_methods rule
assert "stale_outputs" not in _lint(_nb(_GOOD_TITLE, ("code", "x"), outputs=2), sub="DL/DL_tasks")
assert "dl_matplotlib" not in _lint(_nb(_GOOD_TITLE, ("code", "import matplotlib")))
# Theory/ pages are not task notebooks
_d = Path(tempfile.mkdtemp())/"Theory"; _d.mkdir()
(_d/"00_Maths.ipynb").write_text(json.dumps(_nb(_GOOD_TITLE, ("markdown", "## Sets\n\n---"),
                                                ("code", "x"), ("code", "y"))))
assert "dltask_sections" not in lint_notebook(_d/"00_Maths.ipynb", "DL/DL_tasks")

# --- lint_one: index coverage, p_ and build dirs skipped, career never linted ------
_r7 = _fixture(specs=(("site", True, False),))
_n = _r7/"site"/"nbs"
(_n/"Sub").mkdir(); (_n/"_proc").mkdir()
for rel, nb in (("index.ipynb", _nb(_GOOD_TITLE, ("markdown", "[a](01_A.ipynb)"))),
                ("01_A.ipynb", _nb(_GOOD_TITLE)),
                ("Sub/02_B.ipynb", _nb(_GOOD_TITLE)),
                ("p_03_C.ipynb", _nb(("code", "x"))),          # private: never linted
                ("_proc/04_D.ipynb", _nb(("code", "x")))):     # build output: never linted
    (_n/rel).write_text(json.dumps(nb))
test_eq(unlinked_from_index(_n), ["Sub/02_B.ipynb"])
test_eq(lint_one(_r7/"site"), {"site/nbs/Sub/02_B.ipynb": {"index_unlinked": 1}})
test_eq(_skip_reason(lint_one, _r7/"site", sub=type("S", (), {"path": "Personal/bthek1_career"})()),
        "not published")

_tot = lint_totals({"a": {"mermaid_plain": 3}, "b": {"mermaid_plain": 1, "non_ascii": 5}})
test_eq(_tot["mermaid_plain"], (2, 4))
test_eq(_tot["large_file"], (0, 0))                   # every rule is present, even at 0
test_eq(len(_tot), len(NB_RULES))
assert len({k for k, _, _ in NB_RULES}) == len(NB_RULES)  # no duplicate keys

Discovery and reporting

format_summary

Report

Outcome

Skip

finish

walk

Running a console script per submodule

upload_one

script_path

Verbs

fleet_upload

copy_one

push_one

ahead_of_remote

fleet_copy

fleet_push

fleet_status

status_one

pull_root

init_new

dirty_submodules

update_one

fetch_all

fleet_sync

fleet_update

Scanning for personal data

scan_file

is_published

notebook_text

looks_secret

fleet_scan

scan_one

Linting the notebook standard

fleet_nblint

lint_totals

lint_one

unlinked_from_index

lint_notebook

The fleet entry point

fleet

run_verb

usage

parse_verb

Tests

Phase 2: push and copy

Phases 4 and 5: status, update, sync

The scan verb

The nblint verb

The `fleet` entry point

The `scan` verb

The `nblint` verb