Run a file's mutants at the same time in symlink overlays.

Each worker keeps a private copy of the file under test and links the rest of the project, so the original tree stays put. The default is one worker per core, and --max-workers can set a lower cap.

README.md

@@ -24,13 +24,20 @@ The first run of a tree executes every covered mutant. Start with the file you a
 ./mutator --no-coverage             # do not skip uncovered lines
 ./mutator --use-existing-coverage   # read reports already on disk
 ./mutator --test-command "pytest -q tests/test_demo.py"
+./mutator --max-workers 3 src/demo/core.clj
 ```
 
 Once a snapshot exists, the next run is differential. It reruns survivors and every site in a function whose text changed. Killed mutants in an unchanged function stay killed. `--since-last-run` is that same selection. `--mutate-all` ignores it.
 
 Exit `0` when every executed mutant was killed. Exit `2` when the baseline tests fail. Exit `3` when a mutant survives. A failed baseline does not rewrite the snapshot.
 
-Mutants run one at a time in the project tree. The original file is restored after each one. An interrupted run leaves a copy under `target/mutator-backup/`, and the next run puts it back before reading source.
+## Workers
+
+Mutants of one file run at the same time. The default is one worker per core. `--max-workers` sets the cap. The run uses the smaller of that cap, the number of cores, and the number of selected sites. Files are still taken one at a time.
+
+Each worker is a directory under `target/mutation-workers`. It is a symlink overlay of the project: the file being mutated is a private copy, and the rest of the tree is linked. The original files stay as they are. The overlay is removed when that file's mutants finish.
+
+The baseline still runs once, in the real tree, before any worker starts. The mutant command then runs inside the worker, in the same directory it would have used in the real tree, so a relative path such as `src` or `./pkg` refers to the overlay. A copy left under `target/mutator-backup/` by an interrupted older run is restored before a non-scan run.
 
 ## Snapshot
 

src/mutator/cli.py

@@ -55,6 +55,10 @@ Options:
                                 2 seconds.
   --mutation-warning <number>   Warn when a file selects more covered sites
                                 than this. Default: 50.
+  --max-workers <number>        Run at most this many mutants of one file at
+                                once. Default: one per core. The run uses the
+                                smaller of this limit, the cores, and the
+                                number of selected sites.
   --verbose                     Print each test command and each mutant.
 
 Arguments:
@@ -68,6 +72,12 @@ named test, tests, spec, specs, vendor, node_modules, and target are skipped.
 The default, once a snapshot exists, reruns survivors and sites in functions
 whose text changed. Killed mutants in unchanged functions are kept.
 
+Selected mutants of one file run at the same time, one worker per core unless
+--max-workers says otherwise. A worker is a symlink overlay under
+target/mutation-workers with its own copy of the mutated file. The project
+tree is left unchanged. A copy left under target/mutator-backup/ by an
+interrupted older run is restored before a non-scan run.
+
 Exit codes:
   0  every executed mutant was killed, or there was nothing to run
   1  usage error
@@ -104,6 +114,7 @@ class Options:
     test_command: str | None = None
     timeout_factor: float = 10.0
     mutation_warning: int = 50
+    max_workers: int | None = None
     changed: bool = False
     verbose: bool = False
 
@@ -114,6 +125,16 @@ def _take(args: list[str], index: int, option: str) -> str:
     return args[index + 1]
 
 
+def _workers(value: str) -> int:
+    try:
+        count = int(value)
+    except ValueError:
+        raise ValueError("--max-workers requires a positive integer") from None
+    if count < 1:
+        raise ValueError("--max-workers requires a positive integer")
+    return count
+
+
 def _lines(value: str) -> set[int]:
     found = set()
     for piece in value.split(","):
@@ -164,6 +185,10 @@ def parse_args(argv: list[str] | None = None) -> Options:
                 options.mutation_warning = int(_take(args, index, arg))
                 index += 2
                 continue
+            if arg == "--max-workers":
+                options.max_workers = _workers(_take(args, index, arg))
+                index += 2
+                continue
             if arg == "--lines":
                 options.lines = _lines(_take(args, index, arg))
                 index += 2
@@ -220,6 +245,12 @@ def parse_args(argv: list[str] | None = None) -> Options:
             message=f"--scan cannot be combined with --mutate-all\n\n{HELP}",
             exit_code=1,
         )
+    if options.scan and options.max_workers is not None:
+        return Options(
+            action="help",
+            message=f"--scan cannot be combined with --max-workers\n\n{HELP}",
+            exit_code=1,
+        )
     return options
 
 
@@ -424,6 +455,7 @@ def _mutate_files(options: Options, root: Path, files: list[Path]) -> int:
             timeout_factor=options.timeout_factor,
             mutation_warning=options.mutation_warning,
             baselines=baselines,
+            max_workers=options.max_workers,
         )
         outcome = _record(result, forms, written)
         if outcome == "baseline":

src/mutator/engine.py

@@ -11,7 +11,7 @@ from mutator.metrics import load_history, source_key, write_results
 from mutator.model import FormResult, RunResult, Site
 from mutator.report import format_scan
 from mutator.runner import CommandRunner, test_plan
-from mutator.sites import apply_site
+from mutator.workers import run_mutants
 
 
 def restore_backups(root: Path) -> list[Path]:
@@ -148,23 +148,6 @@ def _forms(
     return forms
 
 
-def _run_mutant(path: Path, root: Path, original: bytes, site: Site, run) -> str:
-    current = original[site.start : site.end].decode("utf-8")
-    if current != site.original:
-        print(f"Skipped {path}:{site.line} {site.description}; source bytes moved", file=sys.stderr)
-        return "survived"
-    backup = _backup(root, path, original)
-    mutated = apply_site(original.decode("utf-8"), site.start, site.end, site.mutant)
-    path.write_text(mutated, encoding="utf-8")
-    _drop_bytecode(path)
-    try:
-        return run()
-    finally:
-        path.write_bytes(original)
-        if path.read_bytes() == original and backup.is_file():
-            backup.unlink()
-
-
 def _decode(path: Path) -> tuple[bytes, str] | None:
     original = path.read_bytes()
     try:
@@ -215,26 +198,6 @@ def _baseline_failure(file_key: str, command: str, tail: str) -> RunResult:
     return RunResult(path=file_key, forms=[], written=[], baseline_failed=True, baseline_message=message)
 
 
-def _mutant_status(runner, command: str, cwd: Path, timeout: float) -> str:
-    result = runner.run(command, cwd, timeout)
-    if result.timed_out or result.code != 0:
-        return "killed"
-    return "survived"
-
-
-def _run_selected(path, root, original, selected, runner, command, cwd, timeout, file_key, outcomes) -> None:
-    for site in selected:
-        if runner.verbose:
-            print(f"{file_key}:{site.line} {site.description}", file=sys.stderr)
-        outcomes[site.mutation_id] = _run_mutant(
-            path,
-            root,
-            original,
-            site,
-            lambda: _mutant_status(runner, command, cwd, timeout),
-        )
-
-
 def _apply_selected(
     path,
     root,
@@ -247,6 +210,7 @@ def _apply_selected(
     file_key,
     baselines,
     outcomes,
+    max_workers,
 ) -> RunResult | None:
     if not selected:
         return None
@@ -257,7 +221,19 @@ def _apply_selected(
     if not ok:
         return _baseline_failure(file_key, command, tail)
     timeout = max(2.0, seconds * timeout_factor)
-    _run_selected(path, root, original, selected, runner, command, cwd, timeout, file_key, outcomes)
+    run_mutants(
+        root,
+        path,
+        original,
+        selected,
+        max_workers,
+        runner,
+        command,
+        cwd,
+        timeout,
+        file_key,
+        outcomes,
+    )
     return None
 
 
@@ -289,6 +265,7 @@ def mutate_file(
     timeout_factor: float,
     mutation_warning: int,
     baselines: dict[tuple[str, str], tuple[bool, float, str]],
+    max_workers: int | None = None,
 ) -> RunResult:
     """Mutate one file and write its namespaces into `.metrics/mutate`."""
 
@@ -323,6 +300,7 @@ def mutate_file(
         file_key,
         baselines,
         outcomes,
+        max_workers,
     )
     if failure is not None:
         return failure

src/mutator/runner.py

@@ -19,6 +19,46 @@ class CommandResult:
     output: str
 
 
+def _worker_home(cwd: Path) -> Path | None:
+    """The overlay root when cwd is inside target/mutation-workers."""
+
+    for candidate in [cwd, *cwd.parents]:
+        parent = candidate.parent
+        if not candidate.name.startswith("worker-"):
+            continue
+        if not parent.name.startswith("run-"):
+            continue
+        if parent.parent.name == "mutation-workers":
+            return candidate
+    return None
+
+
+def _source_entries(cwd: Path, worker: Path) -> list[str]:
+    found = []
+    for directory in (cwd / "src", cwd, worker / "src", worker):
+        text = str(directory)
+        if text in found:
+            continue
+        if directory.is_dir():
+            found.append(text)
+    return found
+
+
+def _prefer_worker_sources(cwd: Path, environment: dict) -> None:
+    """An editable install would otherwise import the unmutated tree."""
+
+    worker = _worker_home(cwd)
+    if worker is None:
+        return
+    entries = _source_entries(cwd, worker)
+    current = environment.get("PYTHONPATH")
+    if current:
+        entries.append(current)
+    if not entries:
+        return
+    environment["PYTHONPATH"] = os.pathsep.join(entries)
+
+
 class CommandRunner:
     def __init__(self, verbose: bool = False):
         self.verbose = verbose
@@ -29,6 +69,7 @@ class CommandRunner:
         started = time.monotonic()
         environment = os.environ.copy()
         environment["PYTHONDONTWRITEBYTECODE"] = "1"
+        _prefer_worker_sources(cwd, environment)
         process = subprocess.Popen(
             command,
             cwd=cwd,

src/mutator/workers.py

@@ -0,0 +1,317 @@
+"""Run mutants in symlink overlays, the way clj-mutate runs its workers.
+
+Each worker is a directory under target/mutation-workers. Project files are
+linked in. The file under test is a private copy, so workers can mutate it at
+the same time without touching the original tree. The overlay is removed when
+the file's mutants finish.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from queue import Empty, Queue
+from threading import Lock
+
+from mutator.model import Site
+from mutator.sites import apply_site
+
+# Build output and the worker tree itself must not be shared. A link to
+# ``target`` would point a worker at the directory that contains the worker.
+SKIP_LINK = frozenset(
+    {
+        ".git",
+        ".hg",
+        ".svn",
+        ".idea",
+        ".venv",
+        "venv",
+        "target",
+        "dist",
+        "build",
+        "out",
+        "coverage",
+        ".metrics",
+        "__pycache__",
+        ".pytest_cache",
+        ".mypy_cache",
+        ".clj-kondo",
+        ".gradle",
+    }
+)
+
+CONFIGS = (
+    "deps.edn",
+    "bb.edn",
+    "pom.xml",
+    "build.gradle",
+    "build.gradle.kts",
+    "settings.gradle",
+    "go.mod",
+    "go.sum",
+    "package.json",
+    "package-lock.json",
+    "pnpm-lock.yaml",
+    "yarn.lock",
+    "tsconfig.json",
+    "Cargo.toml",
+    "Cargo.lock",
+    "pyproject.toml",
+    "pytest.ini",
+    "setup.cfg",
+    "setup.py",
+    "conftest.py",
+    "requirements.txt",
+    "Pipfile",
+    "poetry.lock",
+)
+
+
+def worker_count(site_count: int, requested: int | None) -> int:
+    """The smaller of the sites, the cores, and an explicit --max-workers."""
+
+    processors = os.cpu_count() or 1
+    limit = processors if requested is None else requested
+    return max(1, min(site_count, processors, limit))
+
+
+def new_run_dir(root: Path) -> Path:
+    directory = root / "target" / "mutation-workers" / f"run-{uuid.uuid4()}"
+    directory.mkdir(parents=True)
+    return directory
+
+
+def symlink(link: Path, target: Path) -> None:
+    link.parent.mkdir(parents=True, exist_ok=True)
+    if link.is_symlink() or link.exists():
+        link.unlink()
+    link.symlink_to(target.resolve())
+
+
+def delete_tree(path: Path) -> None:
+    """Remove a worker tree without following a link into the real project."""
+
+    if path.is_symlink():
+        path.unlink()
+        return
+    if not path.exists():
+        return
+    if path.is_dir():
+        for child in list(path.iterdir()):
+            delete_tree(child)
+        path.rmdir()
+        return
+    path.unlink()
+
+
+def _copy_configs(worker: Path, root: Path, relative: str) -> None:
+    for name in CONFIGS:
+        source = root / name
+        if not source.is_file():
+            continue
+        if name == relative:
+            continue
+        (worker / name).write_bytes(source.read_bytes())
+
+
+def _link_children(worker_dir: Path, real_dir: Path, skip: set[str]) -> None:
+    if not real_dir.is_dir():
+        return
+    for child in real_dir.iterdir():
+        if child.name in skip or child.name in SKIP_LINK:
+            continue
+        destination = worker_dir / child.name
+        if destination.exists() or destination.is_symlink():
+            continue
+        symlink(destination, child)
+
+
+def _overlay(worker: Path, root: Path, relative: str, original: bytes) -> None:
+    """Real directories along the source path. Every sibling is a link."""
+
+    segments = tuple(relative.split("/"))
+    for index, _segment in enumerate(segments[:-1]):
+        rel_dir = Path(*segments[: index + 1])
+        _link_children(worker / rel_dir, root / rel_dir, {segments[index + 1]})
+    destination = worker.joinpath(*segments)
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    destination.write_bytes(original)
+
+
+def create_workers(
+    base: Path, root: Path, relative: str, original: bytes, count: int
+) -> list[Path]:
+    created = []
+    first = relative.split("/", 1)[0]
+    for index in range(count):
+        worker = base / f"worker-{index}"
+        worker.mkdir(parents=True, exist_ok=True)
+        _copy_configs(worker, root, relative)
+        _link_children(worker, root, {first})
+        _overlay(worker, root, relative, original)
+        created.append(worker)
+    return created
+
+
+def mapped_cwd(worker: Path, root: Path, cwd: Path) -> Path:
+    try:
+        relative = cwd.resolve().relative_to(root.resolve())
+    except ValueError:
+        return worker
+    return (worker / relative).resolve()
+
+
+def _execute(
+    directory: Path,
+    site: Site,
+    root: Path,
+    relative: str,
+    original: bytes,
+    runner,
+    command: str,
+    cwd: Path,
+    timeout: float,
+    file_key: str,
+) -> str:
+    current = original[site.start : site.end].decode("utf-8")
+    if current != site.original:
+        print(
+            f"Skipped {file_key}:{site.line} {site.description}; source bytes moved",
+            file=sys.stderr,
+        )
+        return "survived"
+    destination = directory / relative
+    mutated = apply_site(original.decode("utf-8"), site.start, site.end, site.mutant)
+    destination.write_text(mutated, encoding="utf-8")
+    try:
+        result = runner.run(command, mapped_cwd(directory, root, cwd), timeout)
+    finally:
+        destination.write_bytes(original)
+    if result.timed_out or result.code != 0:
+        return "killed"
+    return "survived"
+
+
+def _drain(
+    directory: Path,
+    pending: Queue,
+    lock: Lock,
+    outcomes: dict[str, str],
+    root: Path,
+    relative: str,
+    original: bytes,
+    runner,
+    command: str,
+    cwd: Path,
+    timeout: float,
+    file_key: str,
+) -> None:
+    while True:
+        try:
+            site = pending.get_nowait()
+        except Empty:
+            return
+        if runner.verbose:
+            with lock:
+                _report(file_key, site)
+        status = _execute(
+            directory,
+            site,
+            root,
+            relative,
+            original,
+            runner,
+            command,
+            cwd,
+            timeout,
+            file_key,
+        )
+        with lock:
+            outcomes[site.mutation_id] = status
+
+
+def _report(file_key: str, site: Site) -> None:
+    print(f"{file_key}:{site.line} {site.description}", file=sys.stderr)
+
+
+def _run_all(
+    directories: list[Path],
+    sites: list[Site],
+    outcomes: dict[str, str],
+    root: Path,
+    relative: str,
+    original: bytes,
+    runner,
+    command: str,
+    cwd: Path,
+    timeout: float,
+    file_key: str,
+) -> None:
+    pending: Queue = Queue()
+    for site in sites:
+        pending.put(site)
+    lock = Lock()
+    with ThreadPoolExecutor(max_workers=len(directories)) as pool:
+        futures = [
+            pool.submit(
+                _drain,
+                directory,
+                pending,
+                lock,
+                outcomes,
+                root,
+                relative,
+                original,
+                runner,
+                command,
+                cwd,
+                timeout,
+                file_key,
+            )
+            for directory in directories
+        ]
+        for future in futures:
+            future.result()
+
+
+def run_mutants(
+    root: Path,
+    source: Path,
+    original: bytes,
+    sites: list[Site],
+    max_workers: int | None,
+    runner,
+    command: str,
+    cwd: Path,
+    timeout: float,
+    file_key: str,
+    outcomes: dict[str, str],
+) -> None:
+    """Mutate ``sites`` in parallel overlays of ``root``. The source stays put."""
+
+    if not sites:
+        return
+    base = new_run_dir(root)
+    try:
+        relative = source.resolve().relative_to(root.resolve()).as_posix()
+        directories = create_workers(
+            base, root, relative, original, worker_count(len(sites), max_workers)
+        )
+        _run_all(
+            directories,
+            sites,
+            outcomes,
+            root,
+            relative,
+            original,
+            runner,
+            command,
+            cwd,
+            timeout,
+            file_key,
+        )
+    finally:
+        delete_tree(base)

tests/test_boundaries.py

@@ -7,6 +7,7 @@ alone when no input can tell the operators apart.
 
 import hashlib
 import sys
+import threading
 from pathlib import Path
 
 import pytest
@@ -119,9 +120,11 @@ class _Ok:
         self.verbose = False
         self.seconds = seconds
         self.timeouts = []
+        self._lock = threading.Lock()
 
     def run(self, command, cwd, timeout):
-        self.timeouts.append(timeout)
+        with self._lock:
+            self.timeouts.append(timeout)
         code = 0 if timeout is None else 1
         return CommandResult(code=code, timed_out=False, seconds=self.seconds, output="")
 
@@ -369,6 +372,10 @@ def test_options_reject_bad_values_and_keep_flag_defaults(monkeypatch):
     assert parse_args(["--reuse-coverage"]).use_existing_coverage is True
     assert parse_args(["--verbose"]).verbose is True
     assert parse_args(["--changed"]).changed is True
+    assert parse_args(["--max-workers", "2"]).max_workers == 2
+    assert parse_args(["--max-workers", "0"]).exit_code == 1
+    assert parse_args(["--max-workers", "nope"]).exit_code == 1
+    assert parse_args(["--scan", "--max-workers", "2"]).exit_code == 1
     assert parse_args(["--not-a-flag"]).exit_code == 1
     monkeypatch.setattr(sys, "argv", ["prog", "--verbose"])
     options = parse_args(None)
@@ -685,9 +692,11 @@ def test_mutate_all_reruns_a_killed_mutant_and_clears_its_backup(tmp_path):
         def __init__(self):
             self.verbose = False
             self.calls = 0
+            self._lock = threading.Lock()
 
         def run(self, command, cwd, timeout):
-            self.calls += 1
+            with self._lock:
+                self.calls += 1
             text = (cwd / "src" / "demo.py").read_text(encoding="utf-8")
             namespace: dict = {}
             exec(text, namespace)

tests/test_engine.py

@@ -1,16 +1,22 @@
+import os
+import threading
+
 from mutator.engine import mutate_file, restore_backups
 from mutator.edn import loads
 from mutator.metrics import snapshot_path
 from mutator.runner import CommandResult
+from mutator.workers import create_workers, delete_tree, new_run_dir, worker_count
 
 
 class FakeRunner:
     def __init__(self):
         self.verbose = False
         self.calls = 0
+        self._lock = threading.Lock()
 
     def run(self, command, cwd, timeout):
-        self.calls += 1
+        with self._lock:
+            self.calls += 1
         text = (cwd / "src" / "demo.py").read_text(encoding="utf-8")
         namespace: dict = {}
         exec(text, namespace)
@@ -133,6 +139,84 @@ def test_a_failed_baseline_does_not_rewrite_metrics(tmp_path):
     assert not snapshot_path(tmp_path, "demo").exists()
 
 
+def test_worker_count_follows_sites_cores_and_the_requested_cap():
+    cores = os.cpu_count() or 1
+    assert worker_count(10, None) == min(10, cores)
+    assert worker_count(10, 3) == min(3, cores)
+    assert worker_count(2, 8) == min(2, cores)
+
+
+def test_workers_keep_a_private_copy_and_link_the_rest(tmp_path):
+    source = tmp_path / "src" / "demo.py"
+    source.parent.mkdir()
+    source.write_text("def place(x):\n    return x > 0\n", encoding="utf-8")
+    sibling = tmp_path / "src" / "other.py"
+    sibling.write_text("kept = True\n", encoding="utf-8")
+    marker = tmp_path / "tests" / "keep.txt"
+    marker.parent.mkdir()
+    marker.write_text("keep", encoding="utf-8")
+    (tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n", encoding="utf-8")
+    original = source.read_bytes()
+
+    base = new_run_dir(tmp_path)
+    try:
+        workers = create_workers(base, tmp_path, "src/demo.py", original, 2)
+        assert len(workers) == 2
+        for worker in workers:
+            private = worker / "src" / "demo.py"
+            assert private.is_file()
+            assert not private.is_symlink()
+            assert private.read_bytes() == original
+            assert (worker / "src" / "other.py").is_symlink()
+            assert (worker / "tests").is_symlink()
+            assert (worker / "tests" / "keep.txt").read_text(encoding="utf-8") == "keep"
+            assert (worker / "pyproject.toml").is_file()
+            assert not (worker / "pyproject.toml").is_symlink()
+        (workers[0] / "src" / "demo.py").write_text("changed\n", encoding="utf-8")
+        assert workers[1].joinpath("src/demo.py").read_bytes() == original
+        assert source.read_bytes() == original
+    finally:
+        delete_tree(base)
+
+    assert source.read_bytes() == original
+    assert sibling.read_text(encoding="utf-8") == "kept = True\n"
+    assert marker.read_text(encoding="utf-8") == "keep"
+    assert not base.exists()
+
+
+def test_a_mutation_run_leaves_the_project_tree_unchanged(tmp_path):
+    path = tmp_path / "src" / "demo.py"
+    path.parent.mkdir()
+    path.write_text("def place(x):\n    return x > 0\n", encoding="utf-8")
+    original = path.read_text(encoding="utf-8")
+
+    class Record:
+        verbose = False
+
+        def run(self, command, cwd, timeout):
+            if timeout is not None:
+                assert "mutation-workers" in cwd.as_posix()
+                assert (cwd / "src" / "demo.py").read_text(encoding="utf-8") != original
+            return CommandResult(code=0 if timeout is None else 1, timed_out=False, seconds=0.01, output="")
+
+    mutate_file(
+        path,
+        tmp_path,
+        runner=Record(),
+        covered_lines=None,
+        ignore_coverage=True,
+        mutate_all=True,
+        lines=None,
+        test_command="fake",
+        timeout_factor=10,
+        mutation_warning=50,
+        baselines={},
+        max_workers=2,
+    )
+    assert path.read_text(encoding="utf-8") == original
+    assert [item for item in (tmp_path / "target").rglob("*") if item.is_file()] == []
+
+
 def test_restore_backups_puts_an_interrupted_mutant_back(tmp_path):
     source = tmp_path / "src" / "demo.py"
     source.parent.mkdir()

tests/test_runner.py

@@ -1,4 +1,6 @@
-from mutator.runner import _clojure_command, _python_command, nearest
+import sys
+
+from mutator.runner import CommandRunner, _clojure_command, _python_command, nearest
 
 
 def test_nearest_walks_up_to_the_marker(tmp_path):
@@ -66,3 +68,13 @@ def test_python_commands(tmp_path):
     conf.mkdir()
     (conf / "conftest.py").write_text("", encoding="utf-8")
     assert _python_command(conf).endswith("-m pytest")
+
+
+def test_a_worker_overlay_is_imported_ahead_of_the_environment(tmp_path):
+    worker = tmp_path / "target" / "mutation-workers" / "run-1" / "worker-0"
+    (worker / "src").mkdir(parents=True)
+    (worker / "src" / "demo.py").write_text("VALUE = 'worker'\n", encoding="utf-8")
+    command = f"{sys.executable} -c 'import demo; print(demo.VALUE)'"
+    result = CommandRunner().run(command, worker, 5)
+    assert result.code == 0
+    assert result.output.strip() == "worker"