diff --git a/README.md b/README.md index b01d9da..95c0733 100644 --- a/README.md +++ b/README.md @@ -200,8 +200,8 @@ def ensure_test_postgres(): os.environ[k] = v ``` -CLI: `pgdb testdb up|reset|run-sql|status|shell|clean`. `up`/`reset` accept -`--env` (default `local_test`) — see "Environment-tagged files" above. +CLI: `pgdb testdb up|reset|run-sql|status|shell|clean|list-orphaned`. `up`/`reset` +accept `--env` (default `local_test`) — see "Environment-tagged files" above. `up`/`reset` accept `--area`/`--exclude-area` and `--schema`/`--exclude-schema` (see "Area and schema filtering" above) to scope which `database/` files get @@ -210,6 +210,27 @@ applied — e.g. `pgdb testdb up --schema billing` for a test DB with only the project's schema to apply. `ensure_testdb`/`reset_testdb` take the same keyword arguments when called from Python (e.g. from a pytest fixture). +Every git worktree/branch of a project gets its own database, named after +`project_name` + branch (see `pgdevkit.testdb.naming.workspace_db_name`). +Removing a worktree (or deleting its directory without `git worktree +remove`) doesn't drop its database — `pgdb testdb list-orphaned` lists +this project's databases whose worktree no longer exists, and +`pgdb testdb clean --orphaned` drops them (as opposed to `--all`, which +drops every database of this project regardless of whether its worktree is +still live). The same is available from Python as +`pgdevkit.testdb.find_orphaned_dbs()` and +`pgdevkit.testdb.clean_testdb(orphaned=True)`. + +If your project's own test setup also creates a sibling database per +worktree (e.g. `_myservice` for a mock service used only by that +project's tests), add its literal suffix so orphan detection knows it +belongs to a live worktree too: + +```toml +[tool.pgdevkit] +extra_db_suffixes = ["_myservice"] +``` + Container connection defaults (`localhost:54322`, `postgres`/`testpwd`) can be overridden with `PGDEVKIT_TESTDB_HOST`, `PGDEVKIT_TESTDB_PORT`, `PGDEVKIT_TESTDB_USER`, `PGDEVKIT_TESTDB_PASSWORD`. Before touching the diff --git a/pgdevkit/cli.py b/pgdevkit/cli.py index c3c27fd..1f2d1e2 100644 --- a/pgdevkit/cli.py +++ b/pgdevkit/cli.py @@ -302,12 +302,33 @@ def testdb_shell() -> None: @testdb_app.command("clean") def testdb_clean( all: bool = typer.Option(False, "--all", help="Drop every database belonging to this project"), + orphaned: bool = typer.Option( + False, + "--orphaned", + help="Drop only databases whose git worktree no longer exists (see `testdb list-orphaned`)", + ), ) -> None: - """Drop this workspace's database (or every database of this project with --all).""" - testdb.clean_testdb(all=all) + """Drop this workspace's database (or every database of this project with --all, + or only its orphaned ones with --orphaned).""" + if all and orphaned: + err_console.print("[red]Error:[/red] pass at most one of --all, --orphaned") + raise typer.Exit(2) + testdb.clean_testdb(all=all, orphaned=orphaned) console.print("[green]Cleaned.[/green]") +@testdb_app.command("list-orphaned") +def testdb_list_orphaned() -> None: + """List this project's databases whose git worktree no longer exists, + without dropping them.""" + names = testdb.find_orphaned_dbs() + if not names: + console.print("No orphaned databases.") + return + for name in names: + console.print(name) + + @migrate_app.command("check") def migrate_check( migrations_dir: Path = typer.Argument(..., help="Directory of numbered .sql migration files"), diff --git a/pgdevkit/testdb/__init__.py b/pgdevkit/testdb/__init__.py index 9c56f9e..82ac13b 100644 --- a/pgdevkit/testdb/__init__.py +++ b/pgdevkit/testdb/__init__.py @@ -1,3 +1,21 @@ -from .api import clean_testdb, dsn_for, ensure_testdb, reset_testdb, run_sql, shell_argv, status +from .api import ( + clean_testdb, + dsn_for, + ensure_testdb, + find_orphaned_dbs, + reset_testdb, + run_sql, + shell_argv, + status, +) -__all__ = ["clean_testdb", "dsn_for", "ensure_testdb", "reset_testdb", "run_sql", "shell_argv", "status"] +__all__ = [ + "clean_testdb", + "dsn_for", + "ensure_testdb", + "find_orphaned_dbs", + "reset_testdb", + "run_sql", + "shell_argv", + "status", +] diff --git a/pgdevkit/testdb/api.py b/pgdevkit/testdb/api.py index b0c7d40..d99814b 100644 --- a/pgdevkit/testdb/api.py +++ b/pgdevkit/testdb/api.py @@ -9,7 +9,14 @@ from . import constants, query from .config import ProjectConfig, load_config from .container import ensure_container -from .naming import current_branch, slugify, workspace_db_name +from .naming import ( + current_branch, + escape_like_prefix, + expected_db_names, + live_worktree_branches, + slugify, + workspace_db_name, +) from .schema import apply_schema @@ -68,6 +75,16 @@ async def _drop_database(db_name: str) -> None: await con.execute(SQL("DROP DATABASE IF EXISTS {}").format(Identifier(db_name))) +async def _dbs_with_prefix(prefix: str) -> list[str]: + escaped_prefix = escape_like_prefix(prefix) + async with await psycopg.AsyncConnection.connect(_admin_dsn(), autocommit=True) as con: + result = await con.execute( + "SELECT datname FROM pg_database WHERE datname LIKE %(pattern)s ESCAPE '\\'", + {"pattern": f"{escaped_prefix}%"}, + ) + return [row[0] for row in await result.fetchall()] + + async def _apply( config: ProjectConfig, db_name: str, @@ -154,27 +171,45 @@ def reset_testdb( ) -def clean_testdb(project_root: Path | None = None, all: bool = False) -> None: +async def _find_orphaned_dbs(config: ProjectConfig) -> list[str]: + prefix = f"{slugify(config.name)}_" + actual = await _dbs_with_prefix(prefix) + expected = expected_db_names(config, live_worktree_branches(config.root)) + return sorted(set(actual) - expected) + + +def find_orphaned_dbs(project_root: Path | None = None) -> list[str]: + """Databases belonging to this project (matched by its name-slug prefix) + that don't belong to any currently live git worktree of this repo -- + i.e. their branch's worktree was removed (or never existed) without + also dropping its database.""" + config, _ = _resolve(project_root) + if config.engine == "mssql": + return _mssql_api().find_orphaned_dbs(config) + return asyncio.run(_find_orphaned_dbs(config)) + + +def clean_testdb(project_root: Path | None = None, all: bool = False, orphaned: bool = False) -> None: """Drop this workspace's database. With all=True, drop every database belonging to this project (matched by its name-slug prefix), across - every worktree/branch.""" + every worktree/branch. With orphaned=True, drop only those without a + currently live git worktree (see `find_orphaned_dbs`). At most one of + all/orphaned may be set.""" + if all and orphaned: + raise ValueError("clean_testdb: pass at most one of all=True, orphaned=True") + config, db_name = _resolve(project_root) if config.engine == "mssql": - _mssql_api().clean_testdb(config, db_name, all) + _mssql_api().clean_testdb(config, db_name, all, orphaned) return async def _run() -> None: - if not all: - await _drop_database(db_name) - return - prefix = f"{slugify(config.name)}_" - escaped_prefix = prefix.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") - async with await psycopg.AsyncConnection.connect(_admin_dsn(), autocommit=True) as con: - result = await con.execute( - "SELECT datname FROM pg_database WHERE datname LIKE %(pattern)s ESCAPE '\\'", - {"pattern": f"{escaped_prefix}%"}, - ) - names = [row[0] for row in await result.fetchall()] + if orphaned: + names = await _find_orphaned_dbs(config) + elif all: + names = await _dbs_with_prefix(f"{slugify(config.name)}_") + else: + names = [db_name] for name in names: await _drop_database(name) diff --git a/pgdevkit/testdb/config.py b/pgdevkit/testdb/config.py index c8e9349..4ec14f3 100644 --- a/pgdevkit/testdb/config.py +++ b/pgdevkit/testdb/config.py @@ -14,6 +14,7 @@ class ProjectConfig: database_dir: str = "database" env_prefix: str = "" extensions: tuple[str, ...] = () + extra_db_suffixes: tuple[str, ...] = () engine: str = "postgres" root: Path = field(default_factory=Path) @@ -48,6 +49,17 @@ def load_config(start: Path | None = None) -> ProjectConfig: f"[tool.pgdevkit].extensions in {pyproject} must be a list, got {type(extensions).__name__}" ) + # Lets a repo that layers an extra, literally-suffixed sibling database + # on top of its main workspace DB (e.g. a mock-service DB used only by + # that repo's own test setup) teach `find_orphaned_dbs`/`clean_testdb` + # about it, without pgdevkit needing to know why that suffix exists. + extra_db_suffixes = section.get("extra_db_suffixes", []) + if not isinstance(extra_db_suffixes, list): + raise TypeError( + f"[tool.pgdevkit].extra_db_suffixes in {pyproject} must be a list, " + f"got {type(extra_db_suffixes).__name__}" + ) + # PGDEVKIT_TESTDB_ENGINE lets CI/ad-hoc runs flip engines without # editing pyproject.toml; the toml value is the durable, per-project # default (a project's database/ tree is written in one dialect, so @@ -59,6 +71,7 @@ def load_config(start: Path | None = None) -> ProjectConfig: database_dir=section.get("database_dir", "database"), env_prefix=section.get("env_prefix", ""), extensions=tuple(extensions), + extra_db_suffixes=tuple(extra_db_suffixes), engine=engine, root=root, ) diff --git a/pgdevkit/testdb/mssql/api.py b/pgdevkit/testdb/mssql/api.py index fbc0a96..6b7438d 100644 --- a/pgdevkit/testdb/mssql/api.py +++ b/pgdevkit/testdb/mssql/api.py @@ -168,26 +168,46 @@ async def _run() -> None: return _env_for(config, db_name) -def clean_testdb(config: ProjectConfig, db_name: str, all: bool) -> None: - from ..naming import slugify +async def _dbs_with_prefix(prefix: str) -> list[str]: + from ..naming import escape_like_prefix - async def _run() -> None: - if not all: - await _drop_database(db_name) - return - prefix = f"{slugify(config.name)}_" - escaped_prefix = prefix.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + escaped_prefix = escape_like_prefix(prefix) - def _list_names() -> list[str]: - conn = mssql_python.connect(_admin_dsn(), autocommit=True) - try: - cur = conn.cursor() - cur.execute("SELECT name FROM sys.databases WHERE name LIKE ? ESCAPE '\\'", [f"{escaped_prefix}%"]) - return [row[0] for row in cur.fetchall()] - finally: - conn.close() + def _list_names() -> list[str]: + conn = mssql_python.connect(_admin_dsn(), autocommit=True) + try: + cur = conn.cursor() + cur.execute("SELECT name FROM sys.databases WHERE name LIKE ? ESCAPE '\\'", [f"{escaped_prefix}%"]) + return [row[0] for row in cur.fetchall()] + finally: + conn.close() - names = await asyncio.to_thread(_list_names) + return await asyncio.to_thread(_list_names) + + +async def _find_orphaned_dbs(config: ProjectConfig) -> list[str]: + from ..naming import expected_db_names, live_worktree_branches, slugify + + prefix = f"{slugify(config.name)}_" + actual = await _dbs_with_prefix(prefix) + expected = expected_db_names(config, live_worktree_branches(config.root)) + return sorted(set(actual) - expected) + + +def find_orphaned_dbs(config: ProjectConfig) -> list[str]: + return asyncio.run(_find_orphaned_dbs(config)) + + +def clean_testdb(config: ProjectConfig, db_name: str, all: bool, orphaned: bool = False) -> None: + from ..naming import slugify + + async def _run() -> None: + if orphaned: + names = await _find_orphaned_dbs(config) + elif all: + names = await _dbs_with_prefix(f"{slugify(config.name)}_") + else: + names = [db_name] for name in names: await _drop_database(name) diff --git a/pgdevkit/testdb/naming.py b/pgdevkit/testdb/naming.py index 7a21240..a954b8c 100644 --- a/pgdevkit/testdb/naming.py +++ b/pgdevkit/testdb/naming.py @@ -4,6 +4,10 @@ import re import subprocess from pathlib import Path +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from .config import ProjectConfig _INVALID_CHARS = re.compile(r"[^a-z0-9_]+") _MAX_SLUG_LEN = 30 @@ -39,3 +43,70 @@ def workspace_db_name(project_name: str, branch: str) -> str: both inputs are already at the per-component truncation limit.""" joined = f"{slugify(project_name)}_{slugify(branch)}" return slugify(joined) + + +def live_worktree_branches(repo: Path) -> list[str]: + """Branches checked out in every currently-live worktree of the repo + containing `repo` -- "live" meaning its filesystem path still exists and + it isn't a bare checkout. `git worktree list` reports every worktree of + a repo regardless of which one it's run from, so this works whether + `repo` is the main checkout or a linked worktree. + + A detached-HEAD worktree (no `branch ...` porcelain line) is reported as + branch "HEAD" -- matching what `current_branch()` (and so + `workspace_db_name()`) computes for that same worktree via `git + rev-parse --abbrev-ref HEAD`. Without this, a detached-HEAD worktree + (e.g. a CI checkout, which defaults to one) would never appear "live" + here, and its database would look orphaned and get dropped out from + under it.""" + result = subprocess.run( + ["git", "worktree", "list", "--porcelain"], + cwd=repo, + capture_output=True, + text=True, + check=True, + ) + + branches: list[str] = [] + path: Path | None = None + branch: str | None = None + bare = False + + def _flush() -> None: + if path is not None and not bare and branch is not None and path.exists(): + branches.append(branch) + + for line in result.stdout.splitlines(): + if not line: + _flush() + path, branch, bare = None, None, False + elif line.startswith("worktree "): + path = Path(line[len("worktree ") :]) + elif line.startswith("branch "): + branch = line[len("branch ") :].removeprefix("refs/heads/") + elif line == "detached": + branch = "HEAD" + elif line == "bare": + bare = True + _flush() + + return branches + + +def escape_like_prefix(prefix: str) -> str: + """Escape a literal string for use as a `LIKE ... ESCAPE '\\'` prefix + pattern (with a trailing `%` the caller adds), so a project name/branch + containing `%` or `_` can't widen the match to an unrelated database.""" + return prefix.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + + +def expected_db_names(config: "ProjectConfig", branches: list[str]) -> set[str]: + """Every DB name a currently live worktree of this project is entitled + to own: the main workspace DB per branch, plus one `
{suffix}` + sibling per `[tool.pgdevkit].extra_db_suffixes` entry.""" + names: set[str] = set() + for branch in branches: + main_db = workspace_db_name(config.name, branch) + names.add(main_db) + names.update(f"{main_db}{suffix}" for suffix in config.extra_db_suffixes) + return names diff --git a/pyproject.toml b/pyproject.toml index aa97db5..17a6c20 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,7 +11,7 @@ packages = ["pgdevkit"] [project] name = "pgdevkit" -version = "0.6.2" +version = "0.7.0" description = "A helper for developing with Postgres" readme = "README.md" requires-python = ">=3.14" diff --git a/tests/testdb/conftest.py b/tests/testdb/conftest.py index ff75818..bf7530f 100644 --- a/tests/testdb/conftest.py +++ b/tests/testdb/conftest.py @@ -74,3 +74,23 @@ def _factory(name: str, branch: str, engine: str = "postgres") -> Path: return _make_project(tmp_path, name, branch, engine) return _factory + + +@pytest.fixture +def worktree_project_factory(tmp_path: Path) -> Callable[..., tuple[Path, Callable[[str], Path]]]: + """Build a real project repo (with a `[tool.pgdevkit]` pyproject.toml, + like `project_factory`) plus a `add_worktree(branch)` helper that adds a + linked `git worktree` of that same repo -- for tests that need more than + one *live* worktree of one repo, e.g. orphaned-DB detection.""" + + def _factory(name: str, main_branch: str = "main", engine: str = "postgres") -> tuple[Path, Callable[[str], Path]]: + repo = _make_project(tmp_path, name, main_branch, engine) + + def _add_worktree(branch: str) -> Path: + path = tmp_path / f"{name}-{branch}" + subprocess.run(["git", "worktree", "add", "-q", str(path), "-b", branch], cwd=repo, check=True) + return path + + return repo, _add_worktree + + return _factory diff --git a/tests/testdb/test_api.py b/tests/testdb/test_api.py index 65e8c73..b17086c 100644 --- a/tests/testdb/test_api.py +++ b/tests/testdb/test_api.py @@ -1,5 +1,7 @@ from __future__ import annotations +import shutil +import subprocess from pathlib import Path from typing import Callable @@ -7,7 +9,7 @@ import pytest from pgdevkit.testdb import constants -from pgdevkit.testdb.api import clean_testdb, ensure_testdb, reset_testdb, status +from pgdevkit.testdb.api import clean_testdb, ensure_testdb, find_orphaned_dbs, reset_testdb, status from pgdevkit.testdb.config import load_config from pgdevkit.testdb.naming import slugify from tests.testdb.conftest import requires_podman @@ -112,3 +114,93 @@ def test_dsn_for_matches_status(project_factory: Callable[[str, str], Path]): assert dsn_for(project) == status(project)["dsn"] finally: clean_testdb(project) + + +def test_clean_testdb_rejects_all_and_orphaned_together(project_factory: Callable[[str, str], Path]): + project = project_factory("apitest6", "main") + with pytest.raises(ValueError, match="all=True, orphaned=True"): + clean_testdb(project, all=True, orphaned=True) + + +@requires_podman +def test_find_orphaned_dbs_excludes_live_worktrees( + worktree_project_factory: Callable[..., tuple[Path, Callable[[str], Path]]], +): + repo, add_worktree = worktree_project_factory("orphtest") + feature = add_worktree("feature") + ghost = add_worktree("ghost") + try: + ensure_testdb(repo) + ensure_testdb(feature) + ghost_db = ensure_testdb(ghost)["ORPHTEST_POSTGRES_DB"] + subprocess.run(["git", "worktree", "remove", "--force", str(ghost)], cwd=repo, check=True) + + assert find_orphaned_dbs(repo) == [ghost_db] + finally: + clean_testdb(repo, all=True) + + +@requires_podman +def test_find_orphaned_dbs_treats_manually_deleted_worktree_as_orphaned( + worktree_project_factory: Callable[..., tuple[Path, Callable[[str], Path]]], +): + # A worktree dir removed with plain `rm -rf` (no `git worktree remove`) + # still shows up in `git worktree list` as prunable -- it must still be + # treated as not-live. + repo, add_worktree = worktree_project_factory("orphtest2") + ghost = add_worktree("ghost") + try: + ghost_db = ensure_testdb(ghost)["ORPHTEST2_POSTGRES_DB"] + shutil.rmtree(ghost) + + assert find_orphaned_dbs(repo) == [ghost_db] + finally: + clean_testdb(repo, all=True) + + +@requires_podman +def test_clean_orphaned_drops_only_orphaned_dbs( + worktree_project_factory: Callable[..., tuple[Path, Callable[[str], Path]]], +): + repo, add_worktree = worktree_project_factory("orphtest3") + feature = add_worktree("feature") + ghost = add_worktree("ghost") + try: + ensure_testdb(repo) + feature_db = ensure_testdb(feature)["ORPHTEST3_POSTGRES_DB"] + ensure_testdb(ghost) + subprocess.run(["git", "worktree", "remove", "--force", str(ghost)], cwd=repo, check=True) + + clean_testdb(repo, orphaned=True) + + assert find_orphaned_dbs(repo) == [] + with psycopg.connect(constants.conninfo("postgres")) as con: + with con.cursor() as cur: + cur.execute("SELECT count(*) FROM pg_database WHERE datname = %s", (feature_db,)) + (count,) = cur.fetchone() + assert count == 1 + finally: + clean_testdb(repo, all=True) + + +@requires_podman +def test_find_orphaned_dbs_respects_extra_db_suffixes( + worktree_project_factory: Callable[..., tuple[Path, Callable[[str], Path]]], +): + repo, _ = worktree_project_factory("orphtest4") + pyproject = repo / "pyproject.toml" + pyproject.write_text( + pyproject.read_text(encoding="utf-8").rstrip("\n") + '\nextra_db_suffixes = ["_sibling"]\n', + encoding="utf-8", + ) + try: + main_db = ensure_testdb(repo)["ORPHTEST4_POSTGRES_DB"] + sibling_db = f"{main_db}_sibling" + stray_db = f"{main_db}_stray" + with psycopg.connect(constants.conninfo("postgres"), autocommit=True) as con: + con.execute(f'CREATE DATABASE "{sibling_db}"') + con.execute(f'CREATE DATABASE "{stray_db}"') + + assert find_orphaned_dbs(repo) == [stray_db] + finally: + clean_testdb(repo, all=True) diff --git a/tests/testdb/test_cli.py b/tests/testdb/test_cli.py index 8ad607a..41f56c9 100644 --- a/tests/testdb/test_cli.py +++ b/tests/testdb/test_cli.py @@ -105,3 +105,27 @@ def test_testdb_clean(project_factory: Callable[[str, str], Path], monkeypatch): runner.invoke(app, ["testdb", "up"]) result = runner.invoke(app, ["testdb", "clean"]) assert result.exit_code == 0, result.output + + +@requires_podman +def test_testdb_list_orphaned_reports_none_for_a_single_live_worktree( + project_factory: Callable[[str, str], Path], monkeypatch +): + project = project_factory("clitest7", "main") + monkeypatch.chdir(project) + try: + runner.invoke(app, ["testdb", "up"]) + result = runner.invoke(app, ["testdb", "list-orphaned"]) + assert result.exit_code == 0, result.output + assert "No orphaned databases." in result.output + finally: + clean_testdb(project) + + +def test_testdb_clean_rejects_all_and_orphaned_together( + project_factory: Callable[[str, str], Path], monkeypatch +): + project = project_factory("clitest8", "main") + monkeypatch.chdir(project) + result = runner.invoke(app, ["testdb", "clean", "--all", "--orphaned"]) + assert result.exit_code == 2 diff --git a/tests/testdb/test_config.py b/tests/testdb/test_config.py index d9904a9..8a6a641 100644 --- a/tests/testdb/test_config.py +++ b/tests/testdb/test_config.py @@ -69,3 +69,26 @@ def test_extensions_must_be_a_list(tmp_path: Path): ) with pytest.raises(TypeError, match="extensions"): load_config(tmp_path) + + +def test_extra_db_suffixes_defaults_empty(tmp_path: Path): + project = tmp_path / "myproj" + project.mkdir() + config = load_config(project) + assert config.extra_db_suffixes == () + + +def test_reads_extra_db_suffixes(tmp_path: Path): + (tmp_path / "pyproject.toml").write_text( + '[tool.pgdevkit]\nname = "ccmt"\nextra_db_suffixes = ["_onetrade"]\n', encoding="utf-8" + ) + config = load_config(tmp_path) + assert config.extra_db_suffixes == ("_onetrade",) + + +def test_extra_db_suffixes_must_be_a_list(tmp_path: Path): + (tmp_path / "pyproject.toml").write_text( + '[tool.pgdevkit]\nname = "x"\nextra_db_suffixes = "_onetrade"\n', encoding="utf-8" + ) + with pytest.raises(TypeError, match="extra_db_suffixes"): + load_config(tmp_path) diff --git a/tests/testdb/test_naming.py b/tests/testdb/test_naming.py index f394bf4..6a71ebd 100644 --- a/tests/testdb/test_naming.py +++ b/tests/testdb/test_naming.py @@ -1,9 +1,11 @@ from __future__ import annotations +import shutil import subprocess from pathlib import Path -from pgdevkit.testdb.naming import current_branch, slugify, workspace_db_name +from pgdevkit.testdb.config import ProjectConfig +from pgdevkit.testdb.naming import current_branch, expected_db_names, live_worktree_branches, slugify, workspace_db_name def test_slugify_lowercases_and_replaces_invalid_chars(): @@ -49,3 +51,71 @@ def test_current_branch_reads_the_checked_out_branch(tmp_path: Path): subprocess.run(["git", "commit", "-q", "-m", "init"], cwd=tmp_path, check=True) assert current_branch(tmp_path) == "my-feature" + + +def _init_repo(repo: Path, initial_branch: str) -> None: + repo.mkdir() + subprocess.run(["git", "init", "-q"], cwd=repo, check=True) + subprocess.run(["git", "config", "user.email", "t@example.com"], cwd=repo, check=True) + subprocess.run(["git", "config", "user.name", "t"], cwd=repo, check=True) + (repo / "f.txt").write_text("x", encoding="utf-8") + subprocess.run(["git", "add", "."], cwd=repo, check=True) + subprocess.run(["git", "commit", "-q", "-m", "init"], cwd=repo, check=True) + subprocess.run(["git", "checkout", "-q", "-b", initial_branch], cwd=repo, check=True) + + +def test_live_worktree_branches_lists_main_and_linked_worktrees(tmp_path: Path): + repo = tmp_path / "repo" + _init_repo(repo, "main") + feature = tmp_path / "wt-feature" + subprocess.run(["git", "worktree", "add", "-q", str(feature), "-b", "feature"], cwd=repo, check=True) + + assert set(live_worktree_branches(repo)) == {"main", "feature"} + + +def test_live_worktree_branches_excludes_a_removed_worktree(tmp_path: Path): + repo = tmp_path / "repo" + _init_repo(repo, "main") + gone = tmp_path / "wt-gone" + subprocess.run(["git", "worktree", "add", "-q", str(gone), "-b", "gone"], cwd=repo, check=True) + subprocess.run(["git", "worktree", "remove", "--force", str(gone)], cwd=repo, check=True) + + assert live_worktree_branches(repo) == ["main"] + + +def test_live_worktree_branches_reports_detached_head_as_head(tmp_path: Path): + # A detached-HEAD worktree (e.g. a CI checkout, which defaults to one) + # has no `branch ...` porcelain line -- it must still be reported as + # "live", using the same "HEAD" value `current_branch()` (and so + # `workspace_db_name()`) would compute for it, or its database would + # look orphaned and get dropped out from under it. + repo = tmp_path / "repo" + _init_repo(repo, "main") + detached = tmp_path / "wt-detached" + subprocess.run(["git", "worktree", "add", "-q", "--detach", str(detached)], cwd=repo, check=True) + + assert set(live_worktree_branches(repo)) == {"main", "HEAD"} + assert current_branch(detached) == "HEAD" + + +def test_live_worktree_branches_excludes_a_manually_deleted_worktree_dir(tmp_path: Path): + # `rm -rf` on a worktree dir without `git worktree remove` leaves it + # registered (and reported by `git worktree list`) but its path gone -- + # still not "live". + repo = tmp_path / "repo" + _init_repo(repo, "main") + gone = tmp_path / "wt-gone" + subprocess.run(["git", "worktree", "add", "-q", str(gone), "-b", "gone"], cwd=repo, check=True) + shutil.rmtree(gone) + + assert live_worktree_branches(repo) == ["main"] + + +def test_expected_db_names_covers_every_live_branch(): + config = ProjectConfig(name="proj") + assert expected_db_names(config, ["main", "feature"]) == {"proj_main", "proj_feature"} + + +def test_expected_db_names_includes_extra_db_suffixes(): + config = ProjectConfig(name="proj", extra_db_suffixes=("_sibling",)) + assert expected_db_names(config, ["main"]) == {"proj_main", "proj_main_sibling"}