Integrated repository, worktree, git stacker, and project manager
Lift shared infra out of the vendored stacker fork
Four package-level modules absorb helpers that were either duplicated
across stacker/ and the pm modules or trapped inside stacker:
- errors.CommandError unifies subprocess failure types; ProjectError
moves up from project/errors.py. stacker.git.GitError becomes an
alias so existing stacker call sites keep working; pool.worktree's
locally-redefined GitError is gone.
- subprocess_run.run + format_command_failure replace three copies
(stacker/git.py, stacker/gh.py, pool/worktree.py). gh.py simplifies
by using the auto-raising check=True path.
- sqlite_db.transaction/readonly take an optional schema; project/db
and stacker/db both delegate. StackerDB drops _init_db and its
connect() is now a generator CM that schema-inits on every open.
- output.style/use_color replace the ANSI logic inlined at the bottom
of stacker/service.py.
Jordan Isaacs
(Apr 23, 2026, 12:53 AM UTC)
04410103
ef82d046
+228
-188
+6
src/project_manager/errors.py
+6
src/project_manager/errors.py
+36
src/project_manager/output.py
+36
src/project_manager/output.py
···
1
+
"""Terminal output helpers — ANSI color gated on TTY + standard env vars."""
2
+
3
+
from __future__ import annotations
4
+
5
+
import os
6
+
import sys
7
+
8
+
_COLORS = {
9
+
"blue": "34",
10
+
"green": "32",
11
+
"yellow": "33",
12
+
"magenta": "35",
13
+
"cyan": "36",
14
+
"red": "31",
15
+
}
16
+
17
+
18
+
def use_color() -> bool:
19
+
if os.environ.get("NO_COLOR"):
20
+
return False
21
+
if os.environ.get("CLICOLOR_FORCE"):
22
+
return True
23
+
return sys.stdout.isatty()
24
+
25
+
26
+
def style(text: str, *, fg: str | None = None, bold: bool = False) -> str:
27
+
if not use_color():
28
+
return text
29
+
codes: list[str] = []
30
+
if bold:
31
+
codes.append("1")
32
+
if fg and fg in _COLORS:
33
+
codes.append(_COLORS[fg])
34
+
if not codes:
35
+
return text
36
+
return f"\033[{';'.join(codes)}m{text}\033[0m"
+2
-1
src/project_manager/pool/add.py
+2
-1
src/project_manager/pool/add.py
···
1
1
import subprocess
2
2
import uuid as uuid_mod
3
3
4
+
from project_manager.errors import CommandError
4
5
from project_manager.paths import Paths
5
6
from project_manager.pool import worktree as wt
6
7
from project_manager.pool.slot import Slot
···
16
17
"""
17
18
main_repo = paths.repo(repo)
18
19
if not (main_repo / ".git").exists():
19
-
raise wt.GitError(f"no git repo at {main_repo}")
20
+
raise CommandError(f"no git repo at {main_repo}")
20
21
21
22
pool_dir = paths.pool(repo)
22
23
pool_dir.mkdir(parents=True, exist_ok=True)
+2
-2
src/project_manager/pool/cli.py
+2
-2
src/project_manager/pool/cli.py
···
2
2
import sys
3
3
4
4
from project_manager import config
5
+
from project_manager.errors import CommandError
5
6
from project_manager.pool import add as add_mod
6
7
from project_manager.pool import ls as ls_mod
7
-
from project_manager.pool import worktree as wt
8
8
from project_manager.stacker import cli as stacker_cli
9
9
10
10
···
12
12
paths = config.load()
13
13
try:
14
14
slot = add_mod.add(paths, args.repo)
15
-
except wt.GitError as e:
15
+
except CommandError as e:
16
16
print(f"pm: {e}", file=sys.stderr)
17
17
return 2
18
18
print(f"{slot.repo}\t{slot.uuid}\t{slot.path}")
+5
-16
src/project_manager/pool/worktree.py
+5
-16
src/project_manager/pool/worktree.py
···
3
3
import subprocess
4
4
from pathlib import Path
5
5
6
-
7
-
class GitError(Exception):
8
-
pass
6
+
from project_manager.errors import CommandError
7
+
from project_manager.subprocess_run import run
9
8
10
9
11
10
def _git(repo: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
12
-
result = subprocess.run(
13
-
["git", "-C", str(repo), *args],
14
-
capture_output=True,
15
-
text=True,
16
-
check=False,
17
-
)
18
-
if check and result.returncode != 0:
19
-
raise GitError(
20
-
f"git -C {repo} {' '.join(args)} failed ({result.returncode}):\n{result.stderr}"
21
-
)
22
-
return result
11
+
return run(["git", "-C", str(repo), *args], check=check)
23
12
24
13
25
14
def default_branch(repo: Path) -> str:
···
35
24
result = _git(repo, "branch", "--show-current", check=False)
36
25
if result.returncode == 0 and result.stdout.strip():
37
26
return result.stdout.strip()
38
-
raise GitError(f"could not determine default branch for {repo}")
27
+
raise CommandError(f"could not determine default branch for {repo}")
39
28
40
29
41
30
def worktree_add_detached(repo: Path, slot: Path, branch: str) -> None:
···
74
63
main_sub = main_repo / path
75
64
worktree_sub = worktree / path
76
65
if not (main_sub / ".git").exists():
77
-
raise GitError(
66
+
raise CommandError(
78
67
f"submodule '{path}' is not initialized in main repo {main_repo}; "
79
68
f"run `git -C {main_repo} submodule update --init --recursive` first"
80
69
)
+1
-1
src/project_manager/project/attach.py
+1
-1
src/project_manager/project/attach.py
···
2
2
import sqlite3
3
3
from pathlib import Path
4
4
5
+
from project_manager.errors import ProjectError
5
6
from project_manager.paths import Paths
6
7
from project_manager.pool import slot as slot_mod
7
8
from project_manager.pool.slot import PoolExhaustedError, Slot, SlotBusyError
8
9
from project_manager.project import db
9
-
from project_manager.project.errors import ProjectError
10
10
11
11
12
12
def _resolve_repos(
+1
-1
src/project_manager/project/cli.py
+1
-1
src/project_manager/project/cli.py
···
3
3
4
4
from project_manager import check as check_mod
5
5
from project_manager import config
6
+
from project_manager.errors import ProjectError
6
7
from project_manager.project import attach as attach_mod
7
8
from project_manager.project import current
8
9
from project_manager.project import delete as delete_mod
···
10
11
from project_manager.project import ls as ls_mod
11
12
from project_manager.project import new as new_mod
12
13
from project_manager.project import status as status_mod
13
-
from project_manager.project.errors import ProjectError
14
14
15
15
16
16
def _parse_repos(value: str) -> list[str]:
+1
-1
src/project_manager/project/current.py
+1
-1
src/project_manager/project/current.py
+5
-22
src/project_manager/project/db.py
+5
-22
src/project_manager/project/db.py
···
10
10
from contextlib import contextmanager
11
11
from pathlib import Path
12
12
13
+
from project_manager import sqlite_db
14
+
13
15
_SCHEMA = """
14
16
CREATE TABLE IF NOT EXISTS repos (
15
17
name TEXT PRIMARY KEY,
···
20
22
21
23
@contextmanager
22
24
def transaction(db_path: Path) -> Iterator[sqlite3.Connection]:
23
-
"""Open the project db, begin a transaction, and yield the connection.
24
-
25
-
Commits on clean exit, rolls back on exception, closes in all cases.
26
-
The schema-init statement runs and is committed before the transaction begins
27
-
so a rollback doesn't drop the table on first use.
28
-
"""
29
-
db_path.parent.mkdir(parents=True, exist_ok=True)
30
-
conn = sqlite3.connect(db_path)
31
-
try:
32
-
conn.execute(_SCHEMA)
33
-
conn.commit()
34
-
with conn:
35
-
yield conn
36
-
finally:
37
-
conn.close()
25
+
with sqlite_db.transaction(db_path, schema=_SCHEMA) as conn:
26
+
yield conn
38
27
39
28
40
29
@contextmanager
41
30
def readonly(db_path: Path) -> Iterator[sqlite3.Connection]:
42
-
"""Open the project db for reads only. No transaction; closes on exit."""
43
-
conn = sqlite3.connect(db_path)
44
-
try:
45
-
conn.execute(_SCHEMA)
46
-
conn.commit()
31
+
with sqlite_db.readonly(db_path, schema=_SCHEMA) as conn:
47
32
yield conn
48
-
finally:
49
-
conn.close()
50
33
51
34
52
35
def add_repo(conn: sqlite3.Connection, name: str, slot_uuid: str) -> None:
+1
-1
src/project_manager/project/delete.py
+1
-1
src/project_manager/project/delete.py
···
1
1
import contextlib
2
2
from pathlib import Path
3
3
4
+
from project_manager.errors import ProjectError
4
5
from project_manager.paths import Paths
5
6
from project_manager.project import db
6
7
from project_manager.project import detach as detach_mod
7
-
from project_manager.project.errors import ProjectError
8
8
9
9
_DB_FILENAME = ".pm.db"
10
10
_ALLOWED_EXTRAS = {_DB_FILENAME, "README.md"}
+1
-1
src/project_manager/project/detach.py
+1
-1
src/project_manager/project/detach.py
···
1
1
import contextlib
2
2
3
+
from project_manager.errors import ProjectError
3
4
from project_manager.paths import Paths
4
5
from project_manager.pool import slot as slot_mod
5
6
from project_manager.pool.slot import Slot
6
7
from project_manager.project import db
7
-
from project_manager.project.errors import ProjectError
8
8
9
9
10
10
def detach(paths: Paths, project: str, repos: list[str] | None) -> list[Slot]:
+1
-1
src/project_manager/project/discovery.py
+1
-1
src/project_manager/project/discovery.py
···
2
2
3
3
from pathlib import Path
4
4
5
+
from project_manager.errors import ProjectError
5
6
from project_manager.paths import Paths
6
7
from project_manager.project import db
7
-
from project_manager.project.errors import ProjectError
8
8
9
9
10
10
def list_project_dbs(paths: Paths) -> list[tuple[str, Path]]:
-2
src/project_manager/project/errors.py
-2
src/project_manager/project/errors.py
+1
-1
src/project_manager/project/new.py
+1
-1
src/project_manager/project/new.py
···
2
2
import sqlite3
3
3
from pathlib import Path
4
4
5
+
from project_manager.errors import ProjectError
5
6
from project_manager.paths import Paths
6
7
from project_manager.pool import slot as slot_mod
7
8
from project_manager.pool.slot import PoolExhaustedError, Slot, SlotBusyError
8
9
from project_manager.project import db
9
-
from project_manager.project.errors import ProjectError
10
10
11
11
_README_TEMPLATE = """\
12
12
# {project}
+56
src/project_manager/sqlite_db.py
+56
src/project_manager/sqlite_db.py
···
1
+
"""Package-level SQLite helpers.
2
+
3
+
Two context managers: `transaction()` wraps a unit of work with commit/rollback,
4
+
`readonly()` is a connection-only CM for reads. Both take an optional `schema`
5
+
string that is executed (idempotent `CREATE TABLE IF NOT EXISTS ...`) and
6
+
committed before the caller's block runs, so a rollback can't drop the tables
7
+
on first use.
8
+
"""
9
+
10
+
from __future__ import annotations
11
+
12
+
import sqlite3
13
+
from collections.abc import Iterator
14
+
from contextlib import contextmanager
15
+
from pathlib import Path
16
+
17
+
18
+
@contextmanager
19
+
def transaction(
20
+
db_path: Path,
21
+
*,
22
+
schema: str = "",
23
+
row_factory: type | None = None,
24
+
) -> Iterator[sqlite3.Connection]:
25
+
db_path.parent.mkdir(parents=True, exist_ok=True)
26
+
conn = sqlite3.connect(db_path)
27
+
if row_factory is not None:
28
+
conn.row_factory = row_factory
29
+
try:
30
+
if schema:
31
+
conn.executescript(schema)
32
+
conn.commit()
33
+
with conn:
34
+
yield conn
35
+
finally:
36
+
conn.close()
37
+
38
+
39
+
@contextmanager
40
+
def readonly(
41
+
db_path: Path,
42
+
*,
43
+
schema: str = "",
44
+
row_factory: type | None = None,
45
+
) -> Iterator[sqlite3.Connection]:
46
+
db_path.parent.mkdir(parents=True, exist_ok=True)
47
+
conn = sqlite3.connect(db_path)
48
+
if row_factory is not None:
49
+
conn.row_factory = row_factory
50
+
try:
51
+
if schema:
52
+
conn.executescript(schema)
53
+
conn.commit()
54
+
yield conn
55
+
finally:
56
+
conn.close()
+47
-49
src/project_manager/stacker/db.py
+47
-49
src/project_manager/stacker/db.py
···
2
2
3
3
import json
4
4
import sqlite3
5
+
from collections.abc import Iterator
6
+
from contextlib import contextmanager
5
7
from pathlib import Path
6
8
9
+
from project_manager import sqlite_db
10
+
7
11
from .models import OperationState, RepoPRConfig, TrackedBranch
8
12
13
+
_SCHEMA = """
14
+
CREATE TABLE IF NOT EXISTS tracked_branches (
15
+
repo_name TEXT NOT NULL,
16
+
branch TEXT NOT NULL,
17
+
parent_repo_name TEXT NOT NULL,
18
+
parent_branch TEXT NOT NULL,
19
+
managed_base_commit TEXT NOT NULL,
20
+
last_synced_parent_commit TEXT,
21
+
last_clean_head TEXT,
22
+
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
23
+
PRIMARY KEY (repo_name, branch)
24
+
);
9
25
10
-
class StackerDB:
11
-
def __init__(self, db_path: Path) -> None:
12
-
self.db_path = db_path
13
-
self.db_path.parent.mkdir(parents=True, exist_ok=True)
14
-
self._init_db()
26
+
CREATE TABLE IF NOT EXISTS operations (
27
+
repo_name TEXT PRIMARY KEY,
28
+
op_type TEXT NOT NULL,
29
+
status TEXT NOT NULL,
30
+
branch TEXT,
31
+
parent_branch TEXT,
32
+
root_branch TEXT,
33
+
queue_json TEXT,
34
+
current_index INTEGER NOT NULL DEFAULT 0,
35
+
start_head TEXT,
36
+
target_parent_head TEXT,
37
+
commit_list_json TEXT,
38
+
next_commit_index INTEGER NOT NULL DEFAULT 0,
39
+
error_message TEXT,
40
+
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
41
+
);
15
42
16
-
def connect(self) -> sqlite3.Connection:
17
-
conn = sqlite3.connect(self.db_path)
18
-
conn.row_factory = sqlite3.Row
19
-
return conn
43
+
CREATE TABLE IF NOT EXISTS repo_pr_config (
44
+
repo_name TEXT PRIMARY KEY,
45
+
mode TEXT NOT NULL,
46
+
trunk_branch TEXT NOT NULL,
47
+
main_repo TEXT,
48
+
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
49
+
);
50
+
"""
20
51
21
-
def _init_db(self) -> None:
22
-
with self.connect() as conn:
23
-
conn.executescript(
24
-
"""
25
-
CREATE TABLE IF NOT EXISTS tracked_branches (
26
-
repo_name TEXT NOT NULL,
27
-
branch TEXT NOT NULL,
28
-
parent_repo_name TEXT NOT NULL,
29
-
parent_branch TEXT NOT NULL,
30
-
managed_base_commit TEXT NOT NULL,
31
-
last_synced_parent_commit TEXT,
32
-
last_clean_head TEXT,
33
-
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
34
-
PRIMARY KEY (repo_name, branch)
35
-
);
36
52
37
-
CREATE TABLE IF NOT EXISTS operations (
38
-
repo_name TEXT PRIMARY KEY,
39
-
op_type TEXT NOT NULL,
40
-
status TEXT NOT NULL,
41
-
branch TEXT,
42
-
parent_branch TEXT,
43
-
root_branch TEXT,
44
-
queue_json TEXT,
45
-
current_index INTEGER NOT NULL DEFAULT 0,
46
-
start_head TEXT,
47
-
target_parent_head TEXT,
48
-
commit_list_json TEXT,
49
-
next_commit_index INTEGER NOT NULL DEFAULT 0,
50
-
error_message TEXT,
51
-
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
52
-
);
53
+
class StackerDB:
54
+
def __init__(self, db_path: Path) -> None:
55
+
self.db_path = db_path
53
56
54
-
CREATE TABLE IF NOT EXISTS repo_pr_config (
55
-
repo_name TEXT PRIMARY KEY,
56
-
mode TEXT NOT NULL,
57
-
trunk_branch TEXT NOT NULL,
58
-
main_repo TEXT,
59
-
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
60
-
);
61
-
"""
62
-
)
57
+
@contextmanager
58
+
def connect(self) -> Iterator[sqlite3.Connection]:
59
+
with sqlite_db.transaction(self.db_path, schema=_SCHEMA, row_factory=sqlite3.Row) as conn:
60
+
yield conn
63
61
64
62
def upsert_branch(self, tracked: TrackedBranch) -> None:
65
63
with self.connect() as conn:
+7
-25
src/project_manager/stacker/gh.py
+7
-25
src/project_manager/stacker/gh.py
···
1
1
from __future__ import annotations
2
2
3
3
import json
4
-
import subprocess
5
4
from dataclasses import dataclass
6
5
from pathlib import Path
7
6
8
-
from . import git
7
+
from project_manager.errors import CommandError
8
+
from project_manager.subprocess_run import run
9
9
10
10
11
11
@dataclass(frozen=True)
···
50
50
cmd = ["gh", "repo", "view", "--json", "name,owner"]
51
51
if repo:
52
52
cmd.extend(["--repo", repo])
53
-
proc = git.run(cmd, cwd=cwd, check=False)
54
-
if proc.returncode != 0:
55
-
raise git.GitError(_format_failure(cmd, proc))
53
+
proc = run(cmd, cwd=cwd)
56
54
payload = json.loads(proc.stdout or "{}")
57
55
owner = payload.get("owner", {})
58
56
owner_login = owner.get("login", "") if isinstance(owner, dict) else str(owner)
59
57
name = payload.get("name", "")
60
58
if not owner_login or not name:
61
-
raise git.GitError("Could not determine GitHub repository owner/name.")
59
+
raise CommandError("Could not determine GitHub repository owner/name.")
62
60
return RepoInfo(name_with_owner=f"{owner_login}/{name}", owner=owner_login, name=name)
63
61
64
62
···
76
74
cmd.extend(["--head", head])
77
75
if search:
78
76
cmd.extend(["--search", search])
79
-
proc = git.run(cmd, check=False)
80
-
if proc.returncode != 0:
81
-
raise git.GitError(_format_failure(cmd, proc))
77
+
proc = run(cmd)
82
78
payload = json.loads(proc.stdout or "[]")
83
79
return [_to_pr(item) for item in payload]
84
80
···
94
90
]
95
91
if request.draft:
96
92
cmd.append("--draft")
97
-
proc = git.run(cmd, check=False)
98
-
if proc.returncode != 0:
99
-
raise git.GitError(_format_failure(cmd, proc))
100
-
return proc.stdout.strip()
93
+
return run(cmd).stdout.strip()
101
94
102
95
103
96
def edit_pr(request: EditPRRequest) -> None:
···
108
101
cmd.extend(["--body-file", str(request.body_file)])
109
102
if request.base is not None:
110
103
cmd.extend(["--base", request.base])
111
-
proc = git.run(cmd, check=False)
112
-
if proc.returncode != 0:
113
-
raise git.GitError(_format_failure(cmd, proc))
104
+
run(cmd)
114
105
115
106
116
107
def _to_pr(item: dict) -> PullRequest:
···
124
115
state=item.get("state", ""),
125
116
is_draft=bool(item.get("isDraft", False)),
126
117
)
127
-
128
-
129
-
def _format_failure(cmd: list[str], proc: subprocess.CompletedProcess[str]) -> str:
130
-
pieces = ["command failed:", " ".join(cmd)]
131
-
if proc.stdout.strip():
132
-
pieces.append(proc.stdout.strip())
133
-
if proc.stderr.strip():
134
-
pieces.append(proc.stderr.strip())
135
-
return "\n".join(pieces)
+9
-33
src/project_manager/stacker/git.py
+9
-33
src/project_manager/stacker/git.py
···
1
1
from __future__ import annotations
2
2
3
-
import os
4
3
import subprocess
5
4
from dataclasses import dataclass
6
5
from pathlib import Path
7
6
7
+
from project_manager.errors import CommandError
8
+
from project_manager.subprocess_run import run
8
9
9
-
class GitError(RuntimeError):
10
-
pass
10
+
# Re-export `run` so `git.run(...)` call sites keep working.
11
+
__all__ = ["run"]
12
+
13
+
# Historical alias: stacker uses `GitError` for both subprocess failures and
14
+
# semantic validation errors. Keep the name so existing `except git.GitError`
15
+
# and `raise git.GitError(...)` sites continue to work.
16
+
GitError = CommandError
11
17
12
18
13
19
@dataclass(frozen=True)
···
26
32
repo_root: Path
27
33
worktree_path: Path
28
34
branch: str
29
-
30
-
31
-
def run(
32
-
cmd: list[str],
33
-
*,
34
-
cwd: Path | None = None,
35
-
env: dict[str, str] | None = None,
36
-
check: bool = True,
37
-
) -> subprocess.CompletedProcess[str]:
38
-
full_env = os.environ.copy()
39
-
if env:
40
-
full_env.update(env)
41
-
proc = subprocess.run(
42
-
cmd,
43
-
cwd=str(cwd) if cwd is not None else None,
44
-
env=full_env,
45
-
text=True,
46
-
capture_output=True,
47
-
check=False,
48
-
)
49
-
if check and proc.returncode != 0:
50
-
raise GitError(_format_failure(cmd, proc))
51
-
return proc
52
35
53
36
54
37
def git(path: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
···
239
222
raise GitError(f"Could not determine a trunk branch for {repo_root}.")
240
223
241
224
242
-
def _format_failure(cmd: list[str], proc: subprocess.CompletedProcess[str]) -> str:
243
-
pieces = ["command failed:", " ".join(cmd)]
244
-
if proc.stdout.strip():
245
-
pieces.append(proc.stdout.strip())
246
-
if proc.stderr.strip():
247
-
pieces.append(proc.stderr.strip())
248
-
return "\n".join(pieces)
+2
-27
src/project_manager/stacker/service.py
+2
-27
src/project_manager/stacker/service.py
···
1
1
from __future__ import annotations
2
2
3
3
import contextlib
4
-
import os
5
4
import re
6
-
import sys
7
5
from collections.abc import Callable, Iterator
8
6
from dataclasses import dataclass
9
7
from pathlib import Path
10
8
from tempfile import NamedTemporaryFile
11
9
from urllib.parse import quote
12
10
11
+
from project_manager import output
13
12
from project_manager.paths import Paths
14
13
from project_manager.pool import slot as slot_mod
15
14
···
1376
1375
return "\n".join([*logs, final_message]) if logs else final_message
1377
1376
1378
1377
def _style(self, text: str, *, fg: str | None = None, bold: bool = False) -> str:
1379
-
if not self._use_color():
1380
-
return text
1381
-
codes: list[str] = []
1382
-
if bold:
1383
-
codes.append("1")
1384
-
colors = {
1385
-
"blue": "34",
1386
-
"green": "32",
1387
-
"yellow": "33",
1388
-
"magenta": "35",
1389
-
"cyan": "36",
1390
-
"red": "31",
1391
-
}
1392
-
if fg and fg in colors:
1393
-
codes.append(colors[fg])
1394
-
if not codes:
1395
-
return text
1396
-
return f"\033[{';'.join(codes)}m{text}\033[0m"
1397
-
1398
-
def _use_color(self) -> bool:
1399
-
if os.environ.get("NO_COLOR"):
1400
-
return False
1401
-
if os.environ.get("CLICOLOR_FORCE"):
1402
-
return True
1403
-
return sys.stdout.isatty()
1378
+
return output.style(text, fg=fg, bold=bold)
+39
src/project_manager/subprocess_run.py
+39
src/project_manager/subprocess_run.py
···
1
+
from __future__ import annotations
2
+
3
+
import os
4
+
import subprocess
5
+
from pathlib import Path
6
+
7
+
from .errors import CommandError
8
+
9
+
10
+
def run(
11
+
cmd: list[str],
12
+
*,
13
+
cwd: Path | None = None,
14
+
env: dict[str, str] | None = None,
15
+
check: bool = True,
16
+
) -> subprocess.CompletedProcess[str]:
17
+
full_env = os.environ.copy()
18
+
if env:
19
+
full_env.update(env)
20
+
proc = subprocess.run(
21
+
cmd,
22
+
cwd=str(cwd) if cwd is not None else None,
23
+
env=full_env,
24
+
text=True,
25
+
capture_output=True,
26
+
check=False,
27
+
)
28
+
if check and proc.returncode != 0:
29
+
raise CommandError(format_command_failure(cmd, proc))
30
+
return proc
31
+
32
+
33
+
def format_command_failure(cmd: list[str], proc: subprocess.CompletedProcess[str]) -> str:
34
+
pieces = ["command failed:", " ".join(cmd)]
35
+
if proc.stdout.strip():
36
+
pieces.append(proc.stdout.strip())
37
+
if proc.stderr.strip():
38
+
pieces.append(proc.stderr.strip())
39
+
return "\n".join(pieces)
+1
-1
tests/test_current.py
+1
-1
tests/test_current.py
···
1
1
import pytest
2
2
3
+
from project_manager.errors import ProjectError
3
4
from project_manager.paths import Paths
4
5
from project_manager.project import current
5
6
from project_manager.project import detach as detach_mod
6
7
from project_manager.project import new as new_mod
7
-
from project_manager.project.errors import ProjectError
8
8
9
9
10
10
def _mk_pool(paths: Paths, repo: str, uuids: list[str]) -> None:
+2
-1
tests/test_pool.py
+2
-1
tests/test_pool.py
···
3
3
4
4
import pytest
5
5
6
+
from project_manager.errors import CommandError
6
7
from project_manager.paths import Paths
7
8
from project_manager.pool import add as add_mod
8
9
from project_manager.pool import ls as ls_mod
···
94
95
95
96
96
97
def test_pool_add_errors_on_missing_repo(pm_env: Paths) -> None:
97
-
with pytest.raises(wt_mod.GitError):
98
+
with pytest.raises(CommandError):
98
99
add_mod.add(pm_env, "nonexistent")
+1
-1
tests/test_project.py
+1
-1
tests/test_project.py
···
3
3
4
4
import pytest
5
5
6
+
from project_manager.errors import ProjectError
6
7
from project_manager.paths import Paths
7
8
from project_manager.pool.slot import PoolExhaustedError
8
9
from project_manager.project import attach as attach_mod
···
11
12
from project_manager.project import detach as detach_mod
12
13
from project_manager.project import ls as ls_mod
13
14
from project_manager.project import new as new_mod
14
-
from project_manager.project.errors import ProjectError
15
15
16
16
17
17
def _mk_pool(paths: Paths, repo: str, uuids: list[str]) -> None:
+1
-1
tests/test_status.py
+1
-1
tests/test_status.py
···
1
1
import pytest
2
2
3
3
from project_manager import check as check_mod
4
+
from project_manager.errors import ProjectError
4
5
from project_manager.paths import Paths
5
6
from project_manager.project import detach as detach_mod
6
7
from project_manager.project import new as new_mod
7
8
from project_manager.project import status as status_mod
8
-
from project_manager.project.errors import ProjectError
9
9
10
10
11
11
def _mk_pool(paths: Paths, repo: str, uuids: list[str]) -> None: