From 7fb229688b2cae4b223860456135b59535d96e6f Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Sat, 7 Nov 2026 15:11:43 +0000 Subject: [PATCH] feat(worktree): keep a per-worktree merge flow that scripts read MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every worktree can carry a deterministic merge flow in /pkgmgr/flow.yml (never committed, gone with the worktree): pull lists the refs merged into the branch, in order, before a gate, deploy or merge; push lists the branches the finished branch is merged into, in order, each then pushed by the operator. - pkgmgr worktree flow [PATH] [--json] shows it; flow set (--pull REF.. | --no-pull) (--push BRANCH.. | --no-push) replaces both lists atomically (0600 temp file, os.replace, pkgmgr/ of mode 0700); flow clear removes it. JSON: worktree, git_dir, branch, flow_branch, pull, push, source (file, none, invalid), warnings. - pkgmgr worktree up gains repeatable --pull/--push; without them pull is the base and push its local branch (origin/main gives main). - pkgmgr branch tree carries "flow": {pull, push, source} or null on every item with a worktree, read from the linked worktree's metadata dir without running git inside the worktree; the Agent Orchestrator shows it. The file and the arguments may come from an agent: every entry must equal its git check-ref-format --branch output (scan mode, cwd /) and must not start with -; the reader opens pkgmgr/ and flow.yml without following links, non-blocking, refuses anything but a regular file, caps at 64 KiB, uses yaml.safe_load, rejects unknown keys, wrong types and bool versions, and never raises into callers: a bad file reads as source invalid with empty lists and a warning. Tests: 30 new (unit and integration on real repositories), unit suite 500 OK, integration 254 with the two known sandbox signing errors of test_publish_integration only. 🎯 Fix: 92 % πŸ’₯ Break: 4 % πŸ’Έ Cost: 97 % ⏱️ Time: 93 % πŸ›‘οΈ Security: 100 % πŸ† Score: 98 🚦 Gate: βœ… 90/5, cost and time <= 100, security 100 % Co-Authored-By: Claude Opus 5.5 --- README.md | 45 +++- src/pkgmgr/actions/branch/tree.py | 29 +- src/pkgmgr/actions/worktree/flow.py | 237 +++++++++++++++++ src/pkgmgr/actions/worktree/flow_actions.py | 158 +++++++++++ src/pkgmgr/actions/worktree/slots.py | 23 +- src/pkgmgr/actions/worktree/up.py | 62 ++++- src/pkgmgr/cli/commands/worktree.py | 60 ++++- src/pkgmgr/cli/parser/worktree_cmd.py | 60 ++++- tests/integration/test_branch_tree_flow.py | 105 ++++++++ tests/integration/test_worktree.py | 4 + tests/integration/test_worktree_flow.py | 247 ++++++++++++++++++ .../unit/pkgmgr/actions/worktree/test_flow.py | 176 +++++++++++++ 12 files changed, 1196 insertions(+), 10 deletions(-) create mode 100644 src/pkgmgr/actions/worktree/flow.py create mode 100644 src/pkgmgr/actions/worktree/flow_actions.py create mode 100644 tests/integration/test_branch_tree_flow.py create mode 100644 tests/integration/test_worktree_flow.py create mode 100644 tests/unit/pkgmgr/actions/worktree/test_flow.py diff --git a/README.md b/README.md index 73c8b75..03447bc 100644 --- a/README.md +++ b/README.md @@ -74,7 +74,10 @@ versioning features it can drive **end-to-end workflows**: | Command | Does | | --- | --- | -| `pkgmgr worktree up [--base REF] [--path DIR]` | Checks the branch out under `////` or at `--path`, assigns a free slot and runs the pinned `worktree-init` hook | +| `pkgmgr worktree up [--base REF] [--path DIR] [--pull REF ...] [--push BRANCH ...]` | Checks the branch out under `////` or at `--path`, assigns a free slot, records the merge flow and runs the pinned `worktree-init` hook | +| `pkgmgr worktree flow [PATH] [--json]` | Shows the merge flow of the worktree at `PATH` (default: `.`) | +| `pkgmgr worktree flow set [PATH] (--pull REF ... \| --no-pull) (--push BRANCH ... \| --no-push) [--json]` | Replaces both lists of the merge flow | +| `pkgmgr worktree flow clear [PATH] [--json]` | Removes the merge flow | | `pkgmgr worktree cleanup [--apply]` | Removes clean worktrees whose commits are merged or pushed: pinned `worktree-fini` hook, compose stack, checkout; the branch stays | | `pkgmgr branch cleanup [--apply]` | Removes merged branches that no worktree holds, locally and on the remotes | | `pkgmgr branch tree [--format json\|yaml] [--output FILE]` | Writes every managed repository's branches and worktrees with their cleanup state, without fetching | @@ -108,6 +111,46 @@ Every cleanup is a dry run until `--apply`, and `--json` prints the plan for other tools. The [agent skills](https://github.com/kevinveenbirkenbach/skills) wrap these commands; see [AGENTS.md](AGENTS.md) for the split. +#### Merge flow + +Every worktree can carry a merge flow that scripts and skills read: `pull` +lists the refs merged into the branch, in order, before a gate, deploy or +merge; `push` lists the branches the branch is merged into, in order, when it +is finished, each then pushed by the operator. It lives in +`pkgmgr/flow.yml` inside the worktree's own git directory (`.git` of the main +worktree, `/worktrees/` of a linked one), so it is never committed +and goes with the worktree: + +```yaml +version: 1 +branch: feature/x +pull: + - origin/main + - store-i18n +push: + - feature/iam-profile-rbac + - production +``` + +`branch` is the branch the flow was set on; another current branch shows as a +warning. `worktree up` records `--pull` and `--push`; without them pull is +the base the worktree came from and push that base's local branch +(`origin/main` gives `main`). `flow set` needs each list or +its `--no-pull`/`--no-push`, and writes atomically into a `pkgmgr/` of mode +0700. Every entry must pass `git check-ref-format --branch` and must not start +with `-`; a remote-tracking name such as `origin/main` passes. An invalid name +exits 2 and writes nothing. A flow file that is a symbolic link, no regular +file, larger than 64 KiB or malformed reads as `invalid`, with empty lists and +a warning, and is never followed. + +`flow --json` prints `worktree` and `git_dir` (absolute paths), `branch` (the +current branch, `null` when detached), `flow_branch` (the file's `branch`), +`pull`, `push`, `source` (`file`, `none` or `invalid`) and `warnings`; it +exits 0 also without a flow, 2 when `PATH` is not a git worktree. In +`pkgmgr branch tree` every item carries `flow`: `{"pull", "push", "source"}`, +or `null` without a worktree or a flow; flow warnings join the repository's +`warnings`. + --- ## Architecture & Setup Map πŸ—ΊοΈ diff --git a/src/pkgmgr/actions/branch/tree.py b/src/pkgmgr/actions/branch/tree.py index cabc752..9fe9f2d 100644 --- a/src/pkgmgr/actions/branch/tree.py +++ b/src/pkgmgr/actions/branch/tree.py @@ -14,6 +14,7 @@ from datetime import datetime, timezone import yaml +from pkgmgr.actions.worktree import flow from pkgmgr.core.credentials.store_keys import build_keyring_key from pkgmgr.core.git.errors import GitBaseError @@ -193,6 +194,27 @@ def _missing(root: str, known: bool, scanned_at: str) -> dict: } +def _git_dir( + root: str, registrations: gc.Registrations, worktree: str | None +) -> str | None: + if worktree is None: + return None + if worktree == root: + return os.path.dirname(registrations[0]) + return gc._linked_meta(registrations, worktree) + + +def _flow(git_dir: str | None, item: dict, warnings: list[str]) -> dict | None: + found = flow.read_flow(git_dir) if git_dir else None + if found is None or found.source == "none": + return None + warnings += [ + f"flow of {item['worktree']}: {warning}" + for warning in found.warnings + flow.mismatch(found, item["branch"]) + ] + return {"pull": found.pull, "push": found.push, "source": found.source} + + def _scan( root: str, known: bool, @@ -227,16 +249,21 @@ def _scan( fetch=False, forge_source=source, ) + registrations = gc._registrations(root) except GitBaseError as exc: entry["warnings"] = [f"scan failed: {str(exc).strip().splitlines()[0]}"] return entry owner = next((r for r in repo.remotes if repo.default.startswith(f"{r}/")), None) + items = [asdict(item) for item in repo.items] + for item in items: + git_dir = _git_dir(root, registrations, item["worktree"]) + item["flow"] = _flow(git_dir, item, repo.warnings) entry.update( remote=repo.urls[repo.remotes[0]] if repo.remotes else None, default=repo.default.removeprefix(f"{owner}/") if owner else repo.default, forge=sources[0], warnings=repo.warnings, - items=[asdict(item) for item in repo.items], + items=items, ) return entry diff --git a/src/pkgmgr/actions/worktree/flow.py b/src/pkgmgr/actions/worktree/flow.py new file mode 100644 index 0000000..210f283 --- /dev/null +++ b/src/pkgmgr/actions/worktree/flow.py @@ -0,0 +1,237 @@ +from __future__ import annotations + +import os +import secrets +import stat +from contextlib import suppress +from dataclasses import dataclass, field + +import yaml + +from pkgmgr.actions.branch.nofollow import DIR_FLAGS +from pkgmgr.core.git.errors import GitBaseError +from pkgmgr.core.git.run import run + +FLOW_DIR = "pkgmgr" +FLOW_FILE = "flow.yml" +RELATIVE = f"{FLOW_DIR}/{FLOW_FILE}" +VERSION = 1 +MAX_BYTES = 64 * 1024 +FOLDER_MODE = 0o700 +FILE_MODE = 0o600 +KEYS = ("version", "branch", "pull", "push") +REQUIRED = ("version", "pull", "push") +# O_NONBLOCK: opening a planted FIFO for reading would block until a writer comes +READ_FLAGS = os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK +WRITE_FLAGS = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW +DETACHED = "a detached HEAD" +NONE_LISTED = "(none)" + + +class FlowError(ValueError): + pass + + +@dataclass +class Flow: + pull: list[str] = field(default_factory=list) + push: list[str] = field(default_factory=list) + branch: str | None = None + source: str = "none" + warnings: list[str] = field(default_factory=list) + + +def describe(exc: BaseException) -> str: + return (str(exc).strip().splitlines() or [type(exc).__name__])[0] + + +def _ref_format(name: str) -> bool: + # check-ref-format needs no repository, so / keeps any repository config + # out; --branch would expand @{-N} into another name inside one + try: + return ( + run(["check-ref-format", "--branch", name], cwd=os.sep, scan=True) == name + ) + except (GitBaseError, OSError, ValueError): + return False + + +def check_name(kind: str, name: object) -> str: + """ + Args: + kind: pull, push or branch, for the message. + name: an entry from the file or the command line. + + Returns: + name, once it is a string git accepts as a branch name that does not + start with `-`; a remote-tracking name such as origin/main passes. + + Raises: + FlowError: name is anything else. + """ + if not isinstance(name, str) or name.startswith("-") or not _ref_format(name): + raise FlowError(f"{kind} {name!r} is not a valid branch name") + return name + + +def check_names(kind: str, names: object) -> list[str]: + """ + Args: + kind: pull or push, for the message. + names: the list from the file or the command line. + + Returns: + The names in order, each passed through check_name. + + Raises: + FlowError: names is not a list or an entry is invalid. + """ + if not isinstance(names, list): + raise FlowError(f"{kind} must be a list of branch names") + return [check_name(kind, name) for name in names] + + +def _folder(git_dir: str, create: bool) -> int: + parent = os.open(git_dir, os.O_RDONLY | os.O_DIRECTORY) + try: + if create: + with suppress(FileExistsError): + os.mkdir(FLOW_DIR, FOLDER_MODE, dir_fd=parent) + return os.open(FLOW_DIR, DIR_FLAGS, dir_fd=parent) + finally: + os.close(parent) + + +def _read_text(git_dir: str) -> str | None: + try: + folder = _folder(git_dir, create=False) + except FileNotFoundError: + return None + try: + fd = os.open(FLOW_FILE, READ_FLAGS, dir_fd=folder) + except FileNotFoundError: + return None + finally: + os.close(folder) + # os.fdopen refuses a directory but leaves its fd open, so check first + if not stat.S_ISREG(os.fstat(fd).st_mode): + os.close(fd) + raise FlowError("not a regular file") + with os.fdopen(fd, "rb") as fh: + data = fh.read(MAX_BYTES + 1) + if len(data) > MAX_BYTES: + raise FlowError(f"larger than {MAX_BYTES} bytes") + return data.decode("utf-8") + + +def _parse(text: str) -> tuple[str | None, list[str], list[str]]: + data = yaml.safe_load(text) + if not isinstance(data, dict): + raise FlowError("not a mapping") + unknown = [repr(key) for key in data if key not in KEYS] + if unknown: + raise FlowError(f"unknown keys {', '.join(unknown)}") + missing = [key for key in REQUIRED if key not in data] + if missing: + raise FlowError(f"missing keys {', '.join(missing)}") + # bool is an int, and True == 1 + if type(data["version"]) is not int or data["version"] != VERSION: + raise FlowError(f"version must be {VERSION}") + branch = data.get("branch") + if branch is not None: + check_name("branch", branch) + return branch, check_names("pull", data["pull"]), check_names("push", data["push"]) + + +def read_flow(git_dir: str) -> Flow: + """ + Read `/pkgmgr/flow.yml`, which an agent may have written. + + Args: + git_dir: the worktree's own git directory. + + Returns: + The flow with source file; source none when there is no file; source + invalid, empty lists and a warning when the file is a symbolic link, + no regular file, larger than MAX_BYTES or malformed. Never raises. + """ + try: + text = _read_text(git_dir) + if text is None: + return Flow() + branch, pull, push = _parse(text) + # RecursionError: safe_load recurses once per nesting level + except (OSError, ValueError, yaml.YAMLError, RecursionError) as exc: + return Flow(source="invalid", warnings=[f"{RELATIVE} ignored: {describe(exc)}"]) + return Flow(pull=pull, push=push, branch=branch, source="file") + + +def write_flow( + git_dir: str, branch: str | None, pull: list[str], push: list[str] +) -> None: + """ + Replace the flow atomically, never writing through a symbolic link. + + Args: + git_dir: the worktree's own git directory. + branch: the worktree's branch now; None when detached. + pull: refs merged into the branch, in order, already through check_names. + push: branches the branch is merged into, in order, likewise checked. + + Raises: + OSError: `pkgmgr` is a symbolic link or no directory, or a write failed. + """ + data = {"version": VERSION, "branch": branch, "pull": pull, "push": push} + text = yaml.safe_dump(data, sort_keys=False, default_flow_style=False) + folder = _folder(git_dir, create=True) + tmp = f".{FLOW_FILE}.{secrets.token_hex(8)}" + try: + fd = os.open(tmp, WRITE_FLAGS, FILE_MODE, dir_fd=folder) + with os.fdopen(fd, "w", encoding="utf-8") as fh: + fh.write(text) + fh.flush() + os.fsync(fh.fileno()) + os.replace(tmp, FLOW_FILE, src_dir_fd=folder, dst_dir_fd=folder) + finally: + with suppress(FileNotFoundError): + os.unlink(tmp, dir_fd=folder) + os.close(folder) + + +def clear_flow(git_dir: str) -> None: + """ + Remove the flow file; a symbolic link in its place goes, never its target. + + Args: + git_dir: the worktree's own git directory. + + Raises: + OSError: `pkgmgr` is a symbolic link or no directory, or the unlink failed. + """ + try: + folder = _folder(git_dir, create=False) + except FileNotFoundError: + return + try: + with suppress(FileNotFoundError): + os.unlink(FLOW_FILE, dir_fd=folder) + finally: + os.close(folder) + + +def mismatch(flow: Flow, branch: str | None) -> list[str]: + """ + Args: + flow: a flow from read_flow. + branch: the worktree's current branch; None when detached. + + Returns: + One warning when the file was written on another branch, else none. + """ + if flow.source != "file" or flow.branch == branch: + return [] + warning = ( + f"flow was set on {flow.branch or DETACHED}, " + f"the worktree is on {branch or DETACHED}" + ) + return [warning] diff --git a/src/pkgmgr/actions/worktree/flow_actions.py b/src/pkgmgr/actions/worktree/flow_actions.py new file mode 100644 index 0000000..3eb8372 --- /dev/null +++ b/src/pkgmgr/actions/worktree/flow_actions.py @@ -0,0 +1,158 @@ +from __future__ import annotations + +import json +import sys + +from pkgmgr.core.git.errors import GitBaseError +from pkgmgr.core.git.run import run + +from .flow import ( + NONE_LISTED, + Flow, + FlowError, + check_names, + clear_flow, + describe, + mismatch, + read_flow, + write_flow, +) + +Located = tuple[str, str, str | None] + + +def locate(path: str) -> Located: + """ + Args: + path: any directory inside the worktree. + + Returns: + The worktree root, its own git directory (`.git` of the main + worktree, `/worktrees/` of a linked one) and its branch, + None when detached. + + Raises: + GitBaseError: path is not inside a worktree, e.g. a bare repository. + OSError: path is no directory. + ValueError: a path git printed spans lines. + """ + out = run( + ["rev-parse", "--absolute-git-dir", "--show-toplevel"], cwd=path, scan=True + ) + git_dir, top = out.split("\n") + try: + ref = run(["symbolic-ref", "--quiet", "HEAD"], cwd=top, scan=True) + except GitBaseError: + ref = "" + branch = ref.removeprefix("refs/heads/") if ref.startswith("refs/heads/") else None + return top, git_dir, branch + + +def _located(path: str) -> Located | None: + try: + return locate(path) + except (GitBaseError, OSError, ValueError) as exc: + print( + f"Refusing: {path} is not a git worktree: {describe(exc)}", + file=sys.stderr, + ) + return None + + +def _report(located: Located, flow: Flow, as_json: bool) -> None: + top, git_dir, branch = located + result = { + "worktree": top, + "git_dir": git_dir, + "branch": branch, + "flow_branch": flow.branch, + "pull": flow.pull, + "push": flow.push, + "source": flow.source, + "warnings": flow.warnings + mismatch(flow, branch), + } + if as_json: + print(json.dumps(result, indent=2)) + return + for warning in result["warnings"]: + print(f"WARN {warning}") + for key in ("pull", "push"): + print(f"{key}: {', '.join(result[key]) or NONE_LISTED}") + + +def worktree_flow_show(path: str, *, as_json: bool) -> int: + """ + Print the flow of the worktree at path. + + Args: + path: any directory inside the worktree. + as_json: print the result as JSON instead of text. + + Returns: + 0, also without a flow or with an invalid file; 2 when path is not a + git worktree. + """ + located = _located(path) + if located is None: + return 2 + _report(located, read_flow(located[1]), as_json) + return 0 + + +def worktree_flow_set( + path: str, *, pull: list[str], push: list[str], as_json: bool +) -> int: + """ + Replace both lists of the worktree's flow and print the result. + + Args: + path: any directory inside the worktree. + pull: refs merged into the branch before a gate, deploy or merge, in + order; may be empty. + push: branches the branch is merged into when finished, in order; + may be empty. + as_json: print the result as JSON instead of text. + + Returns: + 0 when written, 1 when the write failed, 2 on an invalid name or when + path is not a git worktree; nothing is written unless 0. + """ + try: + pull, push = check_names("pull", pull), check_names("push", push) + except FlowError as exc: + print(f"Refusing: {exc}", file=sys.stderr) + return 2 + located = _located(path) + if located is None: + return 2 + try: + write_flow(located[1], located[2], pull, push) + except OSError as exc: + print(f"Cannot write the flow of {located[0]}: {exc}", file=sys.stderr) + return 1 + _report(located, read_flow(located[1]), as_json) + return 0 + + +def worktree_flow_clear(path: str, *, as_json: bool) -> int: + """ + Remove the worktree's flow and print the empty result. + + Args: + path: any directory inside the worktree. + as_json: print the result as JSON instead of text. + + Returns: + 0 when no flow is left, 1 when the removal failed, 2 when path is not + a git worktree. + """ + located = _located(path) + if located is None: + return 2 + try: + clear_flow(located[1]) + except OSError as exc: + print(f"Cannot clear the flow of {located[0]}: {exc}", file=sys.stderr) + return 1 + _report(located, read_flow(located[1]), as_json) + return 0 diff --git a/src/pkgmgr/actions/worktree/slots.py b/src/pkgmgr/actions/worktree/slots.py index bb06619..e6fb90c 100644 --- a/src/pkgmgr/actions/worktree/slots.py +++ b/src/pkgmgr/actions/worktree/slots.py @@ -24,6 +24,21 @@ def next_slot(used: set[int]) -> int: return slot +def new_worktree_meta(path: str) -> str: + """ + Args: + path: a worktree this process just added, whose `.git` file names + its metadata. + + Returns: + The worktree's own git directory, `/worktrees/`. + """ + with open(os.path.join(path, ".git"), encoding="utf-8") as fh: + return os.path.normpath( + os.path.join(path, fh.read().strip().removeprefix("gitdir: ")) + ) + + def write_slot(path: str, slot: int) -> None: """ Record a slot in the metadata of a worktree this process just added. @@ -32,9 +47,7 @@ def write_slot(path: str, slot: int) -> None: path: the new worktree, whose `.git` file names its metadata. slot: the slot to record. """ - with open(os.path.join(path, ".git"), encoding="utf-8") as fh: - meta = os.path.normpath( - os.path.join(path, fh.read().strip().removeprefix("gitdir: ")) - ) - with open(os.path.join(meta, SLOT_FILE), "w", encoding="utf-8") as fh: + with open( + os.path.join(new_worktree_meta(path), SLOT_FILE), "w", encoding="utf-8" + ) as fh: fh.write(f"{slot}\n") diff --git a/src/pkgmgr/actions/worktree/up.py b/src/pkgmgr/actions/worktree/up.py index e513ddf..f127eb8 100644 --- a/src/pkgmgr/actions/worktree/up.py +++ b/src/pkgmgr/actions/worktree/up.py @@ -9,8 +9,9 @@ from pkgmgr.core.git.errors import GitRunError from pkgmgr.core.git.queries import FILTER_CONFIG, local_config from pkgmgr.core.git.run import END_OF_OPTIONS, run +from .flow import NONE_LISTED, FlowError, check_name, check_names, write_flow from .hooks import INIT_HOOK, defined_in, hook_command, run_hook -from .slots import next_slot, used_slots, write_slot +from .slots import new_worktree_meta, next_slot, used_slots, write_slot PORT_PREFIX = re.compile(r"^\d+/") HOST_AND_PATH = re.compile(r"([^:/]+)[:/](.*)") @@ -81,6 +82,51 @@ def _default_base(repo: str) -> str: return "origin/main" if _ref_exists(repo, "origin/main") else "origin/master" +def _local_branch(repo: str, ref: str) -> str: + try: + remotes = run(["remote"], cwd=repo, scan=True).split() + except GitRunError: + return ref + owner = next((r for r in remotes if ref.startswith(f"{r}/")), None) + return ref.removeprefix(f"{owner}/") if owner else ref + + +def _invalid_flow(pull: list[str] | None, push: list[str] | None) -> str | None: + try: + for kind, names in (("pull", pull), ("push", push)): + if names is not None: + check_names(kind, names) + except FlowError as exc: + return str(exc) + return None + + +def _record_flow( + root: str, + path: str, + branch: str, + lists: tuple[list[str] | None, list[str] | None], + base: str | None, + warnings: list[str], +) -> dict[str, list[str]] | None: + pull, push = lists + if pull is None or push is None: + default = base or _default_base(root) + try: + check_name("base", default) + except FlowError as exc: + warnings.append(f"flow not set: {exc}") + return None + pull = [default] if pull is None else pull + push = [_local_branch(root, default)] if push is None else push + try: + write_flow(new_worktree_meta(path), branch, pull, push) + except OSError as exc: + warnings.append(f"flow not written: {exc}") + return None + return {"pull": pull, "push": push} + + def worktree_up( repo_dir: str, branch: str, @@ -90,6 +136,8 @@ def worktree_up( path: str | None, hooks: dict[str, dict[str, list[str]]], as_json: bool, + pull: list[str] | None, + push: list[str] | None, ) -> int: """ Check a branch out into its own worktree under the standard path. @@ -106,6 +154,10 @@ def worktree_up( CLI; the worktree-init hook runs only when the repository root is pinned there, never from the Makefile or by origin's URL. as_json: print the result as JSON instead of text. + pull: refs merged into the branch before a gate, deploy or merge, in + order, for the worktree's flow; None takes the base. + push: branches the branch is merged into when finished, in order; + None takes the base. Returns: 0 when the worktree is ready, 1 when the worktree-init hook failed @@ -127,6 +179,10 @@ def worktree_up( file=sys.stderr, ) return 2 + invalid = _invalid_flow(pull, push) + if invalid: + print(f"Refusing: {invalid}", file=sys.stderr) + return 2 location = worktree_location(url) if path: @@ -192,6 +248,7 @@ def worktree_up( return 2 write_slot(path, slot) + flow = _record_flow(root, path, branch, (pull, push), start or base, warnings) hook = "none" code = 0 @@ -214,6 +271,7 @@ def worktree_up( "origin": origin, "slot": slot, "hook": hook, + "flow": flow, "warnings": warnings, } if as_json: @@ -223,4 +281,6 @@ def worktree_up( print(f"WARN {warning}") for key in ("path", "branch", "origin", "slot", "hook"): print(f"{key:<8} {result[key]}") + for key, names in (flow or {}).items(): + print(f"{key:<8} {', '.join(names) or NONE_LISTED}") return code diff --git a/src/pkgmgr/cli/commands/worktree.py b/src/pkgmgr/cli/commands/worktree.py index 049afce..06bfd2f 100644 --- a/src/pkgmgr/cli/commands/worktree.py +++ b/src/pkgmgr/cli/commands/worktree.py @@ -6,6 +6,11 @@ import sys from pkgmgr.actions.branch.gc import main_repo_root from pkgmgr.actions.gc.report import disk_report, render_text +from pkgmgr.actions.worktree.flow_actions import ( + worktree_flow_clear, + worktree_flow_set, + worktree_flow_show, +) from pkgmgr.actions.worktree.orphans import worktree_orphans from pkgmgr.actions.worktree.up import worktree_up from pkgmgr.cli.context import CLIContext @@ -19,13 +24,61 @@ from .branch import ( ) MEBIBYTE = 1 << 20 +FLOW_ACTIONS = ("set", "clear") + + +def _flow_usage(action: str, args) -> str | None: + if action == "show" and args.path is not None: + return f"unknown flow action {args.target!r}; use set or clear" + if action != "set": + flags = { + "--pull": args.pull, + "--push": args.push, + "--no-pull": args.no_pull, + "--no-push": args.no_push, + } + used = [flag for flag, value in flags.items() if value] + return f"{', '.join(used)} only with `flow set`" if used else None + for kind in ("pull", "push"): + if bool(getattr(args, kind)) == getattr(args, f"no_{kind}"): + return f"flow set takes either --{kind} or --no-{kind}" + return None + + +def _flow(args) -> int: + action, path = ( + (args.target, args.path) + if args.target in FLOW_ACTIONS + else ("show", args.target) + ) + problem = _flow_usage(action, args) + if problem: + print(f"Refusing: {problem}", file=sys.stderr) + return 2 + target = path or "." + if action == "show": + return worktree_flow_show(target, as_json=args.json) + if action == "set": + code = worktree_flow_set( + target, pull=args.pull or [], push=args.push or [], as_json=args.json + ) + else: + code = worktree_flow_clear(target, as_json=args.json) + if code == 0: + refresh_tree([main_repo_root(target)]) + return code def handle_worktree(args, ctx: CLIContext) -> None: """ Handle `pkgmgr worktree` subcommands. - - pkgmgr worktree up [--base REF] [--path DIR] [--json] + - pkgmgr worktree up [--base REF] [--path DIR] [--pull REF ...] + [--push BRANCH ...] [--json] + - pkgmgr worktree flow [PATH] [--json] + - pkgmgr worktree flow set [PATH] (--pull REF ... | --no-pull) + (--push BRANCH ... | --no-push) [--json] + - pkgmgr worktree flow clear [PATH] [--json] - pkgmgr worktree cleanup [--apply] [--json] [scan options] [--min-age HOURS] - pkgmgr worktree orphans --stale-days N --top N [--json] [scan options] """ @@ -38,10 +91,15 @@ def handle_worktree(args, ctx: CLIContext) -> None: path=args.path, hooks=pinned_hooks(ctx), as_json=args.json, + pull=args.pull, + push=args.push, ) refresh_tree([main_repo_root(".")]) sys.exit(code) + if args.subcommand == "flow": + sys.exit(_flow(args)) + if args.subcommand == "cleanup": run_cleanup(args, ctx, "worktree") diff --git a/src/pkgmgr/cli/parser/worktree_cmd.py b/src/pkgmgr/cli/parser/worktree_cmd.py index 77e3eb7..8121f46 100644 --- a/src/pkgmgr/cli/parser/worktree_cmd.py +++ b/src/pkgmgr/cli/parser/worktree_cmd.py @@ -5,10 +5,65 @@ import argparse from .branch_cmd import add_apply_options, add_min_age_option, add_scan_options +def _add_flow_options(parser: argparse.ArgumentParser, default: str) -> None: + parser.add_argument( + "--pull", + action="append", + metavar="REF", + help=( + "Ref merged into the branch before a gate, deploy or merge; " + f"repeatable, kept in order ({default})" + ), + ) + parser.add_argument( + "--push", + action="append", + metavar="BRANCH", + help=( + "Branch the branch is merged into when finished, then pushed by " + f"the operator; repeatable, kept in order ({default})" + ), + ) + + +def _add_flow_parser(worktree_subparsers: argparse._SubParsersAction) -> None: + flow = worktree_subparsers.add_parser( + "flow", + help=( + "Show, set or clear the worktree's merge flow in " + "/pkgmgr/flow.yml: pull refs merged in, push branches " + "merged into" + ), + ) + flow.add_argument( + "target", + nargs="?", + metavar="set|clear|PATH", + help="`set` or `clear`, else the worktree to show (default: .)", + ) + flow.add_argument( + "path", + nargs="?", + metavar="PATH", + help="Worktree of set or clear (default: .)", + ) + _add_flow_options(flow, "set only, replaces the list") + flow.add_argument( + "--no-pull", action="store_true", help="set only: an empty pull list" + ) + flow.add_argument( + "--no-push", action="store_true", help="set only: an empty push list" + ) + flow.add_argument("--json", action="store_true", help="Print the flow as JSON") + + def add_worktree_subparsers(subparsers: argparse._SubParsersAction) -> None: worktree_parser = subparsers.add_parser( "worktree", - help="Worktree utilities: create, clean up and find abandoned worktrees", + help=( + "Worktree utilities: create, keep a merge flow, clean up and find " + "abandoned worktrees" + ), ) worktree_subparsers = worktree_parser.add_subparsers( dest="subcommand", @@ -38,8 +93,11 @@ def add_worktree_subparsers(subparsers: argparse._SubParsersAction) -> None: "an agent sandbox releases" ), ) + _add_flow_options(up, "default: the base, for push its local branch") up.add_argument("--json", action="store_true", help="Print the result as JSON") + _add_flow_parser(worktree_subparsers) + cleanup = worktree_subparsers.add_parser( "cleanup", help=( diff --git a/tests/integration/test_branch_tree_flow.py b/tests/integration/test_branch_tree_flow.py new file mode 100644 index 0000000..0b592ba --- /dev/null +++ b/tests/integration/test_branch_tree_flow.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +import io +import json +import os +import unittest +from contextlib import redirect_stdout +from unittest.mock import patch + +from pkgmgr.actions.branch import gc, tree +from pkgmgr.actions.worktree import flow +from tests.integration.test_branch_gc import GcFixture + + +class TestBranchTreeFlow(GcFixture, unittest.TestCase): + def setUp(self) -> None: + super().setUp() + self.shared = os.path.join(self.root, "shared-worktrees") + os.makedirs(self.shared) + self.metas = os.path.join(self.repo, ".git", "worktrees") + for patcher in ( + patch.object(gc, "host_pid_namespace", return_value=True), + patch.object(gc, "list_compose_containers", return_value=[]), + patch.object(gc, "_process_cwds", return_value={}), + patch.object(tree, "secret_service_token", return_value=(None, "locked")), + ): + patcher.start() + self.addCleanup(patcher.stop) + + def _entry(self) -> dict: + out = io.StringIO() + with redirect_stdout(out): + code = tree.branch_tree( + [self.repo], fmt="json", output=None, worktrees_root=self.shared + ) + self.assertEqual(code, 0) + return json.loads(out.getvalue())["repos"][0] + + def _flows(self, entry: dict) -> dict[str, dict | None]: + return { + item["branch"]: item["flow"] for item in entry["items"] if item["branch"] + } + + def _flow_warnings(self, entry: dict) -> list[str]: + return [w for w in entry["warnings"] if w.startswith("flow of ")] + + def test_items_carry_their_flow_or_null(self) -> None: + flow.write_flow( + os.path.join(self.metas, "wt-merged"), "merged", ["origin/main"], ["prod"] + ) + flow.write_flow(os.path.join(self.repo, ".git"), "main", [], ["origin/main"]) + + entry = self._entry() + flows = self._flows(entry) + self.assertEqual( + flows["merged"], + {"pull": ["origin/main"], "push": ["prod"], "source": "file"}, + ) + self.assertEqual( + flows["main"], {"pull": [], "push": ["origin/main"], "source": "file"} + ) + self.assertIsNone(flows["dirty"]) + self.assertIsNone(flows["wip"]) + self.assertTrue(all("flow" in item for item in entry["items"])) + self.assertEqual(self._flow_warnings(entry), []) + + def test_invalid_and_mismatched_flows_warn_without_failing(self) -> None: + folder = os.path.join(self.metas, "wt-dirty", flow.FLOW_DIR) + os.makedirs(folder) + with open(os.path.join(folder, flow.FLOW_FILE), "w", encoding="utf-8") as fh: + fh.write("version: 1\npull: [a..b]\npush: []\n") + flow.write_flow(os.path.join(self.metas, "wt-merged"), "gone", ["main"], []) + + entry = self._entry() + flows = self._flows(entry) + self.assertEqual(flows["dirty"], {"pull": [], "push": [], "source": "invalid"}) + self.assertEqual(flows["merged"]["source"], "file") + warnings = self._flow_warnings(entry) + self.assertEqual(len(warnings), 2) + self.assertTrue( + any( + w.startswith(f"flow of {self.dirty_wt}: pkgmgr/flow.yml") + for w in warnings + ) + ) + self.assertIn( + f"flow of {self.merged_wt}: flow was set on gone, the worktree is on merged", + warnings, + ) + + def test_a_forged_registration_reads_no_flow(self) -> None: + flow.write_flow(os.path.join(self.metas, "wt-merged"), "merged", ["main"], []) + with open(os.path.join(self.metas, "wt-locked", "gitdir"), "w") as fh: + fh.write(os.path.join(self.merged_wt, ".git") + "\n") + + items = {item["branch"]: item for item in self._entry()["items"]} + + for branch in ("merged", "locked"): + with self.subTest(branch=branch): + self.assertEqual(items[branch]["worktree"], self.merged_wt) + self.assertIsNone(items[branch]["flow"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/integration/test_worktree.py b/tests/integration/test_worktree.py index 7fa80f9..b8b6dc3 100644 --- a/tests/integration/test_worktree.py +++ b/tests/integration/test_worktree.py @@ -314,6 +314,8 @@ class TestWorktreeUp(WorktreeFixture, unittest.TestCase): path=path, hooks=hooks or {}, as_json=True, + pull=None, + push=None, ) return code, json.loads(out.getvalue()) if code != 2 else {} @@ -729,6 +731,8 @@ class TestWorktreeCli(unittest.TestCase): path="/srv/wt/x", hooks={os.path.realpath("/srv/host/o/r"): pins}, as_json=True, + pull=None, + push=None, ) refresh.assert_called_once() diff --git a/tests/integration/test_worktree_flow.py b/tests/integration/test_worktree_flow.py new file mode 100644 index 0000000..1e682f5 --- /dev/null +++ b/tests/integration/test_worktree_flow.py @@ -0,0 +1,247 @@ +from __future__ import annotations + +import io +import json +import os +import unittest +from contextlib import redirect_stderr, redirect_stdout +from types import SimpleNamespace +from unittest.mock import patch + +import yaml + +from pkgmgr.actions.worktree import up +from pkgmgr.cli.commands.worktree import handle_worktree +from pkgmgr.cli.parser import create_parser +from tests.integration.test_branch_gc import GcFixture, git + +PULL = ["origin/main", "store-i18n"] +PUSH = ["feature/iam-profile-rbac", "production"] + + +class FlowFixture(GcFixture): + def setUp(self) -> None: + super().setUp() + self.linked_git_dir = os.path.join(self.repo, ".git", "worktrees", "wt-merged") + self.linked_file = os.path.join(self.linked_git_dir, "pkgmgr", "flow.yml") + self.addCleanup(os.chdir, os.getcwd()) + + def _cli(self, *argv: str) -> tuple[int, str, str]: + out, err = io.StringIO(), io.StringIO() + args = create_parser("test").parse_args(["worktree", *argv]) + with ( + patch("pkgmgr.cli.commands.worktree.refresh_tree") as self.refresh, + redirect_stdout(out), + redirect_stderr(err), + self.assertRaises(SystemExit) as done, + ): + handle_worktree(args, SimpleNamespace()) + return done.exception.code, out.getvalue(), err.getvalue() + + def _json(self, *argv: str) -> dict: + code, out, err = self._cli(*argv, "--json") + self.assertEqual(code, 0, err) + return json.loads(out) + + +class TestFlowCli(FlowFixture, unittest.TestCase): + def test_set_show_clear_json_on_a_linked_worktree(self) -> None: + argv = [f"--pull={ref}" for ref in PULL] + [f"--push={b}" for b in PUSH] + expected = { + "worktree": self.merged_wt, + "git_dir": self.linked_git_dir, + "branch": "merged", + "flow_branch": "merged", + "pull": PULL, + "push": PUSH, + "source": "file", + "warnings": [], + } + + self.assertEqual(self._json("flow", "set", self.merged_wt, *argv), expected) + self.refresh.assert_called_once_with([self.repo]) + self.assertEqual(self._json("flow", self.merged_wt), expected) + self.assertTrue(os.path.isfile(self.linked_file)) + self.assertEqual(git(self.merged_wt, "status", "--porcelain"), "") + cleared = self._json("flow", "clear", self.merged_wt) + + self.assertEqual( + {**expected, "flow_branch": None, "pull": [], "push": [], "source": "none"}, + cleared, + ) + self.assertFalse(os.path.exists(self.linked_file)) + + def test_set_show_clear_text_on_the_main_worktree(self) -> None: + os.chdir(self.repo) + + code, out, _ = self._cli("flow", "set", "--pull", "origin/main", "--no-push") + self.assertEqual((code, out), (0, "pull: origin/main\npush: (none)\n")) + self.assertEqual( + self._cli("flow")[:2], (0, "pull: origin/main\npush: (none)\n") + ) + shown = self._json("flow") + self.assertEqual( + (shown["worktree"], shown["git_dir"], shown["branch"]), + (self.repo, os.path.join(self.repo, ".git"), "main"), + ) + self.assertTrue( + os.path.isfile(os.path.join(self.repo, ".git", "pkgmgr", "flow.yml")) + ) + self.assertEqual( + self._cli("flow", "clear")[:2], (0, "pull: (none)\npush: (none)\n") + ) + + def test_show_without_a_flow_succeeds_empty(self) -> None: + shown = self._json("flow", self.merged_wt) + self.assertEqual( + (shown["source"], shown["flow_branch"], shown["pull"], shown["push"]), + ("none", None, [], []), + ) + self.assertEqual(shown["warnings"], []) + + def test_invalid_names_exit_2_and_write_nothing(self) -> None: + escape = os.path.join(self.root, "escape") + for bad in ("-x", "a..b", f"--output={escape}"): + for flag, other in (("--pull", "--push"), ("--push", "--pull")): + with self.subTest(bad=bad, flag=flag): + code, out, err = self._cli( + "flow", "set", self.merged_wt, f"{flag}={bad}", f"{other}=main" + ) + self.assertEqual((code, out), (2, "")) + self.assertIn("is not a valid branch name", err) + self.assertFalse(os.path.exists(os.path.join(self.linked_git_dir, "pkgmgr"))) + self.assertFalse(os.path.exists(escape)) + + def test_usage_errors_exit_2_and_write_nothing(self) -> None: + cases = ( + ("flow", "set", self.merged_wt, "--pull", "main"), + ("flow", "set", self.merged_wt, "--pull", "main", "--no-pull", "--no-push"), + ("flow", self.merged_wt, "--push", "main"), + ("flow", "clear", self.merged_wt, "--no-pull"), + ("flow", "show", self.merged_wt), + ) + for argv in cases: + with self.subTest(argv=argv[1:]): + self.assertEqual(self._cli(*argv)[:2], (2, "")) + self.assertFalse(os.path.exists(os.path.join(self.linked_git_dir, "pkgmgr"))) + + def test_a_path_that_is_no_worktree_exits_2(self) -> None: + absent = os.path.join(self.root, "absent") + for path in (self.root, self.origin, absent): + for argv in ( + ("flow", path), + ("flow", "set", path, "--no-pull", "--no-push"), + ): + with self.subTest(argv=argv): + code, out, err = self._cli(*argv) + self.assertEqual((code, out), (2, "")) + self.assertIn("is not a git worktree", err) + self.assertFalse(os.path.exists(os.path.join(self.origin, "pkgmgr"))) + + def test_another_branch_warns_of_the_mismatch(self) -> None: + self._json("flow", "set", self.merged_wt, "--pull", "main", "--push", "main") + git(self.merged_wt, "checkout", "--quiet", "-b", "renamed") + + shown = self._json("flow", self.merged_wt) + self.assertEqual( + (shown["branch"], shown["flow_branch"], shown["source"], shown["pull"]), + ("renamed", "merged", "file", ["main"]), + ) + warning = "flow was set on merged, the worktree is on renamed" + self.assertEqual(shown["warnings"], [warning]) + self.assertEqual( + self._cli("flow", self.merged_wt)[1], + f"WARN {warning}\npull: main\npush: main\n", + ) + + def test_a_planted_symlink_reads_invalid_and_set_replaces_it(self) -> None: + target = os.path.join(self.root, "planted.yml") + with open(target, "w", encoding="utf-8") as fh: + yaml.safe_dump({"version": 1, "pull": ["planted"], "push": []}, fh) + os.makedirs(os.path.dirname(self.linked_file)) + os.symlink(target, self.linked_file) + + shown = self._json("flow", self.merged_wt) + self.assertEqual( + (shown["source"], shown["pull"], shown["push"]), ("invalid", [], []) + ) + self.assertTrue(shown["warnings"][0].startswith("pkgmgr/flow.yml ignored: ")) + self._json("flow", "set", self.merged_wt, "--pull", "main", "--no-push") + self.assertFalse(os.path.islink(self.linked_file)) + with open(target, encoding="utf-8") as fh: + self.assertEqual(yaml.safe_load(fh)["pull"], ["planted"]) + + +class TestUpFlow(FlowFixture, unittest.TestCase): + def _up( + self, + branch: str, + pull: list[str] | None, + push: list[str] | None, + base: str | None = None, + ) -> tuple[int, dict]: + out = io.StringIO() + with redirect_stdout(out), redirect_stderr(io.StringIO()): + code = up.worktree_up( + self.repo, + branch, + base=base, + worktrees_dir=os.path.join(self.root, "wts"), + path=None, + hooks={}, + as_json=True, + pull=pull, + push=push, + ) + return code, json.loads(out.getvalue()) if code != 2 else {} + + def _shown(self, path: str) -> tuple: + shown = self._json("flow", path) + return shown["source"], shown["flow_branch"], shown["pull"], shown["push"] + + def test_without_lists_the_flow_is_the_base(self) -> None: + code, result = self._up("feature/d", None, None) + + self.assertEqual(code, 0) + self.assertEqual(result["flow"], {"pull": ["origin/main"], "push": ["main"]}) + self.assertEqual( + self._shown(result["path"]), + ("file", "feature/d", ["origin/main"], ["main"]), + ) + + def test_given_lists_replace_the_base_each_on_its_own(self) -> None: + _, both = self._up("feature/b", PULL, PUSH, base="origin/main") + _, one = self._up("feature/o", None, ["production"]) + + self.assertEqual(self._shown(both["path"]), ("file", "feature/b", PULL, PUSH)) + self.assertEqual(one["flow"], {"pull": ["origin/main"], "push": ["production"]}) + + def test_a_base_that_is_no_branch_name_sets_no_flow(self) -> None: + code, result = self._up("feature/c", None, None, base="main~0") + + self.assertEqual((code, result["flow"]), (0, None)) + self.assertIn("flow not set: base 'main~0'", " ".join(result["warnings"])) + self.assertEqual(self._shown(result["path"])[0], "none") + + def test_invalid_names_refuse_before_the_checkout(self) -> None: + escape = os.path.join(self.root, "escape") + for pull, push in ( + (["-x"], None), + (None, ["a..b"]), + ([f"--output={escape}"], None), + ): + with self.subTest(pull=pull, push=push): + self.assertEqual(self._up("feature/i", pull, push)[0], 2) + self.assertFalse(os.path.exists(os.path.join(self.root, "wts"))) + self.assertFalse(os.path.exists(escape)) + self.assertEqual(git(self.repo, "branch", "--list", "feature/i"), "") + + def test_up_parses_repeatable_lists(self) -> None: + args = create_parser("test").parse_args( + ["worktree", "up", "feature/x", "--pull", "a", "--pull", "b", "--push", "c"] + ) + self.assertEqual((args.pull, args.push), (["a", "b"], ["c"])) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/pkgmgr/actions/worktree/test_flow.py b/tests/unit/pkgmgr/actions/worktree/test_flow.py new file mode 100644 index 0000000..3f25d36 --- /dev/null +++ b/tests/unit/pkgmgr/actions/worktree/test_flow.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +import os +import stat +import tempfile +import unittest +from unittest.mock import patch + +import yaml + +from pkgmgr.actions.worktree import flow + +PULL = ["origin/main", "store-i18n"] +PUSH = ["feature/iam-profile-rbac", "production"] +PLANTED = {"version": 1, "branch": "x", "pull": ["planted"], "push": ["planted"]} + + +class TestFlowFile(unittest.TestCase): + def setUp(self) -> None: + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.git_dir = os.path.realpath(tmp.name) + self.folder = os.path.join(self.git_dir, flow.FLOW_DIR) + self.file = os.path.join(self.folder, flow.FLOW_FILE) + + def _plant(self, data: bytes) -> None: + os.makedirs(self.folder, exist_ok=True) + with open(self.file, "wb") as fh: + fh.write(data) + + def _outside(self, name: str) -> str: + target = os.path.join(self.git_dir, name) + with open(target, "w", encoding="utf-8") as fh: + yaml.safe_dump(PLANTED, fh) + return target + + def _invalid(self) -> str: + found = flow.read_flow(self.git_dir) + self.assertEqual( + (found.source, found.branch, found.pull, found.push), + ("invalid", None, [], []), + ) + self.assertEqual(len(found.warnings), 1) + self.assertTrue(found.warnings[0].startswith("pkgmgr/flow.yml ignored: ")) + return found.warnings[0] + + def test_round_trip_keeps_order_modes_and_schema(self) -> None: + flow.write_flow(self.git_dir, "feature/x", PULL, PUSH) + + found = flow.read_flow(self.git_dir) + self.assertEqual( + (found.source, found.branch, found.pull, found.push, found.warnings), + ("file", "feature/x", PULL, PUSH, []), + ) + self.assertEqual(stat.S_IMODE(os.stat(self.folder).st_mode), 0o700) + self.assertEqual(stat.S_IMODE(os.stat(self.file).st_mode), 0o600) + self.assertEqual(os.listdir(self.folder), [flow.FLOW_FILE]) + with open(self.file, encoding="utf-8") as fh: + self.assertEqual( + yaml.safe_load(fh), + {"version": 1, "branch": "feature/x", "pull": PULL, "push": PUSH}, + ) + + def test_missing_file_reads_as_none(self) -> None: + self.assertEqual(flow.read_flow(self.git_dir), flow.Flow()) + os.makedirs(self.folder) + self.assertEqual(flow.read_flow(self.git_dir).source, "none") + + def test_symlinked_file_is_never_followed(self) -> None: + os.makedirs(self.folder) + os.symlink(self._outside("planted.yml"), self.file) + + self._invalid() + + def test_symlinked_folder_is_never_followed_or_written(self) -> None: + real = os.path.join(self.git_dir, "real") + os.makedirs(real) + os.rename(self._outside("planted.yml"), os.path.join(real, flow.FLOW_FILE)) + os.symlink(real, self.folder) + + self._invalid() + with self.assertRaises(OSError): + flow.write_flow(self.git_dir, None, [], []) + with self.assertRaises(OSError): + flow.clear_flow(self.git_dir) + self.assertEqual(os.listdir(real), [flow.FLOW_FILE]) + with open(os.path.join(real, flow.FLOW_FILE), encoding="utf-8") as fh: + self.assertEqual(yaml.safe_load(fh), PLANTED) + + def test_write_and_clear_replace_a_symlink_never_its_target(self) -> None: + target = self._outside("victim.yml") + os.makedirs(self.folder) + os.symlink(target, self.file) + + flow.write_flow(self.git_dir, "x", ["main"], []) + self.assertFalse(os.path.islink(self.file)) + self.assertEqual(flow.read_flow(self.git_dir).pull, ["main"]) + os.unlink(self.file) + os.symlink(target, self.file) + flow.clear_flow(self.git_dir) + + self.assertFalse(os.path.lexists(self.file)) + with open(target, encoding="utf-8") as fh: + self.assertEqual(yaml.safe_load(fh), PLANTED) + + def test_clear_without_a_file_is_a_no_op(self) -> None: + flow.clear_flow(self.git_dir) + flow.write_flow(self.git_dir, "x", PULL, PUSH) + flow.clear_flow(self.git_dir) + flow.clear_flow(self.git_dir) + self.assertEqual(flow.read_flow(self.git_dir), flow.Flow()) + + def test_oversized_file_reads_as_invalid(self) -> None: + self._plant(b"version: 1\npull: []\npush: []\n" + b"#" * flow.MAX_BYTES) + self.assertIn("larger than", self._invalid()) + + def test_fifo_reads_as_invalid_without_blocking(self) -> None: + os.makedirs(self.folder) + os.mkfifo(self.file) + self.assertIn("not a regular file", self._invalid()) + + def test_directory_reads_as_invalid(self) -> None: + os.makedirs(self.file) + self.assertIn("not a regular file", self._invalid()) + + def test_malformed_files_read_as_invalid(self) -> None: + lists = "pull: []\npush: []\n" + cases = { + "yaml": b"pull: [\n", + "not a mapping": b"- main\n", + "unknown key": f"version: 1\n{lists}extra: 1\n".encode(), + "missing key": b"version: 1\npull: []\n", + "bool version": f"version: true\n{lists}".encode(), + "other version": f"version: 2\n{lists}".encode(), + "string list": b"version: 1\npull: main\npush: []\n", + "number entry": b"version: 1\npull: [1]\npush: []\n", + "option entry": b"version: 1\npull: ['--output=/tmp/x']\npush: []\n", + "range entry": b"version: 1\npull: []\npush: [a..b]\n", + "bad branch": f"version: 1\nbranch: -x\n{lists}".encode(), + "nesting": b"[" * 30000, + "binary": b"\xff\xfe", + } + for name, data in cases.items(): + with self.subTest(case=name): + self._plant(data) + self._invalid() + + +class TestNames(unittest.TestCase): + def test_branch_and_remote_tracking_names_pass_in_order(self) -> None: + self.assertEqual(flow.check_names("pull", PULL + ["@"]), PULL + ["@"]) + + def test_invalid_names_are_refused(self) -> None: + for name in ("-x", "a..b", "--output=/tmp/x", "", "HEAD", "a b", "x\0y", 1): + with self.subTest(name=name), self.assertRaises(flow.FlowError): + flow.check_name("pull", name) + with self.assertRaises(flow.FlowError): + flow.check_names("push", "main") + + def test_option_like_names_never_reach_git(self) -> None: + with patch.object(flow, "run") as run, self.assertRaises(flow.FlowError): + flow.check_name("pull", "--output=/tmp/x") + run.assert_not_called() + + def test_mismatch_warns_only_for_a_file_of_another_branch(self) -> None: + found = flow.Flow(branch="a", source="file") + self.assertEqual(flow.mismatch(found, "a"), []) + self.assertEqual( + flow.mismatch(found, None), + ["flow was set on a, the worktree is on a detached HEAD"], + ) + self.assertEqual(flow.mismatch(flow.Flow(source="invalid"), "a"), []) + + +if __name__ == "__main__": + unittest.main()