diff --git a/README.md b/README.md index dba0ae5..ddbed79 100644 --- a/README.md +++ b/README.md @@ -9,3 +9,44 @@ Want to know more? Go to the [project website](https://thenullpigeons.org). How to install Nihil: [Get started / install](https://thenullpigeons.org/docs/installation/linux). The project documentation (images, CLI, usage) is available on the [documentation site](https://thenullpigeons.org/docs). + +## Customize an image + +The following command asks for GitHub CLI authentication, creates or reuses a +personal fork of `nihil-images`, and opens the interactive tool selector: + +```bash +nihil image customize web +``` + +In the selector, use the arrow keys (or `j`/`k`) to move. Press `/` to search +by tool, category, or command. Press `v` to enter visual mode, move with +`j`/`k` to select a range, and press `Space` to toggle all selected tools. +Press `Enter` to save, and `q` or `Esc` to cancel. + +To use a group repository, add `--repo owner/repo` or its full GitHub URL. + +Git remotes use SSH by default. Use `--git-protocol https` when HTTPS remotes +are preferred. + +Use `--git-del` to remove and re-clone the existing local source directory. +This does not delete the remote GitHub repository. + +Changes are stored on a `nihil/-custom` branch in the fork. With +`--no-push`, the branch is only prepared locally. Switch the active source with: + +```bash +nihil image switch personal +nihil image switch upstream +nihil image status +``` + +After pushing, trigger the fork's build workflow for a specific variant and +install the image published in its GHCR namespace: + +```bash +nihil image build full --wait +nihil install web +``` + +Use `nihil image build` to build all variants. diff --git a/nihil/cli/controller.py b/nihil/cli/controller.py index ff2db7c..44b32af 100644 --- a/nihil/cli/controller.py +++ b/nihil/cli/controller.py @@ -4,6 +4,7 @@ import os import secrets +import subprocess import sys import time from pathlib import Path @@ -54,8 +55,11 @@ def run(self, args: Optional[list] = None) -> int: return self._cmd_config(parsed_args) if parsed_args.command == "resources": return self._cmd_resources(parsed_args) + if parsed_args.command == "image": + return self._cmd_image(parsed_args) try: self.manager = NihilManager() + self._configure_image_registry() except NihilError as e: print(self.formatter.error(str(e)), file=sys.stderr) return e.exit_code @@ -87,6 +91,35 @@ def run(self, args: Optional[list] = None) -> int: return self._cmd_completion(parsed_args) return 0 + def _configure_image_registry(self) -> None: + """Point short image references to the currently active fork.""" + if self.config.image_source_active != "personal" or not self.config.personal_image_repo: + return + owner = self.config.personal_image_repo.split("/", 1)[0].lower() + branch = getattr(self.config, "personal_image_branch", "") + image_tag = branch.replace("/", "-") if branch else "latest" + self.manager.AVAILABLE_IMAGES = { + variant: f"ghcr.io/{owner}/{variant}:{image_tag}" + for variant in ("full", "ad", "web", "blueteam") + } + self.manager.DEFAULT_IMAGE = self.manager.AVAILABLE_IMAGES["full"] + + def _login_personal_registry(self) -> None: + """Use the GitHub CLI token for private GHCR packages in the fork.""" + if self.config.image_source_active != "personal" or not self.config.personal_image_repo: + return + owner = self.config.personal_image_repo.split("/", 1)[0] + try: + token = subprocess.run( + ["gh", "auth", "token"], check=True, text=True, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, + ).stdout.strip() + if token: + self.manager.client.login(username=owner, password=token, registry="ghcr.io") + except (FileNotFoundError, subprocess.CalledProcessError, AttributeError): + # Public images remain downloadable without authentication. + pass + def _cmd_start(self, args) -> int: _NOT_CHECKED = object() _update_cache = [_NOT_CHECKED] @@ -843,6 +876,7 @@ def _cmd_install(self, args) -> int: return 1 print(self.formatter.info(f"Pulling image '{image_tag}'...")) try: + self._login_personal_registry() self.manager._pull_with_progress(image_tag) print(self.formatter.success(f"Image '{image_tag}' installed/updated successfully.")) return 0 @@ -1272,6 +1306,175 @@ def _cmd_build(self, args) -> int: # Catalogue partagé : nihil-resources # ------------------------------------------------------------------ + def _cmd_image(self, args) -> int: + from nihil.features.image_sources import ImageSourceError, ImageSourceManager + + manager = ImageSourceManager( + self.config, self.formatter, upstream_repo=getattr(args, "repo", None) + ) + action = getattr(args, "image_action", None) + if action is None: + self.parser.parse_args(["image", "--help"]) + return 0 + + if action == "status": + print(self.formatter.section_header("NIHIL IMAGE SOURCES")) + print(f"Active source: {self.config.image_source_active}") + print(f"Upstream repo: {manager.upstream_repo}") + print(f"Upstream path: {self.config.image_sources_upstream_path}") + print(f"Personal path: {self.config.personal_image_path or '-'}") + print(f"Personal repo: {self.config.personal_image_repo or '-'}") + print(f"Personal branch: {self.config.personal_image_branch or '-'}") + return 0 + + try: + if action == "switch": + path = manager.switch(args.source) + print(self.formatter.success(f"Active image source: {args.source}")) + print(self.formatter.info(f"Source path: {path}")) + return 0 + + if action == "customize": + return self._customize_image(args, manager) + if action == "build": + manager.trigger_build(variant=args.variant or "all", wait=args.wait) + print(self.formatter.success( + f"Docker build workflow dispatched for {manager.config.personal_image_repo}:" + f"{manager.config.personal_image_branch}" + )) + print(self.formatter.info("Use 'nihil install ' when the workflow has finished.")) + return 0 + except ImageSourceError as exc: + print(self.formatter.error(str(exc)), file=sys.stderr) + return 1 + return 1 + + def _customize_image(self, args, source_manager) -> int: + from contextlib import nullcontext + from rich.prompt import Confirm + from nihil.features.image_sources import ImageSourceError + import json + import subprocess + + if not Confirm.ask( + "Create or use your GitHub fork of nihil-images?", + default=True, + ): + print("Aborted.") + return 0 + + console = getattr(self.formatter, "console", None) + loading = ( + console.status("[cyan]Preparing the GitHub fork and image source...[/]", spinner="dots") + if console + else nullcontext() + ) + try: + with loading: + path, fork_repo, branch = source_manager.ensure_personal_fork( + variant=args.variant, + git_protocol=args.git_protocol, + delete_existing=args.git_del, + ) + except ImageSourceError as exc: + print(self.formatter.error(str(exc)), file=sys.stderr) + return 1 + + manifest_path = path / "build" / "config" / "tools.json" + if not manifest_path.is_file(): + print(self.formatter.error(f"Tools manifest not found: {manifest_path}"), file=sys.stderr) + return 1 + try: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + print(self.formatter.error(f"Could not read tools manifest: {exc}"), file=sys.stderr) + return 1 + + selection_path = path / "build" / "config" / "tool-selection.json" + disabled: set[str] = set() + if selection_path.is_file(): + try: + selection = json.loads(selection_path.read_text(encoding="utf-8")) + if "enabled_tools" in selection: + enabled = {str(name) for name in selection.get("enabled_tools", [])} + selected_names = {name.lower() for name in enabled} + disabled = { + tool["name"] for tool in tools + if not tool["mandatory"] and tool["name"].lower() not in selected_names + } + else: + disabled = { + name for name in selection.get("disabled_tools", []) + if str(name).lower() != "nihil-history" + } + except (OSError, json.JSONDecodeError): + disabled = set() + + tools = [] + for category, entries in manifest.items(): + for entry in entries: + tools.append({ + "name": entry["name"], + "cmd": entry.get("cmd") or "-", + "category": category, + "mandatory": category == "core_tools", + }) + tools.sort(key=lambda item: (item["category"], item["name"].lower())) + mandatory_names = {tool["name"].lower() for tool in tools if tool["mandatory"]} + disabled = {name for name in disabled if str(name).lower() not in mandatory_names} + + while True: + selected = self._select_tools_tui(tools, disabled, title=f"{fork_repo}:{branch}") + if selected is None: + print("Tool selection cancelled.") + return 0 + disabled = selected + + enabled = sorted( + tool["name"] for tool in tools + if tool["mandatory"] or tool["name"] not in disabled + ) + selection_path.write_text( + json.dumps({"version": 2, "enabled_tools": enabled}, indent=2) + "\n", + encoding="utf-8", + ) + print(self.formatter.success(f"Saved tool selection: {len(enabled)} enabled")) + + if args.no_push: + print(self.formatter.info(f"Prepared branch {branch} locally at {path}.")) + return 0 + + if Confirm.ask(f"Commit and push {branch} to {fork_repo}?", default=True): + try: + subprocess.run(["git", "add", "build/config/tool-selection.json"], cwd=path, check=True) + subprocess.run( + ["git", "commit", "-m", f"Customize {args.variant} image tools"], + cwd=path, + check=True, + ) + subprocess.run(["git", "push", "--set-upstream", "origin", branch], cwd=path, check=True) + except subprocess.CalledProcessError as exc: + print(self.formatter.error(f"Git operation failed (exit {exc.returncode})."), file=sys.stderr) + return exc.returncode or 1 + print(self.formatter.success(f"Customization pushed to {fork_repo}:{branch}")) + return 0 + + print(self.formatter.info(f"Changes remain local on {branch}: {path}")) + if not Confirm.ask("Relaunch the tool selector?", default=True): + return 0 + + def _select_tools_tui(self, tools: list[dict], disabled: set[str], *, title: str) -> set[str] | None: + """Run the Textual selector and return the disabled tools.""" + try: + from nihil.features.tool_selector import ToolSelectorApp + except ImportError as exc: + print(self.formatter.error(f"Textual is required for the tool selector: {exc}"), file=sys.stderr) + return None + + app = ToolSelectorApp(tools, disabled, title) + app.run() + return app.return_value + def _cmd_resources(self, args) -> int: from nihil.config import NIHIL_RESOURCES_REPO diff --git a/nihil/cli/parser.py b/nihil/cli/parser.py index 191d61f..c81046a 100644 --- a/nihil/cli/parser.py +++ b/nihil/cli/parser.py @@ -35,6 +35,9 @@ def create_parser() -> argparse.ArgumentParser: nihil resources update git pull the local nihil-resources catalog nihil resources sync Fetch tools listed in catalog/resources.toml nihil resources status Show local nihil-resources status + nihil image customize Create a personal nihil-images fork and select tools + nihil image switch upstream Use the upstream nihil-images source + nihil image switch personal Use the personal fork source """ ) @@ -118,6 +121,44 @@ def create_parser() -> argparse.ArgumentParser: resources_sync.add_argument("--profile", default=None, help="Restrict sync to a profile (full|ad|web|blueteam)") resources_subparsers.add_parser("status", help="Show local nihil-resources status (path, branch, last commit)") + image_parser = subparsers.add_parser("image", help="Manage custom nihil-images sources") + image_subparsers = image_parser.add_subparsers(dest="image_action", metavar="ACTION") + customize_parser = image_subparsers.add_parser( + "customize", help="Fork nihil-images, select tools, commit and push a personal branch" + ) + customize_parser.add_argument( + "variant", choices=["full", "ad", "web", "blueteam"], default="full", nargs="?", + help="Image variant whose source branch should be customized (default: full)", + ) + customize_parser.add_argument( + "--no-push", action="store_true", help="Prepare the branch locally without committing or pushing" + ) + customize_parser.add_argument( + "--repo", default=None, metavar="REPO", + help="GitHub repository (owner/repo or URL; default: TheNullPigeons/nihil-images)", + ) + customize_parser.add_argument( + "--git-protocol", choices=["ssh", "https"], default="ssh", + help="Git remote protocol (default: ssh; use https for HTTPS remotes)", + ) + customize_parser.add_argument( + "--git-del", action="store_true", + help="Delete the existing local image source clone before cloning it again", + ) + switch_parser = image_subparsers.add_parser("switch", help="Switch the active image source") + switch_parser.add_argument("source", choices=["upstream", "personal"]) + image_subparsers.add_parser("status", help="Show configured upstream and personal image sources") + build_image_parser = image_subparsers.add_parser( + "build", help="Trigger the Docker image workflow on the active personal branch" + ) + build_image_parser.add_argument( + "variant", choices=["all", "full", "ad", "web", "blueteam"], nargs="?", default=None, + help="Image variant to build (default: all)", + ) + build_image_parser.add_argument( + "--wait", action="store_true", help="Wait until the GitHub Actions build finishes" + ) + completion_parser = subparsers.add_parser("completion", help="Generate shell completion script") completion_parser.add_argument("shell", choices=["bash", "zsh"], help="Target shell for completion script (bash or zsh)") diff --git a/nihil/config/user_config.py b/nihil/config/user_config.py index 55e8a96..d3a195f 100644 --- a/nihil/config/user_config.py +++ b/nihil/config/user_config.py @@ -45,6 +45,15 @@ "build": { "images_path": None, # path to nihil-images source directory }, + "image_sources": { + "home": str(NIHIL_HOME / "image-sources"), + "active": "upstream", + "upstream_repo": "TheNullPigeons/nihil-images", + "upstream_path": str(NIHIL_HOME / "image-sources" / "upstream" / "nihil-images"), + "personal_path": None, + "personal_repo": None, + "personal_branch": None, + }, } _CONFIG_COMMENT = """\ @@ -64,6 +73,9 @@ # display.x11_by_default : enable X11 forwarding by default # updates.auto_check : check for image updates on start # build.images_path : path to nihil-images source directory (for nihil build) +# image_sources.active : upstream | personal +# image_sources.personal_repo : GitHub fork used for customized images +# image_sources.personal_branch: branch used for customized images """ @@ -210,6 +222,59 @@ def images_path(self) -> Optional[Path]: return Path(raw).expanduser().resolve() return None + # ------------------------------------------------------------------ + # Properties: image_sources + # ------------------------------------------------------------------ + + @property + def image_sources_home(self) -> Path: + raw = self._get("image_sources", "home") + return Path(raw).expanduser().resolve() if raw else NIHIL_HOME / "image-sources" + + @property + def image_source_active(self) -> str: + return self._get("image_sources", "active") or "upstream" + + @property + def image_sources_upstream_path(self) -> Path: + raw = self._get("image_sources", "upstream_path") + return Path(raw).expanduser().resolve() if raw else self.image_sources_home / "upstream" / "nihil-images" + + @property + def personal_image_path(self) -> Optional[Path]: + raw = self._get("image_sources", "personal_path") + return Path(raw).expanduser().resolve() if raw else None + + @property + def personal_image_repo(self) -> Optional[str]: + return self._get("image_sources", "personal_repo") + + @property + def personal_image_branch(self) -> Optional[str]: + return self._get("image_sources", "personal_branch") + + def set_image_source( + self, + *, + active: str, + path: Path, + personal_repo: Optional[str], + personal_branch: Optional[str], + upstream_path: Path, + upstream_repo: Optional[str] = None, + ) -> None: + self._data.setdefault("image_sources", {}) + self._data["image_sources"].update({ + "active": active, + "upstream_repo": upstream_repo or self._data["image_sources"].get("upstream_repo"), + "upstream_path": str(upstream_path), + "personal_path": str(path) if active == "personal" else self._data["image_sources"].get("personal_path"), + "personal_repo": personal_repo, + "personal_branch": personal_branch, + }) + self._data.setdefault("build", {})["images_path"] = str(path) + self.save() + # ------------------------------------------------------------------ # Helpers internes # ------------------------------------------------------------------ diff --git a/nihil/features/image_sources.py b/nihil/features/image_sources.py new file mode 100644 index 0000000..edfd8f7 --- /dev/null +++ b/nihil/features/image_sources.py @@ -0,0 +1,203 @@ +#!/usr/bin/env python3 +"""Manage GitHub sources used to build Nihil images.""" + +from __future__ import annotations + +import subprocess +import shutil +from pathlib import Path +from urllib.parse import urlparse + + +UPSTREAM_REPO = "TheNullPigeons/nihil-images" + + +class ImageSourceError(RuntimeError): + """Error related to a local image source or GitHub repository.""" + + +class ImageSourceManager: + """Prepare the upstream repository and a personal nihil-images fork.""" + + def __init__(self, config, formatter=None, upstream_repo: str | None = None): + self.config = config + self.formatter = formatter + self.home = config.image_sources_home + configured_repo = config._get("image_sources", "upstream_repo") if hasattr(config, "_get") else None + self.upstream_repo = self._normalize_repo(upstream_repo or configured_repo or UPSTREAM_REPO) + + @staticmethod + def _normalize_repo(value: str) -> str: + raw = value.strip().rstrip("/") + if raw.startswith(("https://", "http://")): + raw = urlparse(raw).path.strip("/") + if raw.endswith(".git"): + raw = raw[:-4] + if raw.count("/") != 1 or any(not part for part in raw.split("/")): + raise ImageSourceError("The repository must use the owner/repo format or a GitHub URL.") + return raw + + def _run(self, command: list[str], *, cwd: Path | None = None, capture: bool = True) -> str: + try: + result = subprocess.run( + command, + cwd=str(cwd) if cwd else None, + check=True, + text=True, + stdout=subprocess.PIPE if capture else None, + stderr=subprocess.PIPE if capture else None, + ) + except FileNotFoundError as exc: + raise ImageSourceError(f"Command not found: {command[0]}") from exc + except subprocess.CalledProcessError as exc: + output = (exc.stderr or exc.stdout or "").strip() + detail = f": {output}" if output else "" + raise ImageSourceError(f"Command failed: {' '.join(command)}{detail}") from exc + return (result.stdout or "").strip() if capture else "" + + def _gh_user(self) -> str: + return self._run(["gh", "api", "user", "--jq", ".login"]) + + def _default_branch(self, repo: str) -> str: + return self._run([ + "gh", "repo", "view", repo, + "--json", "defaultBranchRef", + "--jq", ".defaultBranchRef.name", + ]) or "main" + + def _enable_actions(self, repo: str) -> None: + """Enable GitHub Actions for the fork so workflow dispatches work.""" + self._run([ + "gh", "api", "--method", "PUT", + f"repos/{repo}/actions/permissions", + "-F", "enabled=true", + "-f", "allowed_actions=all", + ]) + + def _ensure_git_remote(self, path: Path, name: str, url: str) -> None: + remotes = self._run(["git", "remote"], cwd=path).splitlines() + if name in remotes: + current = self._run(["git", "remote", "get-url", name], cwd=path) + if current != url: + self._run(["git", "remote", "set-url", name, url], cwd=path) + else: + self._run(["git", "remote", "add", name, url], cwd=path) + + def ensure_personal_fork( + self, + *, + variant: str, + git_protocol: str = "ssh", + delete_existing: bool = False, + ) -> tuple[Path, str, str]: + """Create or reuse the fork and prepare a customization branch.""" + if git_protocol not in {"ssh", "https"}: + raise ImageSourceError("Git protocol must be 'ssh' or 'https'.") + login = self._gh_user() + repo_name = self.upstream_repo.rsplit("/", 1)[1] + fork_repo = f"{login}/{repo_name}" + try: + self._run(["gh", "repo", "view", fork_repo, "--json", "name"]) + except ImageSourceError: + self._run(["gh", "repo", "fork", self.upstream_repo, "--clone=false"]) + self._enable_actions(fork_repo) + + branch = f"nihil/{variant}-custom" + path = self.home / login / repo_name + path.parent.mkdir(parents=True, exist_ok=True) + if git_protocol == "ssh": + fork_url = f"git@github.com:{fork_repo}.git" + upstream_url = f"git@github.com:{self.upstream_repo}.git" + else: + fork_url = f"https://github.com/{fork_repo}.git" + upstream_url = f"https://github.com/{self.upstream_repo}.git" + + if delete_existing and path.exists(): + shutil.rmtree(path) + + if not (path / ".git").is_dir(): + self._run(["git", "clone", fork_url, str(path)]) + self._ensure_git_remote(path, "origin", fork_url) + self._ensure_git_remote(path, "upstream", upstream_url) + self._run(["git", "fetch", "upstream"], cwd=path) + + default_branch = self._default_branch(self.upstream_repo) + self._run(["git", "fetch", "origin"], cwd=path) + branches = self._run(["git", "branch", "--format=%(refname:short)"], cwd=path).splitlines() + if branch in branches: + self._run(["git", "switch", branch], cwd=path) + else: + self._run(["git", "switch", "-c", branch, f"upstream/{default_branch}"], cwd=path) + + self.config.set_image_source( + active="personal", + path=path, + personal_repo=fork_repo, + personal_branch=branch, + upstream_path=self.config.image_sources_upstream_path, + upstream_repo=self.upstream_repo, + ) + return path, fork_repo, branch + + def ensure_upstream(self) -> Path: + path = self.config.image_sources_upstream_path + path.parent.mkdir(parents=True, exist_ok=True) + upstream_url = f"https://github.com/{self.upstream_repo}.git" + if not (path / ".git").is_dir(): + self._run(["git", "clone", upstream_url, str(path)]) + else: + self._run(["git", "pull", "--ff-only"], cwd=path) + self.config.set_image_source( + active="upstream", + path=path, + personal_repo=self.config.personal_image_repo, + personal_branch=self.config.personal_image_branch, + upstream_path=path, + upstream_repo=self.upstream_repo, + ) + return path + + def trigger_build(self, *, variant: str = "all", wait: bool = False) -> None: + """Trigger the Docker workflow on the active personal branch.""" + if variant not in {"all", "full", "ad", "web", "blueteam"}: + raise ImageSourceError("Unknown image variant. Choose all, full, ad, web, or blueteam.") + repo = self.config.personal_image_repo + branch = self.config.personal_image_branch + if not repo or not branch: + raise ImageSourceError("No personal fork is configured.") + self._enable_actions(repo) + self._run([ + "gh", "workflow", "run", "docker-build.yml", + "--repo", repo, "--ref", branch, + "-f", f"variant={variant}", + ]) + if wait: + run_id = self._run([ + "gh", "run", "list", "--workflow", "docker-build.yml", + "--repo", repo, "--branch", branch, "--limit", "1", + "--json", "databaseId", "--jq", ".[0].databaseId", + ]) + if not run_id: + raise ImageSourceError("The workflow was dispatched, but its run ID could not be found.") + self._run(["gh", "run", "watch", run_id, "--repo", repo, "--exit-status"], capture=False) + + def switch(self, source: str) -> Path: + if source == "personal": + path = self.config.personal_image_path + if not path or not (path / ".git").is_dir(): + raise ImageSourceError("No personal fork is configured. Run 'nihil image customize' first.") + branch = self.config.personal_image_branch + if not branch: + raise ImageSourceError("The personal fork branch is not configured.") + self._run(["git", "switch", branch], cwd=path) + self.config.set_image_source( + active="personal", path=path, + personal_repo=self.config.personal_image_repo, + personal_branch=branch, + upstream_path=self.config.image_sources_upstream_path, + upstream_repo=self.upstream_repo, + ) + return path + if source == "upstream": + return self.ensure_upstream() + raise ImageSourceError("Unknown source. Choose 'upstream' or 'personal'.") diff --git a/nihil/features/tool_selector.py b/nihil/features/tool_selector.py new file mode 100644 index 0000000..5839b90 --- /dev/null +++ b/nihil/features/tool_selector.py @@ -0,0 +1,253 @@ +#!/usr/bin/env python3 +"""Textual checkbox selector for customizing Nihil image tools.""" + +from __future__ import annotations + +from textual import on +from textual.app import App, ComposeResult +from textual.binding import Binding +from textual.containers import Container +from textual.coordinate import Coordinate +from textual.widgets import DataTable, Footer, Header, Input, Static + + +class ToolSearchInput(Input): + """Search input that accepts Tab as an alternative to Enter.""" + + BINDINGS = [Binding("tab", "confirm_search", "Confirm", show=False)] + + def action_confirm_search(self) -> None: + app = self.app + app._set_search_query(self.value) + app._close_search(clear=False) + + +class ToolSelectorApp(App[set[str] | None]): + """Interactive tool selector with Vim-style search and visual ranges.""" + + CSS = """ + Screen { layout: vertical; } + #title { height: 1; padding-left: 1; } + #status { height: 1; padding-left: 1; color: $text-muted; } + #search { display: none; height: 3; border: solid $accent; } + #tools { height: 1fr; } + """ + + BINDINGS = [ + Binding("q", "cancel", "Cancel"), + Binding("escape", "escape_mode", "Escape"), + Binding("enter", "save", "Save", priority=True), + Binding("space", "toggle", "Toggle"), + Binding("v", "visual_toggle", "Visual"), + Binding("/", "search_open", "Search", show=False), + Binding("j", "move_down", "↓", show=False), + Binding("k", "move_up", "↑", show=False), + Binding("g", "move_top", "Top", show=False), + Binding("G", "move_bottom", "Bottom", show=False), + ] + + def __init__(self, tools: list[dict], disabled: set[str], title: str) -> None: + super().__init__() + self.tools = tools + self.disabled = set(disabled) + self.title_text = title + self.search_query = "" + self.visible_indices: list[int] = [] + self.visual_mode = False + self.visual_anchor = 0 + + def compose(self) -> ComposeResult: + yield Header(show_clock=False) + with Container(): + yield Static(self.title_text, id="title") + yield Static( + "↑/↓ or j/k: move | /: search | Space: toggle | v: select range | Enter: save | q: cancel", + id="status", + ) + yield ToolSearchInput(placeholder="/search...", id="search") + yield DataTable(id="tools", cursor_type="row") + yield Footer() + + def on_mount(self) -> None: + self._render_table() + self.query_one("#tools", DataTable).focus() + + def _table(self) -> DataTable: + return self.query_one("#tools", DataTable) + + def _cursor_row(self) -> int: + return max(0, self._table().cursor_row or 0) + + def _tool_index(self, row: int | None = None) -> int: + row = self._cursor_row() if row is None else row + return self.visible_indices[min(row, len(self.visible_indices) - 1)] + + def _visual_rows(self) -> range: + current = self._cursor_row() + return range(min(self.visual_anchor, current), max(self.visual_anchor, current) + 1) + + def _set_status(self, text: str) -> None: + self.query_one("#status", Static).update(text) + + def _matches_search(self, tool: dict) -> bool: + if not self.search_query: + return True + return any( + self.search_query in str(tool.get(field, "")).lower() + for field in ("name", "category", "cmd") + ) + + def _render_table(self) -> None: + table = self._table() + cursor = self._cursor_row() if self.tools else 0 + self.visible_indices = [ + index for index, tool in enumerate(self.tools) + if self._matches_search(tool) + ] + cursor = min(cursor, max(len(self.visible_indices) - 1, 0)) + visual_rows = ( + range(min(self.visual_anchor, cursor), max(self.visual_anchor, cursor) + 1) + if self.visual_mode + else range(0) + ) + table.clear(columns=True) + table.add_column("", width=3) + table.add_column("STATE", width=6) + table.add_column("TOOL") + table.add_column("CATEGORY") + table.add_column("COMMAND") + for row, index in enumerate(self.visible_indices): + tool = self.tools[index] + enabled = tool["mandatory"] or tool["name"] not in self.disabled + marker = "▶" if row in visual_rows else " " + state = "REQ" if tool["mandatory"] else ("ON" if enabled else "OFF") + table.add_row( + marker, state, tool["name"], tool["category"], tool.get("cmd", "-"), key=str(index) + ) + if self.visible_indices: + table.move_cursor(row=cursor, column=0) + + def _refresh_visual_markers(self) -> None: + """Update range markers without rebuilding the table or its scroll state.""" + table = self._table() + selected = self._visual_rows() + for row in range(table.row_count): + table.update_cell_at(Coordinate(row, 0), "▶" if row in selected else " ") + + def action_move_down(self) -> None: + table = self._table() + table.move_cursor(row=min(self._cursor_row() + 1, max(table.row_count - 1, 0))) + if self.visual_mode: + self._refresh_visual_markers() + self._set_status(self._visual_status()) + + def action_move_up(self) -> None: + self._table().move_cursor(row=max(self._cursor_row() - 1, 0)) + if self.visual_mode: + self._refresh_visual_markers() + self._set_status(self._visual_status()) + + def action_move_top(self) -> None: + self._table().move_cursor(row=0) + if self.visual_mode: + self._refresh_visual_markers() + self._set_status(self._visual_status()) + + def action_move_bottom(self) -> None: + table = self._table() + table.move_cursor(row=max(table.row_count - 1, 0)) + if self.visual_mode: + self._refresh_visual_markers() + self._set_status(self._visual_status()) + + def action_visual_toggle(self) -> None: + if not self.visible_indices: + return + self.visual_mode = not self.visual_mode + if self.visual_mode: + self.visual_anchor = self._cursor_row() + self._set_status(self._visual_status()) + else: + self._set_status("Normal mode") + self._render_table() + + def _visual_status(self) -> str: + return f"VISUAL | {len(self._visual_rows())} row(s) selected | Space: toggle range | v: exit visual mode" + + def action_toggle(self) -> None: + rows = self._visual_rows() if self.visual_mode else range(self._cursor_row(), self._cursor_row() + 1) + indexes = [self._tool_index(row) for row in rows if row < len(self.visible_indices)] + mutable = [index for index in indexes if not self.tools[index]["mandatory"]] + if not mutable: + self.visual_mode = False + self._render_table() + self._set_status("Core tools are required and cannot be disabled") + return + disable = any(self.tools[index]["name"] not in self.disabled for index in mutable) + for index in mutable: + name = self.tools[index]["name"] + (self.disabled.add if disable else self.disabled.discard)(name) + was_visual = self.visual_mode + self.visual_mode = False + self._render_table() + suffix = "; visual mode ended" if was_visual else "" + self._set_status(f"{'Disabled' if disable else 'Enabled'} {len(mutable)} tool(s){suffix}") + + def action_search_open(self) -> None: + search = self.query_one("#search", ToolSearchInput) + search.display = True + search.focus() + + def action_escape_mode(self) -> None: + search = self.query_one("#search", ToolSearchInput) + if search.display and search.has_focus: + self._close_search() + return + if self.visual_mode: + self.visual_mode = False + self._render_table() + self._set_status("Normal mode") + + def _set_search_query(self, value: str) -> None: + self.search_query = value.strip().lower() + self._render_table() + self._set_status(f"Search: {self.search_query or 'all tools'} | {len(self.visible_indices)} result(s)") + + def _close_search(self, *, clear: bool = True) -> None: + search = self.query_one("#search", ToolSearchInput) + search.display = False + if clear: + search.value = "" + self._set_search_query("") + self._table().focus() + + @on(Input.Changed, "#search") + def _on_search_changed(self, event: Input.Changed) -> None: + self._set_search_query(event.value) + + @on(Input.Submitted, "#search") + def _on_search_submitted(self, event: Input.Submitted) -> None: + self._set_search_query(event.value) + self._close_search(clear=False) + + def on_key(self, event) -> None: + search = self.query_one("#search", ToolSearchInput) + if event.key == "enter" and search.display and search.has_focus: + self._set_search_query(search.value) + self._close_search(clear=False) + event.stop() + return + if event.key == "escape" and search.display and search.has_focus: + self._close_search() + event.stop() + + def action_save(self) -> None: + search = self.query_one("#search", ToolSearchInput) + if search.display and search.has_focus: + self._set_search_query(search.value) + self._close_search(clear=False) + return + self.exit(result=self.disabled) + + def action_cancel(self) -> None: + self.exit(result=None) diff --git a/pyproject.toml b/pyproject.toml index d97066f..4007e40 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,6 +8,7 @@ dependencies = [ "docker>=7.0.0", "argcomplete>=3.2.0", "rich", + "textual>=0.50.0", "pyyaml>=6.0", ] diff --git a/tests/test_image_sources.py b/tests/test_image_sources.py new file mode 100644 index 0000000..ca0acfc --- /dev/null +++ b/tests/test_image_sources.py @@ -0,0 +1,141 @@ +from pathlib import Path +from types import SimpleNamespace + +from nihil.cli.parser import create_parser +from nihil.features.image_sources import ImageSourceManager + + +def test_image_commands_are_available(): + parser = create_parser() + + customize = parser.parse_args(["image", "customize", "web", "--no-push"]) + assert customize.command == "image" + assert customize.image_action == "customize" + assert customize.variant == "web" + assert customize.no_push is True + assert customize.repo is None + assert customize.git_protocol == "ssh" + assert customize.git_del is False + + https = parser.parse_args(["image", "customize", "web", "--git-protocol", "https"]) + assert https.git_protocol == "https" + + delete = parser.parse_args(["image", "customize", "web", "--git-del"]) + assert delete.git_del is True + + switch = parser.parse_args(["image", "switch", "personal"]) + assert switch.image_action == "switch" + assert switch.source == "personal" + + build = parser.parse_args(["image", "build", "web", "--wait"]) + assert build.image_action == "build" + assert build.variant == "web" + assert build.wait is True + + +def test_repository_urls_are_normalized(tmp_path): + config = SimpleNamespace(image_sources_home=tmp_path) + manager = ImageSourceManager(config, upstream_repo="https://github.com/acme/security-images.git") + assert manager.upstream_repo == "acme/security-images" + + +def test_existing_fork_is_reused_and_custom_branch_is_created(tmp_path): + home = tmp_path / "sources" + path = home / "alice" / "nihil-images" + (path / ".git").mkdir(parents=True) + + config = SimpleNamespace( + image_sources_home=home, + image_sources_upstream_path=home / "upstream" / "nihil-images", + ) + saved = {} + + def set_image_source(**kwargs): + saved.update(kwargs) + + config.set_image_source = set_image_source + manager = ImageSourceManager(config) + + calls = [] + + def fake_run(command, *, cwd=None, capture=True): + calls.append(command) + if command[:3] == ["gh", "api", "user"]: + return "alice" + if command[:4] == ["gh", "repo", "view", "alice/nihil-images"]: + return "name" + if command == ["git", "remote"]: + return "origin\nupstream" + if command[:4] == ["git", "remote", "get-url", "origin"]: + return "https://github.com/alice/nihil-images.git" + if command[:4] == ["git", "remote", "get-url", "upstream"]: + return "https://github.com/TheNullPigeons/nihil-images.git" + if command[:2] == ["git", "branch"]: + return "" + if command[:4] == ["gh", "repo", "view", "TheNullPigeons/nihil-images"]: + return "main" + return "" + + manager._run = fake_run + result_path, repo, branch = manager.ensure_personal_fork(variant="web") + + assert result_path == path + assert repo == "alice/nihil-images" + assert branch == "nihil/web-custom" + assert saved["active"] == "personal" + assert ["gh", "repo", "fork", "TheNullPigeons/nihil-images", "--clone=false"] not in calls + assert ["git", "switch", "-c", "nihil/web-custom", "upstream/main"] in calls + + +def test_trigger_build_dispatches_and_can_wait(tmp_path): + config = SimpleNamespace( + image_sources_home=tmp_path, + personal_image_repo="alice/nihil-images", + personal_image_branch="nihil/web-custom", + ) + manager = ImageSourceManager(config) + calls = [] + + def fake_run(command, *, cwd=None, capture=True): + calls.append(command) + if command[:3] == ["gh", "run", "list"]: + return "12345" + return "" + + manager._run = fake_run + manager.trigger_build(wait=True) + assert [ + "gh", "workflow", "run", "docker-build.yml", + "--repo", "alice/nihil-images", "--ref", "nihil/web-custom", "-f", "variant=all", + ] in calls + assert ["gh", "run", "watch", "12345", "--repo", "alice/nihil-images", "--exit-status"] in calls + + +def test_personal_source_repoints_docker_image_references(): + from nihil.cli.controller import NihilController + + controller = NihilController.__new__(NihilController) + controller.config = SimpleNamespace( + image_source_active="personal", + personal_image_repo="Alice/nihil-images", + personal_image_branch="nihil/web-custom", + ) + controller.manager = SimpleNamespace() + NihilController._configure_image_registry(controller) + + assert controller.manager.AVAILABLE_IMAGES["web"] == "ghcr.io/alice/web:nihil-web-custom" + assert controller.manager.DEFAULT_IMAGE == "ghcr.io/alice/full:nihil-web-custom" + + +def test_personal_source_uses_latest_without_a_custom_branch(): + from nihil.cli.controller import NihilController + + controller = NihilController.__new__(NihilController) + controller.config = SimpleNamespace( + image_source_active="personal", + personal_image_repo="Alice/nihil-images", + ) + controller.manager = SimpleNamespace() + NihilController._configure_image_registry(controller) + + assert controller.manager.AVAILABLE_IMAGES["full"] == "ghcr.io/alice/full:latest"