utils.py

#
#  Copyright © 2021-2026 Mergify SAS
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

from __future__ import annotations

import asyncio
import dataclasses
import functools
import os
import pathlib
import typing
from urllib import parse

import click
import httpx

from mergify_cli import VERSION
from mergify_cli import console
from mergify_cli import console_error
from mergify_cli.ci import github_event
from mergify_cli.exit_codes import ExitCode


if typing.TYPE_CHECKING:
    from collections.abc import Callable
    from collections.abc import Coroutine
    from collections.abc import Mapping


_DEBUG = False


def set_debug(*, debug: bool) -> None:
    global _DEBUG
    _DEBUG = debug


def is_debug() -> bool:
    return _DEBUG


MERGIFY_API_DEFAULT_URL = "https://api.mergify.com"


def get_mergify_api_url() -> str:
    return os.getenv("MERGIFY_API_URL", MERGIFY_API_DEFAULT_URL)


async def check_for_status(response: httpx.Response) -> None:
    if response.status_code < 400:
        return

    await response.aread()

    detail = response.text
    try:
        data = response.json()
        if isinstance(data, dict) and "detail" in data:
            detail = data["detail"]
    except ValueError:
        pass

    console_error(f"API error (HTTP {response.status_code}): {detail}")
    console.print(f"url: {response.request.url}", style="red")

    if is_debug():
        console.print(
            f"request data: {response.request.content.decode('utf-8', errors='replace')}",
            style="red",
        )
        console.print(f"response body: {response.text}", style="red")

    response.raise_for_status()


@dataclasses.dataclass
class CommandError(Exception):
    command_args: tuple[str, ...]
    returncode: int | None
    stdout: bytes

    def __str__(self) -> str:
        # ``errors="replace"`` so str(CommandError) never raises on
        # non-UTF-8 process output — callers in error paths (warnings,
        # CLI top-level handler) rely on this being safe.
        return (
            f"failed to run `{' '.join(self.command_args)}`: "
            f"{self.stdout.decode(errors='replace')}"
        )


class MergifyError(click.ClickException):
    """CLI-level error with a typed exit code.

    Raised from any command path that encounters a semantic failure.
    Click's standalone-mode handler (used by the real entrypoint and by
    CliRunner in tests) catches ClickException subclasses, calls
    ``show()``, then exits with ``self.exit_code``.
    """

    def __init__(
        self,
        message: str,
        *,
        exit_code: ExitCode = ExitCode.GENERIC_ERROR,
    ) -> None:
        super().__init__(message)
        # click.ClickException declares exit_code as an int class attribute.
        # Override per-instance with the typed ExitCode; IntEnum satisfies
        # any `int` consumer (sys.exit, click's standalone-mode handler).
        self.exit_code: ExitCode = exit_code

    def show(self, file: typing.IO[str] | None = None) -> None:
        click.secho(f"error: {self.message}", file=file, err=True, fg="red")


async def run_command(*args: str) -> str:
    if is_debug():
        console.print(f"[purple]DEBUG: running: git {' '.join(args)} [/]")
    try:
        proc = await asyncio.create_subprocess_exec(
            *args,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.STDOUT,
        )
    except FileNotFoundError as e:
        raise CommandError(args, -1, str(e).encode()) from e
    stdout, _ = await proc.communicate()
    if proc.returncode != 0:
        raise CommandError(args, proc.returncode, stdout)
    return stdout.decode().strip()


async def git(*args: str) -> str:
    return await run_command("git", *args)


async def git_get_branch_name() -> str:
    return await git("rev-parse", "--abbrev-ref", "HEAD")


async def git_get_target_branch(branch: str) -> str:
    return (await git("config", "--get", "branch." + branch + ".merge")).removeprefix(
        "refs/heads/",
    )


async def git_get_target_remote(branch: str) -> str:
    return await git("config", "--get", "branch." + branch + ".remote")


async def get_default_branch_prefix(author: str) -> str:
    try:
        result = await git("config", "--get", "mergify-cli.stack-branch-prefix")
    except CommandError:
        result = ""

    return result or f"stack/{author}"


async def get_default_keep_pr_title_body() -> bool:
    try:
        result = await git(
            "config",
            "--get",
            "mergify-cli.stack-keep-pr-title-body",
        )
    except CommandError:
        return False

    return result == "true"


async def get_default_create_as_draft() -> bool:
    try:
        result = await git(
            "config",
            "--get",
            "mergify-cli.stack-create-as-draft",
        )
    except CommandError:
        return False

    return result == "true"


async def get_default_revision_history() -> bool:
    try:
        result = await git(
            "config",
            "--get",
            "mergify-cli.stack-revision-history",
        )
    except CommandError:
        return True

    return result != "false"


async def _get_default_remote_branch() -> tuple[str, str]:
    """Detect the default branch from the remote (e.g. origin/main).

    Returns (remote, branch) tuple.
    """
    ref = await git("symbolic-ref", "refs/remotes/origin/HEAD")
    # ref is like "refs/remotes/origin/main"
    prefix = "refs/remotes/"
    ref = ref.removeprefix(prefix)
    remote, _, branch = ref.partition("/")
    return remote, branch


async def get_trunk() -> str:
    try:
        branch_name = await git_get_branch_name()
    except CommandError:
        console_error("can't get the current branch")
        raise

    target_branch = None
    target_remote = None
    try:
        target_branch = await git_get_target_branch(branch_name)
    except CommandError:
        pass
    try:
        target_remote = await git_get_target_remote(branch_name)
    except CommandError:
        pass

    if target_branch is None or target_remote is None:
        try:
            default_remote, default_branch = await _get_default_remote_branch()
        except CommandError:
            console_error(
                f"can't detect the remote target branch for {branch_name}",
            )
            console.print(
                f"Please set it with `git branch {branch_name} --set-upstream-to=<remote>/<branch>`",
            )
            raise

        if target_branch is None:
            target_branch = default_branch
        if target_remote is None:
            target_remote = default_remote

        await git(
            "branch",
            branch_name,
            "--set-upstream-to",
            f"{target_remote}/{target_branch}",
        )
        console.print(
            f"Upstream not set for {branch_name}, "
            f"automatically set to {target_remote}/{target_branch}",
            style="yellow",
        )

    return f"{target_remote}/{target_branch}"


def get_slug(url: str) -> tuple[str, str]:
    parsed = parse.urlparse(url)
    if not parsed.netloc:
        # Probably ssh
        _, _, path = parsed.path.partition(":")
    else:
        path = parsed.path[1:].rstrip("/")

    user, repo = path.split("/", 1)
    repo = repo.removesuffix(".git")
    return user, repo


# NOTE: must be async for httpx
async def log_httpx_request(request: httpx.Request) -> None:  # noqa: RUF029
    console.print(
        f"[purple]DEBUG: request: {request.method} {request.url} - Waiting for response[/]",
    )


# NOTE: must be async for httpx
async def log_httpx_response(response: httpx.Response) -> None:
    request = response.request
    await response.aread()
    elapsed = response.elapsed.total_seconds()
    console.print(
        f"[purple]DEBUG: response: {request.method} {request.url} - Status {response.status_code} - Elasped {elapsed} s[/]",
    )


def get_http_client(
    server: str,
    *,
    headers: dict[str, typing.Any] | None = None,
    event_hooks: Mapping[str, list[Callable[..., typing.Any]]] | None = None,
    follow_redirects: bool = False,
) -> httpx.AsyncClient:
    default_headers = {"User-Agent": f"mergify_cli/{VERSION}"}
    if headers is not None:
        default_headers |= headers

    default_event_hooks: Mapping[str, list[Callable[..., typing.Any]]] = {
        "request": [],
        "response": [],
    }
    if event_hooks is not None:
        default_event_hooks["request"] += event_hooks["request"]
        default_event_hooks["response"] += event_hooks["response"]
    if is_debug():
        default_event_hooks["request"].insert(0, log_httpx_request)
        default_event_hooks["response"].insert(0, log_httpx_response)

    return httpx.AsyncClient(
        base_url=server,
        headers=default_headers,
        event_hooks=default_event_hooks,
        follow_redirects=follow_redirects,
        timeout=5.0,
    )


def get_mergify_http_client(api_url: str, token: str) -> httpx.AsyncClient:
    event_hooks: Mapping[str, list[Callable[..., typing.Any]]] = {
        "request": [],
        "response": [check_for_status],
    }
    if is_debug():
        event_hooks["request"].insert(0, log_httpx_request)
        event_hooks["response"].insert(0, log_httpx_response)

    return get_http_client(
        api_url,
        headers={
            "Accept": "application/json",
            "Authorization": f"Bearer {token}",
        },
        event_hooks=event_hooks,
        follow_redirects=True,
    )


def get_github_http_client(github_server: str, token: str) -> httpx.AsyncClient:
    event_hooks: Mapping[str, list[Callable[..., typing.Any]]] = {
        "request": [],
        "response": [check_for_status],
    }
    if is_debug():
        event_hooks["request"].insert(0, log_httpx_request)
        event_hooks["response"].insert(0, log_httpx_response)

    return get_http_client(
        github_server,
        headers={
            "Accept": "application/vnd.github.v3+json",
            "Authorization": f"token {token}",
        },
        event_hooks=event_hooks,
        follow_redirects=True,
    )


def get_boolean_env(name: str, *, default: bool = False) -> bool:
    v = os.getenv(name)
    if v is None:
        return default
    return v.strip().lower() in {
        "y",
        "yes",
        "t",
        "true",
        "on",
        "1",
    }


P = typing.ParamSpec("P")
R = typing.TypeVar("R")


def run_with_asyncio[**P, R](
    func: Callable[
        P,
        Coroutine[typing.Any, typing.Any, R],
    ],
) -> functools._Wrapped[
    P,
    Coroutine[typing.Any, typing.Any, R],
    P,
    R,
]:
    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        result = func(*args, **kwargs)
        return asyncio.run(result)

    return wrapper


async def get_default_token() -> str | None:
    token = os.environ.get("MERGIFY_TOKEN") or os.environ.get("GITHUB_TOKEN")
    if not token:
        try:
            token = await run_command("gh", "auth", "token")
        except CommandError:
            console_error(
                "please set the 'MERGIFY_TOKEN' or 'GITHUB_TOKEN' environment variable, "
                "or make sure that gh client is installed and you are authenticated",
            )
            return None
    return token


async def get_default_repository() -> str | None:
    repo = os.environ.get("GITHUB_REPOSITORY")
    if repo:
        return repo

    try:
        remote_url = await git(
            "config",
            "--get",
            "remote.origin.url",
        )
    except CommandError:
        return None

    try:
        user, repo_name = get_slug(remote_url)
    except (ValueError, IndexError):
        return None

    return f"{user}/{repo_name}"


class GitHubEventNotFoundError(Exception):
    pass


def get_github_event() -> tuple[str, github_event.GitHubEvent]:
    event_name = os.environ.get("GITHUB_EVENT_NAME")
    if not event_name:
        raise GitHubEventNotFoundError
    event_path = os.environ.get("GITHUB_EVENT_PATH")
    if event_path and pathlib.Path(event_path).is_file():
        try:
            return event_name, github_event.GitHubEvent.model_validate_json(
                pathlib.Path(event_path).read_text(encoding="utf-8"),
            )
        except FileNotFoundError:
            pass
    raise GitHubEventNotFoundError