diff --git a/.pylintrc b/.pylintrc new file mode 100644 index 0000000..8fa8b97 --- /dev/null +++ b/.pylintrc @@ -0,0 +1,4 @@ +# .pylintrc or pylintrc + +[MAIN] +max-line-length=120 diff --git a/comfy_cli/cmdline.py b/comfy_cli/cmdline.py index 28b619c..fa718f7 100644 --- a/comfy_cli/cmdline.py +++ b/comfy_cli/cmdline.py @@ -6,7 +6,7 @@ import questionary import typer -from rich import print +from rich import print as rprint from rich.console import Console from typing_extensions import Annotated, List @@ -14,6 +14,7 @@ from comfy_cli.command import custom_nodes from comfy_cli.command import install as install_inner from comfy_cli.command import run as run_inner +from comfy_cli.command.install import validate_version from comfy_cli.command.launch import launch as launch_command from comfy_cli.command.models import models as models_command from comfy_cli.config_manager import ConfigManager @@ -56,7 +57,7 @@ def validate(self, _ctx: typer.Context, param: typer.CallbackParam, value: str): @app.command(help="Display help for commands") def help(ctx: typer.Context): - print(ctx.find_root().get_help()) + rprint(ctx.find_root().get_help()) ctx.exit(0) @@ -115,7 +116,7 @@ def entry( ), ): if version: - print(ConfigManager().get_cli_version()) + rprint(ConfigManager().get_cli_version()) ctx.exit(0) workspace_manager.setup_workspace_manager(workspace, here, recent, skip_prompt) @@ -123,8 +124,8 @@ def entry( tracking.prompt_tracking_consent(skip_prompt, default_value=enable_telemetry) if ctx.invoked_subcommand is None: - print("[bold yellow]Welcome to Comfy CLI![/bold yellow]: https://github.com/Comfy-Org/comfy-cli") - print(ctx.get_help()) + rprint("[bold yellow]Welcome to Comfy CLI![/bold yellow]: https://github.com/Comfy-Org/comfy-cli") + rprint(ctx.get_help()) ctx.exit() # TODO: Move this to proper place @@ -135,6 +136,15 @@ def entry( # logging.info(f"scan_dir took {end_time - start_time:.2f} seconds to run") +def validate_commit_and_version(commit: Optional[str], ctx: typer.Context): + """ + Validate that the commit is not specified unless the version is 'nightly'. + """ + version = ctx.params.get("version") + if commit and version != "nightly": + raise typer.BadParameter("You can only specify the commit if the version is 'nightly'.") + + @app.command(help="Download and install ComfyUI and ComfyUI-Manager") @tracking.track_command() def install( @@ -145,6 +155,14 @@ def install( help="url or local path pointing to the ComfyUI core git repo to be installed. A specific branch can optionally be specified using a setuptools-like syntax, eg https://foo.git@bar", ), ] = constants.COMFY_GITHUB_URL, + version: Annotated[ + str, + typer.Option( + show_default=False, + help="Specify version of ComfyUI to install. Default is nightl, which is the latest commit on master branch. Other options include: latest, which is the latest stable release. Or a specific version number, eg. 0.2.0", + callback=validate_version, + ), + ] = "nightly", manager_url: Annotated[ str, typer.Option( @@ -212,9 +230,11 @@ def install( callback=g_gpu_exclusivity.validate, ), ] = None, - commit: Annotated[Optional[str], typer.Option(help="Specify commit hash for ComfyUI")] = None, + commit: Annotated[ + Optional[str], typer.Option(help="Specify commit hash for ComfyUI", callback=validate_commit_and_version) + ] = None, fast_deps: Annotated[ - Optional[bool], + bool, typer.Option( "--fast-deps", show_default=False, @@ -229,8 +249,8 @@ def install( is_comfy_installed_at_path, repo_dir = check_comfy_repo(comfy_path) if is_comfy_installed_at_path and not restore: - print(f"[bold red]ComfyUI is already installed at the specified path:[/bold red] {comfy_path}\n") - print( + rprint(f"[bold red]ComfyUI is already installed at the specified path:[/bold red] {comfy_path}\n") + rprint( "[bold yellow]If you want to restore dependencies, add the '--restore' option.[/bold yellow]", ) raise typer.Exit(code=1) @@ -239,11 +259,11 @@ def install( comfy_path = str(repo_dir.working_dir) if checker.python_version.major < 3 or checker.python_version.minor < 9: - print("[bold red]Python version 3.9 or higher is required to run ComfyUI.[/bold red]") - print(f"You are currently using Python version {env_checker.format_python_version(checker.python_version)}.") + rprint("[bold red]Python version 3.9 or higher is required to run ComfyUI.[/bold red]") + rprint(f"You are currently using Python version {env_checker.format_python_version(checker.python_version)}.") platform = utils.get_os() if cpu: - print("[bold yellow]Installing for CPU[/bold yellow]") + rprint("[bold yellow]Installing for CPU[/bold yellow]") install_inner.execute( url, manager_url, @@ -251,22 +271,23 @@ def install( restore, skip_manager, commit=commit, - gpu=None, + version=version, + gpu=GPU_OPTION.CPU, cuda_version=cuda_version, plat=platform, skip_torch_or_directml=skip_torch_or_directml, skip_requirement=skip_requirement, fast_deps=fast_deps, ) - print(f"ComfyUI is installed at: {comfy_path}") + rprint(f"ComfyUI is installed at: {comfy_path}") return None if nvidia and platform == constants.OS.MACOS: - print("[bold red]Nvidia GPU is never on MacOS. What are you smoking? šŸ¤”[/bold red]") + rprint("[bold red]Nvidia GPU is never on MacOS. What are you smoking? šŸ¤”[/bold red]") raise typer.Exit(code=1) if platform != constants.OS.MACOS and m_series: - print(f"[bold red]You are on {platform} bruh [/bold red]") + rprint(f"[bold red]You are on {platform} bruh [/bold red]") gpu = None @@ -291,10 +312,10 @@ def install( ) if gpu == GPU_OPTION.INTEL_ARC: - print("[bold yellow]Installing on Intel ARC is not yet completely supported[/bold yellow]") + rprint("[bold yellow]Installing on Intel ARC is not yet completely supported[/bold yellow]") env_check = env_checker.EnvChecker() if env_check.conda_env is None: - print("[bold red]Intel ARC support requires conda environment to be activated.[/bold red]") + rprint("[bold red]Intel ARC support requires conda environment to be activated.[/bold red]") raise typer.Exit(code=1) if intel_arc is None: confirm_result = ui.prompt_confirm_action( @@ -302,10 +323,10 @@ def install( ) if not confirm_result: raise typer.Exit(code=0) - print("[bold yellow]Installing on Intel ARC is in beta stage.[/bold yellow]") + rprint("[bold yellow]Installing on Intel ARC is in beta stage.[/bold yellow]") if gpu is None and not cpu: - print( + rprint( "[bold red]No GPU option selected or `--cpu` enabled, use --\\[gpu option] flag (e.g. --nvidia) to pick GPU. use `--cpu` to install for CPU. Exiting...[/bold red]" ) raise typer.Exit(code=1) @@ -318,6 +339,7 @@ def install( skip_manager, commit=commit, gpu=gpu, + version=version, cuda_version=cuda_version, plat=platform, skip_torch_or_directml=skip_torch_or_directml, @@ -325,7 +347,7 @@ def install( fast_deps=fast_deps, ) - print(f"ComfyUI is installed at: {comfy_path}") + rprint(f"ComfyUI is installed at: {comfy_path}") @app.command(help="Update ComfyUI Environment [all|comfy]") @@ -349,9 +371,9 @@ def update( if "all" == target: custom_nodes.command.execute_cm_cli(["update", "all"]) else: - print(f"Updating ComfyUI in {comfy_path}...") + rprint(f"Updating ComfyUI in {comfy_path}...") if comfy_path is None: - print("ComfyUI path is not found.") + rprint("ComfyUI path is not found.") raise typer.Exit(code=1) os.chdir(comfy_path) subprocess.run(["git", "pull"], check=True) @@ -416,7 +438,7 @@ def run( def validate_comfyui(_env_checker): if _env_checker.comfy_repo is None: - print("[bold red]If ComfyUI is not installed, this feature cannot be used.[/bold red]") + rprint("[bold red]If ComfyUI is not installed, this feature cannot be used.[/bold red]") raise typer.Exit(code=1) @@ -424,19 +446,19 @@ def validate_comfyui(_env_checker): @tracking.track_command() def stop(): if constants.CONFIG_KEY_BACKGROUND not in ConfigManager().config["DEFAULT"]: - print("[bold red]No ComfyUI is running in the background.[/bold red]\n") + rprint("[bold red]No ComfyUI is running in the background.[/bold red]\n") raise typer.Exit(code=1) bg_info = ConfigManager().background if not bg_info: - print("[bold red]No ComfyUI is running in the background.[/bold red]\n") + rprint("[bold red]No ComfyUI is running in the background.[/bold red]\n") raise typer.Exit(code=1) is_killed = utils.kill_all(bg_info[2]) if not is_killed: - print("[bold red]Failed to stop ComfyUI in the background.[/bold red]\n") + rprint("[bold red]Failed to stop ComfyUI in the background.[/bold red]\n") else: - print(f"[bold yellow]Background ComfyUI is stopped.[/bold yellow] ({bg_info[0]}:{bg_info[1]})") + rprint(f"[bold yellow]Background ComfyUI is stopped.[/bold yellow] ({bg_info[0]}:{bg_info[1]})") ConfigManager().remove_background() @@ -459,7 +481,7 @@ def set_default( comfy_path = os.path.abspath(os.path.expanduser(workspace_path)) if not os.path.exists(comfy_path): - print( + rprint( f"\nPath not found: {comfy_path}.\n", file=sys.stderr, ) @@ -467,7 +489,7 @@ def set_default( is_comfy_repo, comfy_repo = check_comfy_repo(comfy_path) if not is_comfy_repo: - print( + rprint( f"\nSpecified path is not a ComfyUI path: {comfy_path}.\n", file=sys.stderr, ) @@ -475,7 +497,7 @@ def set_default( comfy_path = comfy_repo.working_dir - print(f"Specified path is set as default ComfyUI path: {comfy_path} ") + rprint(f"Specified path is set as default ComfyUI path: {comfy_path} ") workspace_manager.set_default_workspace(comfy_path) workspace_manager.set_default_launch_extras(launch_extras) @@ -485,12 +507,12 @@ def set_default( def which(): comfy_path = workspace_manager.workspace_path if comfy_path is None: - print( + rprint( "ComfyUI not found, please run 'comfy install', run 'comfy' in a ComfyUI directory, or specify the workspace path with '--workspace'." ) raise typer.Exit(code=1) - print(f"Target ComfyUI path: {comfy_path}") + rprint(f"Target ComfyUI path: {comfy_path}") @app.command(help="Print out current environment variables.") @@ -506,19 +528,19 @@ def env(): @app.command(hidden=True) @tracking.track_command() def nodes(): - print("\n[bold red] No such command, did you mean 'comfy node' instead?[/bold red]\n") + rprint("\n[bold red] No such command, did you mean 'comfy node' instead?[/bold red]\n") @app.command(hidden=True) @tracking.track_command() def models(): - print("\n[bold red] No such command, did you mean 'comfy model' instead?[/bold red]\n") + rprint("\n[bold red] No such command, did you mean 'comfy model' instead?[/bold red]\n") @app.command(help="Provide feedback on the Comfy CLI tool.") @tracking.track_command() def feedback(): - print("Feedback Collection for Comfy CLI Tool\n") + rprint("Feedback Collection for Comfy CLI Tool\n") # General Satisfaction general_satisfaction_score = ui.prompt_select( @@ -541,7 +563,7 @@ def feedback(): tracking.track_event("feedback_additional") webbrowser.open("https://github.com/Comfy-Org/comfy-cli/issues/new/choose") - print("Thank you for your feedback!") + rprint("Thank you for your feedback!") @app.command(help="Download a standalone Python interpreter and dependencies based on an existing comfyui workspace") diff --git a/comfy_cli/command/install.py b/comfy_cli/command/install.py index c25d970..ceece79 100644 --- a/comfy_cli/command/install.py +++ b/comfy_cli/command/install.py @@ -2,18 +2,24 @@ import platform import subprocess import sys -from typing import Optional +from typing import Dict, List, Optional, TypedDict +import requests +import semver import typer -from rich import print +from rich import print as rprint +from rich.console import Console +from rich.panel import Panel from comfy_cli import constants, ui, utils from comfy_cli.command.custom_nodes.command import update_node_id_cache from comfy_cli.constants import GPU_OPTION +from comfy_cli.git_utils import git_checkout_tag from comfy_cli.uv import DependencyCompiler from comfy_cli.workspace_manager import WorkspaceManager, check_comfy_repo workspace_manager = WorkspaceManager() +console = Console() def get_os_details(): @@ -103,7 +109,7 @@ def pip_install_comfyui_dependencies( check=False, ) if result and result.returncode != 0: - print("Failed to install PyTorch dependencies. Please check your environment (`comfy env`) and try again") + rprint("Failed to install PyTorch dependencies. Please check your environment (`comfy env`) and try again") sys.exit(1) # install directml for AMD windows @@ -133,7 +139,7 @@ def pip_install_comfyui_dependencies( return result = subprocess.run([sys.executable, "-m", "pip", "install", "-r", "requirements.txt"], check=False) if result.returncode != 0: - print("Failed to install ComfyUI dependencies. Please check your environment (`comfy env`) and try again.") + rprint("Failed to install ComfyUI dependencies. Please check your environment (`comfy env`) and try again.") sys.exit(1) @@ -149,6 +155,7 @@ def execute( comfy_path: str, restore: bool, skip_manager: bool, + version: str, commit: Optional[str] = None, gpu: constants.GPU_OPTION = None, cuda_version: constants.CUDAVersion = constants.CUDAVersion.v12_1, @@ -159,14 +166,17 @@ def execute( *args, **kwargs, ): + """ + Install ComfyUI from a given URL. + """ if not workspace_manager.skip_prompting: res = ui.prompt_confirm_action(f"Install from {url} to {comfy_path}?", True) if not res: - print("Aborting...") + rprint("Aborting...") raise typer.Exit(code=1) - print(f"Installing from [bold yellow]'{url}'[/bold yellow] to '{comfy_path}'") + rprint(f"Installing from repository [bold yellow]'{url}'[/bold yellow] to '{comfy_path}'") repo_dir = comfy_path parent_path = os.path.abspath(os.path.join(repo_dir, "..")) @@ -175,15 +185,13 @@ def execute( os.makedirs(parent_path, exist_ok=True) if not os.path.exists(repo_dir): - if "@" in url: - # clone specific branch - url, branch = url.rsplit("@", 1) - subprocess.run(["git", "clone", "-b", branch, url, repo_dir], check=True) - else: - subprocess.run(["git", "clone", url, repo_dir], check=True) + clone_comfyui(url=url, repo_dir=repo_dir) + + if version != "nightly": + checkout_stable_comfyui(version=version, repo_dir=repo_dir) elif not check_comfy_repo(repo_dir)[0]: - print( + rprint( f"[bold red]'{repo_dir}' already exists. But it is an invalid ComfyUI repository. Remove it and retry.[/bold red]" ) exit(-1) @@ -199,11 +207,11 @@ def execute( WorkspaceManager().set_recent_workspace(repo_dir) workspace_manager.setup_workspace_manager(specified_workspace=repo_dir) - print("") + rprint("") # install ComfyUI-Manager if skip_manager: - print("Skipping installation of ComfyUI-Manager. (by --skip-manager)") + rprint("Skipping installation of ComfyUI-Manager. (by --skip-manager)") else: manager_repo_dir = os.path.join(repo_dir, "custom_nodes", "ComfyUI-Manager") @@ -211,16 +219,19 @@ def execute( if restore and not fast_deps: pip_install_manager_dependencies(repo_dir) else: - print( + rprint( f"Directory {manager_repo_dir} already exists. Skipping installation of ComfyUI-Manager.\nIf you want to restore dependencies, add the '--restore' option." ) else: - print("\nInstalling ComfyUI-Manager..") + rprint("\nInstalling ComfyUI-Manager..") if "@" in manager_url: # clone specific branch manager_url, manager_branch = manager_url.rsplit("@", 1) - subprocess.run(["git", "clone", "-b", manager_branch, manager_url, manager_repo_dir], check=True) + subprocess.run( + ["git", "clone", "-b", manager_branch, manager_url, manager_repo_dir], + check=True, + ) else: subprocess.run(["git", "clone", manager_url, manager_repo_dir], check=True) @@ -237,4 +248,168 @@ def execute( os.chdir(repo_dir) - print("") + rprint("") + + +def validate_version(version: str) -> Optional[str]: + """ + Validates the version string as 'latest', 'nightly', or a semantically version number. + + Args: + version (str): The version string to validate. + + Returns: + Optional[str]: The validated version string, or None if invalid. + + Raises: + ValueError: If the version string is invalid. + """ + if version.lower() in ["nightly", "latest"]: + return version.lower() + + # Remove 'v' prefix if present + if version.startswith("v"): + version = version[1:] + + try: + semver.VersionInfo.parse(version) + return version + except ValueError as exc: + raise ValueError( + f"Invalid version format: {version}. " + "Please use 'nightly', 'latest', or a valid semantic version (e.g., '1.2.3')." + ) from exc + + +def fetch_github_releases(repo_owner: str, repo_name: str) -> List[Dict[str, str]]: + """ + Fetch the list of releases from the GitHub API. + """ + url = f"https://api.github.com/repos/{repo_owner}/{repo_name}/releases" + response = requests.get(url) + return response.json() + + +class GithubRelease(TypedDict): + """ + A dictionary representing a GitHub release. + + Fields: + - version: The version number of the release. (Removed the v prefix) + - tag: The tag name of the release. + - download_url: The URL to download the release. + """ + + version: Optional[semver.VersionInfo] + tag: str + download_url: str + + +def parse_releases(releases: List[Dict[str, str]]) -> List[GithubRelease]: + """ + Parse the list of releases fetched from the GitHub API into a list of GithubRelease objects. + """ + parsed_releases: List[GithubRelease] = [] + for release in releases: + tag = release["tag_name"] + if tag.lower() in ["latest", "nightly"]: + parsed_releases.append({"version": None, "download_url": release["zipball_url"], "tag": tag}) + else: + version = semver.VersionInfo.parse(tag.lstrip("v")) + parsed_releases.append({"version": version, "download_url": release["zipball_url"], "tag": tag}) + + return parsed_releases + + +def select_version(releases: List[GithubRelease], version: str) -> Optional[GithubRelease]: + """ + Given a list of Github releases, select the release that matches the specified version. + """ + if version.lower() == "latest": + return next((r for r in releases if r["tag"].lower() == version.lower()), None) + + version = version.lstrip("v") + + try: + requested_version = semver.VersionInfo.parse(version) + return next( + (r for r in releases if isinstance(r["version"], semver.VersionInfo) and r["version"] == requested_version), + None, + ) + except ValueError: + return None + + +def clone_comfyui(url: str, repo_dir: str): + """ + Clone the ComfyUI repository from the specified URL. + """ + if "@" in url: + # clone specific branch + url, branch = url.rsplit("@", 1) + subprocess.run(["git", "clone", "-b", branch, url, repo_dir], check=True) + else: + subprocess.run(["git", "clone", url, repo_dir], check=True) + + +def checkout_stable_comfyui(version: str, repo_dir: str): + """ + Supports installing stable releases of Comfy (semantic versioning) or the 'latest' version. + """ + rprint(f"Looking for ComfyUI version '{version}'...") + selected_release = None + if version == "latest": + selected_release = get_latest_release("comfyanonymous", "ComfyUI") + else: + releases = fetch_github_releases("comfyanonymous", "ComfyUI") + parsed_releases = parse_releases(releases) + selected_release = select_version(parsed_releases, version) + + if selected_release is None: + rprint(f"Error: No release found for version '{version}'.") + sys.exit(1) + + tag = str(selected_release["tag"]) + console.print( + Panel( + f"šŸ” Checking out ComfyUI version: [bold cyan]{selected_release['tag']}[/bold cyan]", + title="[yellow]ComfyUI Checkout[/yellow]", + border_style="green", + expand=False, + ) + ) + + with console.status("[bold green]Checking out tag...", spinner="dots"): + success = git_checkout_tag(repo_dir, tag) + if not success: + console.print("\nāŒ [bold red]Failed to checkout tag![/bold red]") + sys.exit(1) + + console.print("\nāœ… [bold green]Successfully checked out tag![/bold green]") + + +def get_latest_release(repo_owner: str, repo_name: str) -> Optional[GithubRelease]: + """ + Fetch the latest release information from GitHub API. + + :param repo_owner: The owner of the repository + :param repo_name: The name of the repository + :return: A dictionary containing release information, or None if failed + """ + url = f"https://api.github.com/repos/{repo_owner}/{repo_name}/releases/latest" + + try: + response = requests.get(url, timeout=5) + response.raise_for_status() + + data = response.json() + + return GithubRelease( + tag=data["tag_name"], + version=semver.VersionInfo.parse(data["tag_name"].lstrip("v")), + download_url=data["zipball_url"], + ) + + except requests.RequestException as e: + rprint(f"Error fetching latest release: {e}") + return None diff --git a/comfy_cli/git_utils.py b/comfy_cli/git_utils.py new file mode 100644 index 0000000..1acdad5 --- /dev/null +++ b/comfy_cli/git_utils.py @@ -0,0 +1,58 @@ +import os +import subprocess + +from rich.console import Console +from rich.panel import Panel +from rich.text import Text + +console = Console() + + +def git_checkout_tag(repo_path: str, tag: str) -> bool: + """ + Checkout a specific Git tag in the given repository. + + :param repo_path: Path to the Git repository + :param tag: The tag to checkout + :return: The output of the git command if successful, None if an error occurred + """ + original_dir = os.getcwd() + try: + # Change to the repository directory + + os.chdir(repo_path) + + # Fetch the latest tags + subprocess.run(["git", "fetch", "--tags"], check=True, capture_output=True, text=True) + + # Checkout the specified tag + subprocess.run(["git", "checkout", tag], check=True, capture_output=True, text=True) + + console.print(f"[bold green]Successfully checked out tag: [cyan]{tag}[/cyan][/bold green]") + + return True + except subprocess.CalledProcessError as e: + error_message = Text() + error_message.append("Git Checkout Error", style="bold red on white") + error_message.append("\n\nFailed to checkout tag: ", style="bold yellow") + error_message.append(f"[cyan]{tag}[/cyan]") + error_message.append("\n\nError details:", style="bold red") + error_message.append(f"\n{str(e)}", style="italic") + + if e.stderr: + error_message.append("\n\nError output:", style="bold red") + error_message.append(f"\n{e.stderr}", style="italic yellow") + + console.print( + Panel( + error_message, + title="[bold white on red]Git Checkout Failed[/bold white on red]", + border_style="red", + expand=False, + ) + ) + + return False + finally: + # Ensure we always return to the original directory + os.chdir(original_dir) diff --git a/pyproject.toml b/pyproject.toml index 272d47c..1c6937e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,51 +4,48 @@ build-backend = "setuptools.build_meta" [project] name = "comfy-cli" -license = {text = "GPL-3.0-only"} -version = "0.0.0" # Will be filled in by the CI/CD pipeline. Check publish_package.py. +license = { text = "GPL-3.0-only" } +version = "0.0.0" # Will be filled in by the CI/CD pipeline. Check publish_package.py. requires-python = ">= 3.9" description = "A CLI tool for installing and using ComfyUI." readme = "README.md" keywords = ["comfyui", "stable diffusion"] maintainers = [ - {name = "Yoland Yan", email = "yoland@drip.art"}, - {name = "James Kwon", email = "hongilkwon316@gmail.com"}, - {name = "Robin Huang", email = "robin@drip.art"}, - {name = "Dr.Lt.Data", email = "dr.lt.data@gmail.com"}, + { name = "Yoland Yan", email = "yoland@drip.art" }, + { name = "James Kwon", email = "hongilkwon316@gmail.com" }, + { name = "Robin Huang", email = "robin@drip.art" }, + { name = "Dr.Lt.Data", email = "dr.lt.data@gmail.com" }, ] classifiers = [ - "Development Status :: 4 - Beta", - "Intended Audience :: Developers", - "License :: OSI Approved :: GNU General Public License v3 (GPLv3)", + "Development Status :: 4 - Beta", + "Intended Audience :: Developers", + "License :: OSI Approved :: GNU General Public License v3 (GPLv3)", ] dependencies = [ - "charset-normalizer>=3.0.0", - "GitPython", - "httpx", - "mixpanel", - "packaging", - "pathspec", - "psutil", - "pyyaml", - "questionary", - "requests", - "rich", - "tomlkit", - "typer>=0.9.0", - "typing-extensions>=4.7.0", - "uv", - "websocket-client", + "charset-normalizer>=3.0.0", + "GitPython", + "httpx", + "mixpanel", + "packaging", + "pathspec", + "psutil", + "pyyaml", + "questionary", + "requests", + "rich", + "tomlkit", + "typer>=0.9.0", + "typing-extensions>=4.7.0", + "uv", + "websocket-client", + "semver~=3.0.2", ] [project.optional-dependencies] -dev = [ - "pre-commit", - "pytest", - "ruff", -] +dev = ["pre-commit", "pytest", "ruff"] [project.scripts] comfy = "comfy_cli.__main__:main" @@ -68,9 +65,9 @@ target-version = "py39" [tool.ruff.lint] select = [ - "E4", # default - "E7", # default - "E9", # default - "F", # default - "I", # isort-like behavior (import statement sorting) + "E4", # default + "E7", # default + "E9", # default + "F", # default + "I", # isort-like behavior (import statement sorting) ] diff --git a/tests/comfy_cli/test_install.py b/tests/comfy_cli/test_install.py new file mode 100644 index 0000000..1e97b96 --- /dev/null +++ b/tests/comfy_cli/test_install.py @@ -0,0 +1,205 @@ +from typing import Dict, List +from unittest.mock import MagicMock, patch + +import pytest +import requests +import semver + +from comfy_cli.command.install import ( + GithubRelease, + fetch_github_releases, + parse_releases, + select_version, + validate_version, +) + + +def test_validate_version_nightly(): + assert validate_version("nightly") == "nightly" + assert validate_version("NIGHTLY") == "nightly" + + +def test_validate_version_latest(): + assert validate_version("latest") == "latest" + assert validate_version("LATEST") == "latest" + + +def test_validate_version_valid_semver(): + assert validate_version("1.2.3") == "1.2.3" + assert validate_version("v1.2.3") == "1.2.3" + assert validate_version("1.2.3-alpha") == "1.2.3-alpha" + + +def test_validate_version_invalid(): + with pytest.raises(ValueError): + validate_version("invalid_version") + + +def test_validate_version_empty(): + with pytest.raises(ValueError): + validate_version("") + + +# Tests for fetch_github_releases function +@patch("requests.get") +def test_fetch_releases_success(mock_get): + # Mock the response + mock_response = MagicMock() + mock_response.json.return_value = [{"id": 1, "tag_name": "v1.0.0"}, {"id": 2, "tag_name": "v1.1.0"}] + mock_get.return_value = mock_response + + releases = fetch_github_releases("owner", "repo") + + assert len(releases) == 2 + assert releases[0]["tag_name"] == "v1.0.0" + assert releases[1]["tag_name"] == "v1.1.0" + mock_get.assert_called_once_with("https://api.github.com/repos/owner/repo/releases") + + +@patch("requests.get") +def test_fetch_releases_empty(mock_get): + # Mock an empty response + mock_response = MagicMock() + mock_response.json.return_value = [] + mock_get.return_value = mock_response + + releases = fetch_github_releases("owner", "repo") + + assert len(releases) == 0 + + +@patch("requests.get") +def test_fetch_releases_error(mock_get): + # Mock a request exception + mock_get.side_effect = requests.RequestException("API error") + + with pytest.raises(requests.RequestException): + fetch_github_releases("owner", "repo") + + +def test_parse_releases_with_semver(): + input_releases = [ + {"tag_name": "v1.2.3", "zipball_url": "https://api.github.com/repos/owner/repo/zipball/v1.2.3"}, + {"tag_name": "2.0.0", "zipball_url": "https://api.github.com/repos/owner/repo/zipball/2.0.0"}, + ] + + result = parse_releases(input_releases) + + assert len(result) == 2 + assert result[0]["version"] == semver.VersionInfo.parse("1.2.3") + assert result[0]["tag"] == "v1.2.3" + assert result[0]["download_url"] == "https://api.github.com/repos/owner/repo/zipball/v1.2.3" + assert result[1]["version"] == semver.VersionInfo.parse("2.0.0") + assert result[1]["tag"] == "2.0.0" + + +def test_parse_releases_with_special_tags(): + input_releases = [ + {"tag_name": "latest", "zipball_url": "https://api.github.com/repos/owner/repo/zipball/latest"}, + {"tag_name": "nightly", "zipball_url": "https://api.github.com/repos/owner/repo/zipball/nightly"}, + ] + + result = parse_releases(input_releases) + + assert len(result) == 2 + assert result[0]["version"] is None + assert result[0]["tag"] == "latest" + assert result[1]["version"] is None + assert result[1]["tag"] == "nightly" + + +def test_parse_releases_mixed(): + input_releases = [ + {"tag_name": "v1.0.0", "zipball_url": "https://api.github.com/repos/owner/repo/zipball/v1.0.0"}, + {"tag_name": "latest", "zipball_url": "https://api.github.com/repos/owner/repo/zipball/latest"}, + {"tag_name": "2.0.0-beta", "zipball_url": "https://api.github.com/repos/owner/repo/zipball/2.0.0-beta"}, + ] + + result = parse_releases(input_releases) + + assert len(result) == 3 + assert result[0]["version"] == semver.VersionInfo.parse("1.0.0") + assert result[1]["version"] is None + assert result[1]["tag"] == "latest" + assert result[2]["version"] == semver.VersionInfo.parse("2.0.0-beta") + + +def test_parse_releases_empty_list(): + input_releases: List[Dict[str, str]] = [] + + result = parse_releases(input_releases) + + assert len(result) == 0 + + +def test_parse_releases_invalid_semver(): + input_releases = [ + {"tag_name": "invalid", "zipball_url": "https://api.github.com/repos/owner/repo/zipball/invalid"}, + ] + + with pytest.raises(ValueError): + parse_releases(input_releases) + + +# Sample data for tests +sample_releases: List[GithubRelease] = [ + {"version": semver.VersionInfo.parse("1.0.0"), "tag": "v1.0.0", "download_url": "url1"}, + {"version": semver.VersionInfo.parse("1.1.0"), "tag": "v1.1.0", "download_url": "url2"}, + {"version": semver.VersionInfo.parse("2.0.0"), "tag": "v2.0.0", "download_url": "url3"}, + {"version": None, "tag": "latest", "download_url": "url_latest"}, + {"version": None, "tag": "nightly", "download_url": "url_nightly"}, +] + + +def test_select_version_latest(): + result = select_version(sample_releases, "latest") + assert result is not None + assert result["tag"] == "latest" + assert result["download_url"] == "url_latest" + + +def test_select_version_specific(): + result = select_version(sample_releases, "1.1.0") + assert result is not None + assert result["version"] == semver.VersionInfo.parse("1.1.0") + assert result["tag"] == "v1.1.0" + + +def test_select_version_with_v_prefix(): + result = select_version(sample_releases, "v2.0.0") + assert result is not None + assert result["version"] == semver.VersionInfo.parse("2.0.0") + assert result["tag"] == "v2.0.0" + + +def test_select_version_nonexistent(): + result = select_version(sample_releases, "3.0.0") + assert result is None + + +def test_select_version_invalid(): + result = select_version(sample_releases, "invalid_version") + assert result is None + + +def test_select_version_case_insensitive_latest(): + result = select_version(sample_releases, "LATEST") + assert result is not None + assert result["tag"] == "latest" + + +def test_select_version_nightly(): + # Note: This test will fail with the current implementation + # as it doesn't handle "nightly" specifically + result = select_version(sample_releases, "nightly") + assert result is None # or assert result is not None if you want to handle nightly + + +def test_select_version_empty_list(): + result = select_version([], "1.0.0") + assert result is None + + +# Run the tests +if __name__ == "__main__": + pytest.main([__file__])