Async job runner

[build-system]
requires = ["setuptools >= 77.0.3"]
build-backend = "setuptools.build_meta"

[project]
name = "jobrunner"
version = "1.0.0"
description = "Run jobs concurrently with asyncio: timeouts, retries with backoff, and cancellation."
requires-python = ">= 3.14"
dependencies = []

[project.scripts]
jobrunner = "jobrunner.cli:main"

[tool.setuptools]
packages = ["jobrunner"]

"""Run jobs concurrently with asyncio: timeouts, retries with backoff, and cancellation."""

__version__ = "1.0.0"

"""python -m jobrunner runs the command line."""

import sys

from .cli import main

sys.exit(main())

"""The domain: jobs, retry policies and results, generic in what a job returns."""

from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from enum import StrEnum
from typing import Any


class Status(StrEnum):
    """How a job ended."""

    SUCCEEDED = "succeeded"
    FAILED = "failed"
    TIMED_OUT = "timed out"
    CANCELLED = "cancelled"


@dataclass(frozen=True, slots=True)
class RetryPolicy:
    """How often to try a job, and how long to wait between tries.

    The wait before try n + 1 is backoff * factor ** (n - 1), capped at
    max_backoff: 0.1, 0.2, 0.4, ... with the defaults.
    """

    attempts: int = 1
    backoff: float = 0.1
    factor: float = 2.0
    max_backoff: float = 30.0
    retry_on: tuple[type[Exception], ...] = (Exception,)

    def __post_init__(self) -> None:
        # TODO: raise ValueError if attempts < 1, backoff or max_backoff < 0,
        # or factor < 1.
        pass

    def delay(self, attempt: int) -> float:
        """Seconds to wait after failed try number attempt (counting from 1)."""
        raise NotImplementedError

    def should_retry(self, error: BaseException) -> bool:
        """Whether this kind of error is worth another try (isinstance with retry_on)."""
        raise NotImplementedError


# TODO: make Job generic in what its action returns, with the type parameter
# syntax: class Job[T], and action: Callable[[], Awaitable[T]].
@dataclass(frozen=True, slots=True)
class Job:
    """A named unit of work. action is called afresh for every try."""

    name: str
    action: Callable[[], Awaitable[Any]]
    timeout: float | None = None
    retry: RetryPolicy = field(default_factory=RetryPolicy)


# TODO: make Result generic too: class Result[T], with value: T | None.
@dataclass(frozen=True, slots=True)
class Result:
    """What happened to one job."""

    name: str
    status: Status
    value: Any = None
    error: BaseException | None = None
    attempts: int = 0
    elapsed: float = 0.0

    @property
    def ok(self) -> bool:
        """True if the job succeeded."""
        raise NotImplementedError

"""Running jobs concurrently: a TaskGroup, a semaphore, timeouts, retries and cancellation."""

import asyncio
import logging
from collections.abc import Awaitable, Callable, Iterable

from .model import Job, Result, Status

logger = logging.getLogger(__name__)

type Sleep = Callable[[float], Awaitable[None]]


class _FailFast(Exception):
    """Raised inside the TaskGroup to cancel the other jobs after a failure."""


class Runner:
    """Runs jobs at most `concurrency` at a time and reports a Result for each."""

    def __init__(self, concurrency: int = 4, *, fail_fast: bool = False, sleep: Sleep = asyncio.sleep) -> None:
        # TODO: raise ValueError if concurrency is below 1.
        self.concurrency = concurrency
        self.fail_fast = fail_fast
        # Injected so tests can record the backoff instead of waiting for it.
        self._sleep = sleep

    # TODO: make run generic: async def run[T](self, jobs: Iterable[Job[T]]) -> list[Result[T]]
    async def run(self, jobs: Iterable[Job]) -> list[Result]:
        """Run every job; return their results in the order the jobs were given.

        With fail_fast, the first job that fails or times out cancels the
        others, which are reported as cancelled. If the caller cancels run()
        itself, every job is cancelled and the cancellation propagates.
        """
        # TODO:
        # - raise ValueError if two jobs have the same name;
        # - one asyncio.Semaphore(self.concurrency) for all jobs;
        # - create a task per job (self._run_job) in an asyncio.TaskGroup;
        # - catch _FailFast with except* around the TaskGroup;
        # - return the results in the order of the jobs.
        raise NotImplementedError

    async def _run_job(self, job: Job, slots: asyncio.Semaphore, stopping: asyncio.Event, results: dict[str, Result]) -> None:
        """Try job until it succeeds or runs out of attempts; store its Result in results."""
        # TODO:
        # - each try inside `async with slots:` and `async with asyncio.timeout(job.timeout):`;
        # - TimeoutError -> Status.TIMED_OUT, another Exception -> Status.FAILED;
        # - retry if attempts remain and job.retry.should_retry(error), after
        #   `await self._sleep(job.retry.delay(attempt))` OUTSIDE the slot;
        # - with fail_fast, a final failure sets stopping and raises _FailFast
        #   while still holding the slot; a job that finds stopping set when it
        #   gets its slot is cancelled without starting;
        # - on asyncio.CancelledError store Status.CANCELLED and re-raise it.
        raise NotImplementedError


def describe(error: BaseException | None) -> str:
    """A short description of an error, for logs and reports."""
    if error is None:
        return ""
    if isinstance(error, TimeoutError) and not str(error):
        return "timed out"
    return f"{type(error).__name__}: {error}" if str(error) else type(error).__name__

"""Jobs that run commands, and loading them from a TOML job file."""

import asyncio
import sys
import tomllib
from collections.abc import Awaitable, Callable, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from .model import Job, RetryPolicy


class ConfigError(Exception):
    """The job file is missing, is not valid TOML, or describes a job wrongly."""


@dataclass(frozen=True, slots=True)
class CommandOutput:
    """What a finished command printed, and its exit status."""

    argv: tuple[str, ...]
    returncode: int
    stdout: str
    stderr: str


class CommandFailed(Exception):
    """A command exited with a status other than 0."""

    def __init__(self, output: CommandOutput) -> None:
        last = output.stderr.strip().splitlines()[-1:] or [""]
        super().__init__(f"exit status {output.returncode}" + (f": {last[0]}" if last[0] else ""))
        self.output = output


async def run_command(argv: Sequence[str]) -> CommandOutput:
    """Run a command without a shell and wait for it; raise CommandFailed if it fails.

    A command whose first word is "python" runs with the interpreter running
    this program. If the waiting task is cancelled (by a timeout, say), the
    process is killed before the cancellation goes on.
    """
    # TODO: replace a first word "python" with sys.executable;
    # asyncio.create_subprocess_exec with stdout and stderr piped;
    # await process.communicate(), and on asyncio.CancelledError kill the
    # process, await process.wait() and re-raise; decode the output.
    raise NotImplementedError


def command_action(argv: Sequence[str]) -> Callable[[], Awaitable[CommandOutput]]:
    """A job action that runs argv afresh each time it is called."""
    frozen = tuple(argv)

    async def action() -> CommandOutput:
        return await run_command(frozen)

    return action


JOB_KEYS = {"name", "command", "timeout", "attempts", "backoff"}


def _number(table: dict[str, Any], key: str, where: str, *, minimum: float) -> float | None:
    value = table.get(key)
    if value is None:
        return None
    if isinstance(value, bool) or not isinstance(value, int | float) or value < minimum:
        raise ConfigError(f"{where}: {key} must be a number of at least {minimum}, not {value!r}")
    return float(value)


def load_jobs(path: Path) -> list[Job]:
    """Read the jobs of a TOML job file: an optional [defaults] table and [[job]] tables.

    Raise ConfigError for a file that cannot be read, invalid TOML, an unknown
    key, a job without a name or a list of strings as its command, a bad
    number, or a name used twice.
    """
    # TODO: tomllib.load needs the file opened in binary mode ("rb").
    # Each job: Job(name, command_action(command), timeout, RetryPolicy(...)),
    # with values from [defaults] where the job has none (backoff defaults to 0.5).
    raise NotImplementedError

"""Summing up a run: counts per status and a table of results."""

from collections import Counter
from collections.abc import Sequence
from dataclasses import dataclass

from .model import Result, Status
from .runner import describe


@dataclass(frozen=True, slots=True)
class Summary:
    """The shape of a finished run."""

    total: int
    counts: dict[Status, int]
    attempts: int
    slowest: str | None

    @property
    def ok(self) -> bool:
        """True if every job succeeded."""
        raise NotImplementedError

    def __str__(self) -> str:
        """For example "4 jobs: 2 succeeded, 1 failed, 1 timed out" (statuses in Status order, none that are 0)."""
        raise NotImplementedError


# TODO: generic in T, like Runner.run.
def summarise(results: Sequence[Result]) -> Summary:
    """Count the results per status, add up the attempts and find the slowest job."""
    raise NotImplementedError


def format_table(results: Sequence[Result]) -> str:
    """A header JOB STATUS TRIES SECONDS DETAIL, then one aligned row per job."""
    raise NotImplementedError

"""The command line: jobrunner [options] JOBFILE."""

import argparse
import asyncio
import cProfile
import logging
import sys
from pathlib import Path

from . import __version__
from .commands import ConfigError, load_jobs
from .report import format_table, summarise
from .runner import Runner

logger = logging.getLogger("jobrunner")

EXIT_OK = 0
EXIT_FAILED = 1  # at least one job did not succeed
EXIT_USAGE = 2  # a bad job file; argparse also exits with 2 on bad arguments


def positive_int(text: str) -> int:
    """An argparse type: a whole number of at least 1."""
    # TODO: raise argparse.ArgumentTypeError for anything else.
    raise NotImplementedError


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="jobrunner",
        description="Run the jobs of a TOML job file concurrently, with timeouts and retries.",
    )
    parser.add_argument("jobfile", type=Path, metavar="JOBFILE", help="a TOML file of [[job]] tables")
    # TODO: -c/--concurrency (default 4), --fail-fast, --profile FILE,
    # -v/--verbose (a count) and --version.
    return parser


def configure_logging(verbosity: int) -> None:
    """Warnings (retries, failures) and, with -v, progress go to standard error."""
    handler = logging.StreamHandler(sys.stderr)
    handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
    logger.handlers[:] = [handler]
    logger.setLevel(logging.WARNING if verbosity == 0 else logging.INFO if verbosity == 1 else logging.DEBUG)
    logger.propagate = False


def main(argv: list[str] | None = None) -> int:
    """Run the tool with argv (default: sys.argv[1:]); return the exit status."""
    # TODO: parse, configure logging, load the jobs (a ConfigError is logged
    # and returns EXIT_USAGE), asyncio.run(Runner(...).run(jobs)), inside
    # `with cProfile.Profile() as profiler:` when --profile is given, then
    # print the table and the summary and return EXIT_OK or EXIT_FAILED.
    raise NotImplementedError


if __name__ == "__main__":
    sys.exit(main())

# A job file for jobrunner. Try: python -m jobrunner samples/jobs.toml -v
# A command whose first word is "python" runs with the Python running jobrunner.

[defaults]
timeout = 5.0
backoff = 0.2

[[job]]
name = "compile"
command = ["python", "-c", "print('compiled 12 modules')"]

[[job]]
name = "unit-tests"
command = ["python", "-c", "import time; time.sleep(0.05); print('84 passed')"]

[[job]]
name = "lint"
command = ["python", "-c", "import sys; sys.exit('style: line 12 is too long')"]
attempts = 2

[[job]]
name = "slow-download"
command = ["python", "-c", "import time; time.sleep(30)"]
timeout = 0.3

import asyncio
import contextlib
import io
import json
import pstats
import sys
import tempfile
import time
from pathlib import Path

from jobrunner.cli import main
from jobrunner.commands import CommandFailed, ConfigError, command_action, load_jobs, run_command
from jobrunner.model import Job, Result, RetryPolicy, Status
from jobrunner.report import format_table, summarise
from jobrunner.runner import Runner


def job(name, value=None, *, delay=0.0, error=None, **options):
    """A job that waits delay seconds, then raises error or returns value (default: its name)."""
    async def action():
        await asyncio.sleep(delay)
        if error is not None:
            raise error
        return name if value is None else value
    return Job(name, action, **options)


def blocked(name, log, **options):
    """A job that waits for ever, and notes in log when it is cancelled."""
    async def action():
        try:
            await asyncio.Event().wait()
        finally:
            log.append(name)
    return Job(name, action, **options)


def run(jobs, **options):
    return asyncio.run(Runner(**options).run(jobs))


def cli(argv):
    """Call main(argv) and return (status, stdout, stderr)."""
    out, err = io.StringIO(), io.StringIO()
    with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
        status = main(argv)
    return status, out.getvalue(), err.getvalue()


def py(code):
    """A command line that runs this Python code, as a TOML array."""
    return json.dumps(["python", "-c", code])


def job_file(folder, text):
    path = Path(folder) / "jobs.toml"
    path.write_text(text, encoding="utf-8")
    return path


def test_retry_policy():
    """RetryPolicy doubles the wait after each try, up to max_backoff, and rejects bad settings"""
    policy = RetryPolicy(attempts=5, backoff=0.1, factor=2, max_backoff=0.5)
    got = [round(policy.delay(n), 6) for n in range(1, 5)]
    assert got == [0.1, 0.2, 0.4, 0.5], f"the delays after tries 1 to 4 are {got!r}, expected [0.1, 0.2, 0.4, 0.5]"
    assert RetryPolicy(retry_on=(ConnectionError,)).should_retry(ConnectionResetError()), "ConnectionResetError is a ConnectionError, so it should be retried"
    assert not RetryPolicy(retry_on=(ConnectionError,)).should_retry(ValueError()), "ValueError is not a ConnectionError, so it should not be retried"
    for bad in [{"attempts": 0}, {"backoff": -1}, {"factor": 0.5}]:
        try:
            RetryPolicy(**bad)
        except ValueError:
            continue
        raise AssertionError(f"RetryPolicy(**{bad!r}) should raise ValueError")


def test_generic_types():
    """Job, Result and Runner.run are generic, written with the Python 3.12 type parameter syntax"""
    for thing in [Job, Result, Runner.run]:
        params = getattr(thing, "__type_params__", ())
        assert len(params) == 1, f"{thing.__qualname__} has type parameters {params!r}, expected one, such as [T]"
    result = Result("x", Status.SUCCEEDED, 42, attempts=1)
    assert result.ok and not Result("y", Status.FAILED).ok, "Result.ok should be True only for a job that succeeded"


def test_runs_all_in_order():
    """run() runs every job and returns the results in the order the jobs were given"""
    results = run([job("slow", delay=0.03), job("fast", delay=0.0), job("middle", delay=0.01)])
    got = [(r.name, r.status, r.value, r.attempts) for r in results]
    want = [("slow", Status.SUCCEEDED, "slow", 1), ("fast", Status.SUCCEEDED, "fast", 1), ("middle", Status.SUCCEEDED, "middle", 1)]
    assert got == want, f"run gave {got!r}, expected {want!r}"


def test_concurrency_limit():
    """No more than concurrency jobs run at the same time, and they do overlap"""
    running = 0
    most = 0

    async def action():
        nonlocal running, most
        running += 1
        most = max(most, running)
        await asyncio.sleep(0.01)
        running -= 1

    started = time.perf_counter()
    run([Job(f"job{n}", action) for n in range(6)], concurrency=2)
    took = time.perf_counter() - started
    assert most == 2, f"at most {most} jobs ran at once, expected exactly 2"
    assert took < 0.5, f"six 0.01 s jobs, two at a time, took {took:.2f} s"


def test_rejects_bad_input():
    """Duplicate job names and a concurrency below 1 raise ValueError"""
    try:
        run([job("same"), job("same")])
    except ValueError:
        pass
    else:
        raise AssertionError("two jobs called 'same' should make run() raise ValueError")
    try:
        Runner(0)
    except ValueError:
        pass
    else:
        raise AssertionError("Runner(0) should raise ValueError")


def test_timeout():
    """A job that takes longer than its timeout is stopped and reported as timed out"""
    log = []
    results = run([blocked("stuck", log, timeout=0.02), job("quick")])
    stuck, quick = results
    assert stuck.status is Status.TIMED_OUT, f"the stuck job is {stuck.status!r}, expected Status.TIMED_OUT"
    assert isinstance(stuck.error, TimeoutError), f"its error is {stuck.error!r}, expected a TimeoutError"
    assert log == ["stuck"], "the stuck job should have been cancelled when its time was up"
    assert quick.status is Status.SUCCEEDED, f"the other job is {quick.status!r}, expected Status.SUCCEEDED"


def test_retries_with_backoff():
    """A failing job is retried with growing waits; other errors, or running out of tries, fail it"""
    waits = []

    async def fake_sleep(seconds):
        waits.append(seconds)

    calls = 0

    async def flaky():
        nonlocal calls
        calls += 1
        if calls < 3:
            raise ConnectionError(f"try {calls} failed")
        return "done"

    policy = RetryPolicy(attempts=4, backoff=0.1, retry_on=(ConnectionError,))
    [result] = asyncio.run(Runner(sleep=fake_sleep).run([Job("flaky", flaky, retry=policy)]))
    assert (result.status, result.value, result.attempts) == (Status.SUCCEEDED, "done", 3), f"flaky ended {result.status!r} with {result.value!r} after {result.attempts} attempts, expected succeeded, 'done', 3"
    assert [round(w, 6) for w in waits] == [0.1, 0.2], f"the runner waited {waits!r} between tries, expected [0.1, 0.2]"
    waits.clear()
    results = asyncio.run(Runner(sleep=fake_sleep).run([
        job("wrong", error=ValueError("bad data"), retry=policy),
        job("down", error=ConnectionError("no route"), retry=RetryPolicy(attempts=3, backoff=0)),
    ]))
    got = [(r.status, r.attempts) for r in results]
    assert got == [(Status.FAILED, 1), (Status.FAILED, 3)], f"got {got!r}; a ValueError should fail at once, a ConnectionError after all 3 attempts"
    assert isinstance(results[1].error, ConnectionError), f"the error kept is {results[1].error!r}, expected the last ConnectionError"


def test_backoff_frees_the_slot():
    """While a job waits to retry, it does not hold its slot, so another job can run"""
    other_ran = None

    async def sleep_until_other_ran(seconds):
        await other_ran.wait()

    calls = 0

    async def fails_once():
        nonlocal calls
        calls += 1
        if calls == 1:
            raise ConnectionError("first try fails")
        return "second try"

    async def other():
        other_ran.set()
        return "other"

    async def scenario():
        nonlocal other_ran
        other_ran = asyncio.Event()
        runner = Runner(1, sleep=sleep_until_other_ran)
        jobs = [Job("retrying", fails_once, retry=RetryPolicy(attempts=2)), Job("other", other)]
        async with asyncio.timeout(1):
            return await runner.run(jobs)

    try:
        results = asyncio.run(scenario())
    except TimeoutError:
        raise AssertionError("with concurrency 1 the run got stuck: the retrying job kept its slot while it waited") from None
    got = [(r.name, r.status) for r in results]
    assert got == [("retrying", Status.SUCCEEDED), ("other", Status.SUCCEEDED)], f"got {got!r}"


def test_fail_fast():
    """With fail_fast the first failure cancels the other jobs, which are reported as cancelled"""
    log = []
    results = run([blocked("waiting", log), job("broken", error=RuntimeError("boom")), blocked("queued", log)], concurrency=2, fail_fast=True)
    got = [(r.name, r.status) for r in results]
    want = [("waiting", Status.CANCELLED), ("broken", Status.FAILED), ("queued", Status.CANCELLED)]
    assert got == want, f"fail_fast gave {got!r}, expected {want!r}"
    assert log == ["waiting"], f"the jobs that were cancelled while running are {log!r}, expected ['waiting'] (queued never started)"
    assert results[2].attempts == 0, f"the queued job never started, but reports {results[2].attempts} attempts"


def test_external_cancellation():
    """Cancelling run() from outside cancels every job and is not swallowed"""
    log = []

    async def scenario():
        async with asyncio.timeout(0.02):
            await Runner(3).run([blocked(name, log) for name in ["a", "b", "c"]])

    try:
        asyncio.run(scenario())
    except TimeoutError:
        pass
    else:
        raise AssertionError("the outer timeout should end in TimeoutError; did run() swallow the cancellation?")
    assert sorted(log) == ["a", "b", "c"], f"the jobs cancelled are {sorted(log)!r}, expected ['a', 'b', 'c']"


def test_run_command():
    """run_command runs a command without a shell, and raises CommandFailed if it exits with an error"""
    output = asyncio.run(run_command(["python", "-c", "print('hello from', 'python')"]))
    assert (output.returncode, output.stdout.strip()) == (0, "hello from python"), f"got exit status {output.returncode} and output {output.stdout!r}"
    assert output.argv[0] == sys.executable, f"'python' should be replaced by sys.executable, but argv is {output.argv!r}"
    try:
        asyncio.run(run_command(["python", "-c", "import sys; sys.exit('disk full')"]))
    except CommandFailed as err:
        assert err.output.returncode == 1 and "disk full" in str(err), f"CommandFailed says {str(err)!r} with status {err.output.returncode}"
    else:
        raise AssertionError("a command that exits with status 1 should raise CommandFailed")


def test_command_timeout():
    """A command that runs past its timeout is killed and reported as timed out"""
    sleeper = Job("sleeper", command_action(["python", "-c", "import time; time.sleep(30)"]), timeout=0.2)
    started = time.perf_counter()
    [result] = run([sleeper])
    took = time.perf_counter() - started
    assert result.status is Status.TIMED_OUT, f"the sleeping command is {result.status!r}, expected Status.TIMED_OUT"
    assert took < 5, f"the run took {took:.1f} s; the command should have been killed after 0.2 s"


def test_load_jobs():
    """load_jobs reads [[job]] tables and fills in the [defaults]"""
    text = f"""
[defaults]
timeout = 10
attempts = 2

[[job]]
name = "build"
command = {py("print(1)")}

[[job]]
name = "deploy"
command = ["python", "deploy.py"]
timeout = 2.5
attempts = 1
backoff = 0
"""
    with tempfile.TemporaryDirectory() as folder:
        jobs = load_jobs(job_file(folder, text))
    got = [(j.name, j.timeout, j.retry.attempts) for j in jobs]
    assert got == [("build", 10.0, 2), ("deploy", 2.5, 1)], f"load_jobs gave (name, timeout, attempts) {got!r}"
    assert jobs[1].retry.backoff == 0, f"deploy has backoff {jobs[1].retry.backoff}, expected 0"


def test_load_jobs_rejects():
    """A missing file, bad TOML or a wrong job raises ConfigError that says what is wrong"""
    bad = {
        "not toml": "[[job]\nname = 1",
        "command": '[[job]]\nname = "a"\ncommand = "echo hi"\n',
        "twice": '[[job]]\nname = "a"\ncommand = ["x"]\n[[job]]\nname = "a"\ncommand = ["y"]\n',
        "retries": '[[job]]\nname = "a"\ncommand = ["x"]\nretries = 3\n',
        "timeout": '[[job]]\nname = "a"\ncommand = ["x"]\ntimeout = -1\n',
    }
    with tempfile.TemporaryDirectory() as folder:
        try:
            load_jobs(Path(folder) / "missing.toml")
        except ConfigError as err:
            assert "missing.toml" in str(err), f"the error {str(err)!r} should name the file"
        else:
            raise AssertionError("a missing job file should raise ConfigError")
        for word, text in bad.items():
            try:
                load_jobs(job_file(folder, text))
            except ConfigError as err:
                message = str(err)
                assert ("TOML" if word == "not toml" else word) in message, f"for {text!r} the error {message!r} should mention {word!r}"
            else:
                raise AssertionError(f"load_jobs accepted {text!r}; expected ConfigError")


def test_summary_and_table():
    """summarise counts the results per status, and format_table prints a row per job"""
    results = [
        Result("a", Status.SUCCEEDED, 1, attempts=1, elapsed=0.5),
        Result("b", Status.FAILED, error=RuntimeError("exit status 1"), attempts=3, elapsed=2.0),
        Result("c", Status.TIMED_OUT, error=TimeoutError(), attempts=1, elapsed=1.0),
        Result("d", Status.SUCCEEDED, 4, attempts=2, elapsed=0.25),
    ]
    summary = summarise(results)
    assert str(summary) == "4 jobs: 2 succeeded, 1 failed, 1 timed out", f"the summary reads {str(summary)!r}"
    assert (summary.attempts, summary.slowest, summary.ok) == (7, "b", False), f"attempts, slowest and ok are {(summary.attempts, summary.slowest, summary.ok)!r}, expected (7, 'b', False)"
    lines = format_table(results).splitlines()
    assert len(lines) == 5 and lines[0].split()[:2] == ["JOB", "STATUS"], f"the table is {lines!r}; expected a header and 4 rows"
    got = [line.split()[:3] for line in lines[1:]]
    want = [["a", "succeeded", "1"], ["b", "failed", "3"], ["c", "timed", "out"], ["d", "succeeded", "2"]]
    assert got == want, f"the rows start {got!r}, expected {want!r}"


def test_cli_runs_file():
    """jobrunner JOBFILE prints a table and a summary, logs retries, and returns 1 if a job failed"""
    text = f"""
[defaults]
backoff = 0

[[job]]
name = "hello"
command = {py("print('hello')")}

[[job]]
name = "broken"
command = {py("import sys; sys.exit(3)")}
attempts = 2

[[job]]
name = "count"
command = {py("print(sum(range(10)))")}
"""
    with tempfile.TemporaryDirectory() as folder:
        status, out, err = cli([str(job_file(folder, text)), "-c", "2"])
    lines = out.splitlines()
    assert status == 1, f"main returned {status}, expected 1 because a job failed"
    assert lines[-1] == "3 jobs: 2 succeeded, 1 failed", f"the last line is {lines[-1]!r}"
    rows = [line.split()[:3] for line in lines[1:-1]]
    assert rows == [["hello", "succeeded", "1"], ["broken", "failed", "2"], ["count", "succeeded", "1"]], f"the rows start {rows!r}"
    assert "retrying" in err and "broken" in err, f"standard error was {err!r}; expected a warning that broken is retried"


def test_cli_fail_fast():
    """All jobs succeeding returns 0; --fail-fast cancels the rest after a failure"""
    good = f'[[job]]\nname = "a"\ncommand = {py("pass")}\n[[job]]\nname = "b"\ncommand = {py("pass")}\n'
    mixed = f'[[job]]\nname = "sleeper"\ncommand = {py("import time; time.sleep(30)")}\n[[job]]\nname = "broken"\ncommand = {py("raise SystemExit(1)")}\n'
    with tempfile.TemporaryDirectory() as folder:
        status, out, _ = cli([str(job_file(folder, good))])
        assert (status, out.splitlines()[-1]) == (0, "2 jobs: 2 succeeded"), f"main returned {status} and ended with {out.splitlines()[-1:]!r}"
        started = time.perf_counter()
        status, out, _ = cli([str(job_file(folder, mixed)), "--fail-fast"])
        took = time.perf_counter() - started
    assert (status, out.splitlines()[-1]) == (1, "2 jobs: 1 failed, 1 cancelled"), f"with --fail-fast main returned {status} and ended with {out.splitlines()[-1:]!r}"
    assert took < 5, f"--fail-fast took {took:.1f} s; the sleeping command should have been cancelled"


def test_cli_bad_input():
    """A bad job file returns 2 with an error on standard error; bad arguments exit with 2"""
    with tempfile.TemporaryDirectory() as folder:
        status, out, err = cli([str(Path(folder) / "nope.toml")])
    assert (status, out) == (2, ""), f"for a missing job file main returned {status} and printed {out!r}, expected 2 and nothing"
    assert "nope.toml" in err, f"the error {err!r} should name nope.toml"
    for argv in [[], ["jobs.toml", "-c", "0"], ["jobs.toml", "--concurrency", "many"]]:
        try:
            with contextlib.redirect_stderr(io.StringIO()):
                main(argv)
        except SystemExit as exc:
            assert exc.code == 2, f"for {argv!r} the exit status was {exc.code!r}, expected 2"
        else:
            raise AssertionError(f"main({argv!r}) returned instead of exiting with status 2")


def test_cli_profile():
    """--profile FILE saves cProfile statistics that pstats can read"""
    text = f'[[job]]\nname = "a"\ncommand = {py("pass")}\n'
    with tempfile.TemporaryDirectory() as folder:
        target = Path(folder) / "run.prof"
        status, _, _ = cli([str(job_file(folder, text)), "--profile", str(target)])
        assert status == 0 and target.exists(), f"main returned {status}; the profile file exists: {target.exists()}"
        stats = pstats.Stats(str(target), stream=io.StringIO())
    functions = {name for _, _, name in stats.stats}
    assert "run" in functions and "_run_job" in functions, "the profile should include Runner.run and Runner._run_job"


def test_sample_jobs():
    """samples/jobs.toml ends with 2 succeeded, 1 failed and 1 timed out"""
    status, out, _ = cli(["samples/jobs.toml"])
    last = out.splitlines()[-1] if out else ""
    assert (status, last) == (1, "4 jobs: 2 succeeded, 1 failed, 1 timed out"), f"for samples/jobs.toml main returned {status} and ended with {last!r}"