Bootstrap project and safe work-item workflow
This commit is contained in:
5
work_item/__init__.py
Normal file
5
work_item/__init__.py
Normal file
@@ -0,0 +1,5 @@
|
||||
"""Safe Gitea-backed work-item workflow helpers."""
|
||||
|
||||
from .core import WorkItemError, Workflow
|
||||
|
||||
__all__ = ["WorkItemError", "Workflow"]
|
||||
3
work_item/__main__.py
Normal file
3
work_item/__main__.py
Normal file
@@ -0,0 +1,3 @@
|
||||
from .core import main
|
||||
|
||||
raise SystemExit(main())
|
||||
547
work_item/core.py
Normal file
547
work_item/core.py
Normal file
@@ -0,0 +1,547 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import fnmatch
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shlex
|
||||
import subprocess
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterable, Sequence
|
||||
|
||||
try:
|
||||
import tomllib
|
||||
except ModuleNotFoundError: # pragma: no cover - Python < 3.11 is unsupported
|
||||
tomllib = None
|
||||
|
||||
|
||||
STORY_RE = re.compile(r"^(US(?P<epic>\d{2})-(?P<story>\d{2}))\s+—\s+(?P<title>.+)$")
|
||||
SECRET_PATTERNS = (
|
||||
re.compile(r"(?i)(api[_-]?key|access[_-]?token|client[_-]?secret|password)\s*[:=]\s*['\"]?[A-Za-z0-9_./+\-=]{12,}"),
|
||||
re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
|
||||
re.compile(r"(?i)authorization:\s*(?:bearer|token)\s+[A-Za-z0-9._~+\-/=]{12,}"),
|
||||
)
|
||||
|
||||
|
||||
class WorkItemError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
class Runner:
|
||||
def run(
|
||||
self,
|
||||
args: Sequence[str],
|
||||
*,
|
||||
cwd: Path,
|
||||
check: bool = True,
|
||||
input_text: str | None = None,
|
||||
) -> subprocess.CompletedProcess[str]:
|
||||
result = subprocess.run(
|
||||
list(args),
|
||||
cwd=cwd,
|
||||
text=True,
|
||||
input=input_text,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
)
|
||||
if check and result.returncode:
|
||||
command = " ".join(shlex.quote(part) for part in args)
|
||||
detail = (result.stderr or result.stdout).strip()
|
||||
raise WorkItemError(f"Command failed ({result.returncode}): {command}\n{detail}")
|
||||
return result
|
||||
|
||||
def run_shell(self, command: str, *, cwd: Path) -> subprocess.CompletedProcess[str]:
|
||||
result = subprocess.run(command, cwd=cwd, text=True, shell=True)
|
||||
if result.returncode:
|
||||
raise WorkItemError(f"Test command failed ({result.returncode}): {command}")
|
||||
return result
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Config:
|
||||
repo_slug: str
|
||||
login: str
|
||||
assignee: str
|
||||
remote: str = "origin"
|
||||
main_branch: str = "main"
|
||||
branch_prefix: str = "us"
|
||||
require_ci: bool = False
|
||||
max_file_bytes: int = 5_000_000
|
||||
required_tests: tuple[str, ...] = ()
|
||||
denied_patterns: tuple[str, ...] = ()
|
||||
allowed_patterns: tuple[str, ...] = ()
|
||||
|
||||
@classmethod
|
||||
def load(cls, path: Path) -> "Config":
|
||||
if tomllib is None:
|
||||
raise WorkItemError("Python 3.11 or newer is required")
|
||||
if not path.is_file():
|
||||
raise WorkItemError(f"Missing workflow configuration: {path}")
|
||||
with path.open("rb") as handle:
|
||||
raw = tomllib.load(handle)
|
||||
repo = raw.get("repository", {})
|
||||
workflow = raw.get("workflow", {})
|
||||
safety = raw.get("safety", {})
|
||||
required = ("slug", "login", "assignee")
|
||||
missing = [key for key in required if not repo.get(key)]
|
||||
if missing:
|
||||
raise WorkItemError(f"Missing repository configuration: {', '.join(missing)}")
|
||||
return cls(
|
||||
repo_slug=repo["slug"],
|
||||
login=repo["login"],
|
||||
assignee=repo["assignee"],
|
||||
remote=repo.get("remote", "origin"),
|
||||
main_branch=repo.get("main_branch", "main"),
|
||||
branch_prefix=workflow.get("branch_prefix", "us"),
|
||||
require_ci=bool(workflow.get("require_ci", False)),
|
||||
max_file_bytes=int(safety.get("max_file_bytes", 5_000_000)),
|
||||
required_tests=tuple(workflow.get("required_tests", ())),
|
||||
denied_patterns=tuple(safety.get("deny", ())),
|
||||
allowed_patterns=tuple(safety.get("allow", ())),
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Story:
|
||||
number: int
|
||||
story_id: str
|
||||
title: str
|
||||
epic: int
|
||||
sequence: int
|
||||
labels: frozenset[str]
|
||||
assignees: tuple[str, ...]
|
||||
raw: dict[str, Any]
|
||||
|
||||
@classmethod
|
||||
def from_issue(cls, issue: dict[str, Any]) -> "Story | None":
|
||||
match = STORY_RE.match(issue.get("title", ""))
|
||||
if not match:
|
||||
return None
|
||||
labels = frozenset(label["name"] for label in issue.get("labels") or ())
|
||||
assignees = tuple(user["login"] for user in issue.get("assignees") or ())
|
||||
return cls(
|
||||
number=int(issue["number"]),
|
||||
story_id=match.group(1),
|
||||
title=match.group("title"),
|
||||
epic=int(match.group("epic")),
|
||||
sequence=int(match.group("story")),
|
||||
labels=labels,
|
||||
assignees=assignees,
|
||||
raw=issue,
|
||||
)
|
||||
|
||||
@property
|
||||
def branch_slug(self) -> str:
|
||||
value = re.sub(r"[^a-z0-9]+", "-", self.title.lower()).strip("-")
|
||||
return value[:48].rstrip("-")
|
||||
|
||||
|
||||
class GitRepo:
|
||||
def __init__(self, root: Path, config: Config, runner: Runner) -> None:
|
||||
self.root = root
|
||||
self.config = config
|
||||
self.runner = runner
|
||||
|
||||
def git(self, *args: str, check: bool = True) -> str:
|
||||
return self.runner.run(("git", *args), cwd=self.root, check=check).stdout.strip()
|
||||
|
||||
def verify(self) -> None:
|
||||
actual = Path(self.git("rev-parse", "--show-toplevel")).resolve()
|
||||
if actual != self.root.resolve():
|
||||
raise WorkItemError(f"Run from repository root {actual}; current root is {self.root}")
|
||||
remote_url = self.git("remote", "get-url", self.config.remote)
|
||||
normalized = remote_url.removesuffix(".git").replace(":", "/")
|
||||
if self.config.repo_slug not in normalized:
|
||||
raise WorkItemError(
|
||||
f"Remote {self.config.remote!r} does not match {self.config.repo_slug!r}: {remote_url}"
|
||||
)
|
||||
|
||||
def ensure_clean(self) -> None:
|
||||
if self.git("status", "--porcelain=v1"):
|
||||
raise WorkItemError("Working tree must be clean before claiming a story")
|
||||
|
||||
def update_main(self) -> None:
|
||||
self.git("fetch", "--prune", self.config.remote)
|
||||
self.git("switch", self.config.main_branch)
|
||||
self.git("pull", "--ff-only", self.config.remote, self.config.main_branch)
|
||||
|
||||
def current_branch(self) -> str:
|
||||
return self.git("branch", "--show-current")
|
||||
|
||||
def create_story_branch(self, story: Story) -> str:
|
||||
branch = f"{self.config.branch_prefix}/{story.story_id}-{story.branch_slug}"
|
||||
self.git("switch", "-c", branch)
|
||||
return branch
|
||||
|
||||
def changed_paths(self) -> list[Path]:
|
||||
output = self.git("status", "--porcelain=v1", "--untracked-files=all", "-z")
|
||||
if not output:
|
||||
return []
|
||||
entries = output.split("\0")
|
||||
paths: list[Path] = []
|
||||
skip_rename_source = False
|
||||
for entry in entries:
|
||||
if not entry:
|
||||
continue
|
||||
if skip_rename_source:
|
||||
skip_rename_source = False
|
||||
continue
|
||||
if len(entry) < 4:
|
||||
continue
|
||||
status, name = entry[:2], entry[3:]
|
||||
paths.append(Path(name))
|
||||
if "R" in status or "C" in status:
|
||||
skip_rename_source = True
|
||||
return paths
|
||||
|
||||
def assert_safe_changes(self) -> list[Path]:
|
||||
paths = self.changed_paths()
|
||||
if not paths:
|
||||
raise WorkItemError("No changes to submit")
|
||||
violations: list[str] = []
|
||||
for relative in paths:
|
||||
portable = relative.as_posix()
|
||||
allowed = any(fnmatch.fnmatch(portable, pattern) for pattern in self.config.allowed_patterns)
|
||||
denied = any(fnmatch.fnmatch(portable, pattern) for pattern in self.config.denied_patterns)
|
||||
if denied and not allowed:
|
||||
violations.append(f"denied path: {portable}")
|
||||
continue
|
||||
candidate = self.root / relative
|
||||
if candidate.is_file() and candidate.stat().st_size > self.config.max_file_bytes:
|
||||
violations.append(f"file exceeds {self.config.max_file_bytes} bytes: {portable}")
|
||||
continue
|
||||
if candidate.is_file() and candidate.stat().st_size <= self.config.max_file_bytes:
|
||||
try:
|
||||
content = candidate.read_text(encoding="utf-8")
|
||||
except (UnicodeDecodeError, OSError):
|
||||
content = ""
|
||||
for pattern in SECRET_PATTERNS:
|
||||
if pattern.search(content):
|
||||
violations.append(f"possible secret in: {portable}")
|
||||
break
|
||||
if violations:
|
||||
raise WorkItemError("Unsafe changes detected:\n- " + "\n- ".join(violations))
|
||||
self.git("diff", "--check")
|
||||
return paths
|
||||
|
||||
def stage_commit_push(self, story: Story, branch: str) -> str:
|
||||
self.git("add", "--all")
|
||||
staged = self.git("diff", "--cached", "--name-only")
|
||||
if not staged:
|
||||
raise WorkItemError("Nothing was staged")
|
||||
subject = f"{story.story_id}: {story.title}"
|
||||
self.git("commit", "-m", subject)
|
||||
self.git("push", "--set-upstream", self.config.remote, branch)
|
||||
return self.git("rev-parse", "HEAD")
|
||||
|
||||
|
||||
class Gitea:
|
||||
def __init__(self, root: Path, config: Config, runner: Runner) -> None:
|
||||
self.root = root
|
||||
self.config = config
|
||||
self.runner = runner
|
||||
|
||||
def tea(self, *args: str) -> str:
|
||||
command = ("tea", *args, "--login", self.config.login)
|
||||
return self.runner.run(command, cwd=self.root).stdout.strip()
|
||||
|
||||
def api(self, endpoint: str, *, method: str = "GET", data: Any = None) -> Any:
|
||||
args = ["api", endpoint, "--method", method]
|
||||
if data is not None:
|
||||
args.extend(("--data", json.dumps(data)))
|
||||
output = self.tea(*args)
|
||||
return json.loads(output) if output else None
|
||||
|
||||
@property
|
||||
def base(self) -> str:
|
||||
return f"/repos/{self.config.repo_slug}"
|
||||
|
||||
def open_stories(self) -> list[Story]:
|
||||
issues = self.api(f"{self.base}/issues?state=open&type=issues&limit=100")
|
||||
stories = [story for issue in issues if (story := Story.from_issue(issue))]
|
||||
return sorted(stories, key=lambda story: (story.epic, story.sequence, story.number))
|
||||
|
||||
def dependencies_closed(self, issue_number: int) -> bool:
|
||||
dependencies = self.api(f"{self.base}/issues/{issue_number}/dependencies?limit=100")
|
||||
return all(issue.get("state") == "closed" for issue in dependencies)
|
||||
|
||||
def next_story(self) -> Story:
|
||||
excluded = {"status/in-progress", "status/review", "status/blocked", "status/done"}
|
||||
for story in self.open_stories():
|
||||
if story.labels & excluded or story.assignees:
|
||||
continue
|
||||
if self.dependencies_closed(story.number):
|
||||
return story
|
||||
raise WorkItemError("No eligible open user story is available")
|
||||
|
||||
def edit_labels(self, issue: int, *, add: Iterable[str], remove: Iterable[str]) -> None:
|
||||
args = ["issues", "edit", str(issue), "--repo", self.config.repo_slug]
|
||||
add_value = ",".join(add)
|
||||
remove_value = ",".join(remove)
|
||||
if add_value:
|
||||
args.extend(("--add-labels", add_value))
|
||||
if remove_value:
|
||||
args.extend(("--remove-labels", remove_value))
|
||||
self.tea(*args)
|
||||
|
||||
def claim(self, story: Story) -> None:
|
||||
self.tea(
|
||||
"issues", "edit", str(story.number),
|
||||
"--repo", self.config.repo_slug,
|
||||
"--add-assignees", self.config.assignee,
|
||||
"--add-labels", "status/in-progress",
|
||||
"--remove-labels", "status/backlog,status/ready,status/blocked,status/review,status/done",
|
||||
)
|
||||
|
||||
def comment(self, issue: int, body: str) -> None:
|
||||
self.tea("comments", "add", str(issue), body, "--repo", self.config.repo_slug)
|
||||
|
||||
def create_pr(self, story: Story, branch: str, tests: Sequence[str], commit: str) -> dict[str, Any]:
|
||||
body = "\n".join((
|
||||
f"Implements #{story.number}",
|
||||
"",
|
||||
"Tests:",
|
||||
*(f"- `{command}`" for command in tests),
|
||||
"",
|
||||
f"Commit: `{commit}`",
|
||||
))
|
||||
output = self.tea(
|
||||
"pulls", "create",
|
||||
"--repo", self.config.repo_slug,
|
||||
"--head", branch,
|
||||
"--base", self.config.main_branch,
|
||||
"--title", f"{story.story_id}: {story.title}",
|
||||
"--description", body,
|
||||
"--output", "json",
|
||||
)
|
||||
try:
|
||||
return json.loads(output)
|
||||
except json.JSONDecodeError as error:
|
||||
raise WorkItemError(f"Could not parse created pull request: {output}") from error
|
||||
|
||||
def pr(self, number: int) -> dict[str, Any]:
|
||||
return self.api(f"{self.base}/pulls/{number}")
|
||||
|
||||
def merge_pr(self, number: int) -> None:
|
||||
self.tea("pulls", "merge", str(number), "--repo", self.config.repo_slug, "--style", "squash")
|
||||
|
||||
def require_ci_success(self, pr: dict[str, Any]) -> None:
|
||||
if not self.config.require_ci:
|
||||
return
|
||||
sha = pr["head"]["sha"]
|
||||
combined = self.api(f"{self.base}/commits/{sha}/status")
|
||||
if combined.get("state") != "success":
|
||||
raise WorkItemError(f"Required CI is not successful: {combined.get('state', 'unknown')}")
|
||||
|
||||
def mark_review(self, issue: int, pr: dict[str, Any], tests: Sequence[str]) -> None:
|
||||
self.edit_labels(
|
||||
issue,
|
||||
add=("status/review",),
|
||||
remove=("status/in-progress", "status/backlog", "status/ready", "status/blocked", "status/done"),
|
||||
)
|
||||
number = pr.get("index") or pr.get("number")
|
||||
url = pr.get("url") or pr.get("html_url") or f"PR #{number}"
|
||||
self.comment(issue, f"Submitted for review: {url}\n\nTests passed:\n" + "\n".join(f"- `{x}`" for x in tests))
|
||||
|
||||
def mark_done(self, issue: int, pr_number: int) -> None:
|
||||
self.edit_labels(
|
||||
issue,
|
||||
add=("status/done",),
|
||||
remove=("status/in-progress", "status/backlog", "status/ready", "status/blocked", "status/review"),
|
||||
)
|
||||
self.comment(issue, f"Completed and merged via PR #{pr_number}.")
|
||||
self.tea("issues", "close", str(issue), "--repo", self.config.repo_slug)
|
||||
|
||||
|
||||
class Workflow:
|
||||
STATE_NAME = "work-item-state.json"
|
||||
|
||||
def __init__(self, root: Path, config: Config, runner: Runner | None = None) -> None:
|
||||
self.root = root.resolve()
|
||||
self.config = config
|
||||
self.runner = runner or Runner()
|
||||
self.git = GitRepo(self.root, config, self.runner)
|
||||
self.gitea = Gitea(self.root, config, self.runner)
|
||||
|
||||
@property
|
||||
def state_path(self) -> Path:
|
||||
git_dir = self.git.git("rev-parse", "--git-dir")
|
||||
path = Path(git_dir)
|
||||
if not path.is_absolute():
|
||||
path = self.root / path
|
||||
return path / self.STATE_NAME
|
||||
|
||||
def load_state(self) -> dict[str, Any]:
|
||||
if not self.state_path.is_file():
|
||||
raise WorkItemError("No story is currently claimed")
|
||||
return json.loads(self.state_path.read_text(encoding="utf-8"))
|
||||
|
||||
def save_state(self, state: dict[str, Any]) -> None:
|
||||
temporary = self.state_path.with_suffix(".tmp")
|
||||
temporary.write_text(json.dumps(state, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
||||
temporary.replace(self.state_path)
|
||||
|
||||
def clear_state(self) -> None:
|
||||
self.state_path.unlink(missing_ok=True)
|
||||
|
||||
def next(self) -> Story:
|
||||
self.git.verify()
|
||||
return self.gitea.next_story()
|
||||
|
||||
def claim(self) -> dict[str, Any]:
|
||||
self.git.verify()
|
||||
self.git.ensure_clean()
|
||||
if self.state_path.exists():
|
||||
raise WorkItemError("A story is already claimed; complete or block it first")
|
||||
self.git.update_main()
|
||||
story = self.gitea.next_story()
|
||||
branch = self.git.create_story_branch(story)
|
||||
self.gitea.claim(story)
|
||||
state = {
|
||||
"issue": story.number,
|
||||
"story_id": story.story_id,
|
||||
"title": story.title,
|
||||
"branch": branch,
|
||||
"status": "in-progress",
|
||||
}
|
||||
self.save_state(state)
|
||||
self.gitea.comment(story.number, f"Claimed for implementation on branch `{branch}`.")
|
||||
return state
|
||||
|
||||
def status(self) -> dict[str, Any]:
|
||||
self.git.verify()
|
||||
state = self.load_state()
|
||||
return {**state, "current_branch": self.git.current_branch(), "changes": [p.as_posix() for p in self.git.changed_paths()]}
|
||||
|
||||
def submit(self, tests: Sequence[str], *, confirmed: bool) -> dict[str, Any]:
|
||||
self.git.verify()
|
||||
state = self.load_state()
|
||||
if self.git.current_branch() != state["branch"]:
|
||||
raise WorkItemError(f"Expected branch {state['branch']}, found {self.git.current_branch()}")
|
||||
commands = tuple(dict.fromkeys((*self.config.required_tests, *tests)))
|
||||
if not commands:
|
||||
raise WorkItemError("At least one automated test command is required")
|
||||
self.git.assert_safe_changes()
|
||||
for command in commands:
|
||||
self.runner.run_shell(command, cwd=self.root)
|
||||
paths = self.git.assert_safe_changes()
|
||||
if not confirmed:
|
||||
summary = self.git.git("diff", "--stat")
|
||||
raise WorkItemError(
|
||||
"Review the changes and rerun with --yes to commit:\n"
|
||||
f"{summary}\nFiles: {', '.join(path.as_posix() for path in paths)}"
|
||||
)
|
||||
story = Story(
|
||||
number=int(state["issue"]), story_id=state["story_id"], title=state["title"],
|
||||
epic=int(state["story_id"][2:4]), sequence=int(state["story_id"][5:7]),
|
||||
labels=frozenset(), assignees=(), raw={},
|
||||
)
|
||||
commit = self.git.stage_commit_push(story, state["branch"])
|
||||
pr = self.gitea.create_pr(story, state["branch"], commands, commit)
|
||||
pr_number = int(pr.get("index") or pr.get("number"))
|
||||
state.update({"status": "review", "pr": pr_number, "commit": commit, "tests": list(commands)})
|
||||
self.save_state(state)
|
||||
self.gitea.mark_review(story.number, pr, commands)
|
||||
return state
|
||||
|
||||
def complete(self, *, merge: bool) -> dict[str, Any]:
|
||||
self.git.verify()
|
||||
state = self.load_state()
|
||||
if state.get("status") != "review" or not state.get("pr"):
|
||||
raise WorkItemError("The claimed story has not been submitted for review")
|
||||
pr = self.gitea.pr(int(state["pr"]))
|
||||
self.gitea.require_ci_success(pr)
|
||||
if not pr.get("merged"):
|
||||
if not merge:
|
||||
raise WorkItemError("Pull request is not merged; rerun with --merge after review")
|
||||
self.gitea.merge_pr(int(state["pr"]))
|
||||
pr = self.gitea.pr(int(state["pr"]))
|
||||
if not pr.get("merged"):
|
||||
raise WorkItemError("Pull request merge could not be verified")
|
||||
self.gitea.mark_done(int(state["issue"]), int(state["pr"]))
|
||||
branch = state["branch"]
|
||||
self.git.git("switch", self.config.main_branch)
|
||||
self.git.git("pull", "--ff-only", self.config.remote, self.config.main_branch)
|
||||
# A squash-merged PR is verified by Gitea but its feature commit is not an
|
||||
# ancestor of main, so ordinary `-d` would incorrectly reject safe cleanup.
|
||||
self.git.git("branch", "-D", branch)
|
||||
result = {**state, "status": "done"}
|
||||
self.clear_state()
|
||||
return result
|
||||
|
||||
def block(self, reason: str) -> dict[str, Any]:
|
||||
self.git.verify()
|
||||
if not reason.strip():
|
||||
raise WorkItemError("A non-empty blocking reason is required")
|
||||
state = self.load_state()
|
||||
self.gitea.edit_labels(
|
||||
int(state["issue"]),
|
||||
add=("status/blocked",),
|
||||
remove=("status/in-progress", "status/ready", "status/review", "status/done"),
|
||||
)
|
||||
self.gitea.comment(int(state["issue"]), f"Blocked: {reason.strip()}")
|
||||
state.update({"status": "blocked", "reason": reason.strip()})
|
||||
self.save_state(state)
|
||||
return state
|
||||
|
||||
|
||||
def find_root(start: Path) -> Path:
|
||||
result = subprocess.run(
|
||||
("git", "rev-parse", "--show-toplevel"), cwd=start, text=True,
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||||
)
|
||||
if result.returncode:
|
||||
raise WorkItemError("This command must run inside the photoanalyzer Git repository")
|
||||
return Path(result.stdout.strip()).resolve()
|
||||
|
||||
|
||||
def parser() -> argparse.ArgumentParser:
|
||||
result = argparse.ArgumentParser(prog="work-item", description="Safe issue/Git workflow helper")
|
||||
result.add_argument("--config", default=".work-item.toml")
|
||||
commands = result.add_subparsers(dest="command", required=True)
|
||||
commands.add_parser("next", help="Show the top eligible user story")
|
||||
commands.add_parser("claim", help="Claim the top story and create its branch")
|
||||
commands.add_parser("status", help="Show the current claim and Git state")
|
||||
submit = commands.add_parser("submit", help="Test, commit, push, and open a pull request")
|
||||
submit.add_argument("--test", action="append", default=[], help="Additional automated test command; repeatable")
|
||||
submit.add_argument("--yes", action="store_true", help="Confirm the reviewed diff may be committed")
|
||||
complete = commands.add_parser("complete", help="Verify merge, close the issue, and return to main")
|
||||
complete.add_argument("--merge", action="store_true", help="Merge the reviewed PR before completing")
|
||||
block = commands.add_parser("block", help="Mark the current story blocked")
|
||||
block.add_argument("--reason", required=True)
|
||||
return result
|
||||
|
||||
|
||||
def main(argv: Sequence[str] | None = None) -> int:
|
||||
args = parser().parse_args(argv)
|
||||
try:
|
||||
root = find_root(Path.cwd())
|
||||
config_path = Path(args.config)
|
||||
if not config_path.is_absolute():
|
||||
config_path = root / config_path
|
||||
workflow = Workflow(root, Config.load(config_path))
|
||||
if args.command == "next":
|
||||
story = workflow.next()
|
||||
payload: Any = {"issue": story.number, "story_id": story.story_id, "title": story.title}
|
||||
elif args.command == "claim":
|
||||
payload = workflow.claim()
|
||||
elif args.command == "status":
|
||||
payload = workflow.status()
|
||||
elif args.command == "submit":
|
||||
payload = workflow.submit(args.test, confirmed=args.yes)
|
||||
elif args.command == "complete":
|
||||
payload = workflow.complete(merge=args.merge)
|
||||
else:
|
||||
payload = workflow.block(args.reason)
|
||||
print(json.dumps(payload, indent=2, sort_keys=True))
|
||||
return 0
|
||||
except WorkItemError as error:
|
||||
print(f"work-item: {error}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user