From e5d5b20798f7efb276bae6f3bef0e891815e06af Mon Sep 17 00:00:00 2001 From: Julien Danjou Date: Thu, 26 Feb 2026 20:03:59 +0100 Subject: [PATCH] feat(freeze): add scheduled freeze CLI commands Add `mergify freeze` command group with list, create, update, and delete subcommands for managing scheduled merge freezes. Supports both Mergify application keys and GitHub tokens for authentication, auto-detects repository from git remote, and provides both table and JSON output. Co-Authored-By: Claude Opus 4.6 Change-Id: I414d168066f7b5070e75520bdcbb8189d694ad72 Claude-Session-Id: 77d8be2d-854a-4d80-ac2d-4489fa50e31d --- mergify_cli/cli.py | 2 + mergify_cli/freeze/__init__.py | 0 mergify_cli/freeze/api.py | 99 +++++++ mergify_cli/freeze/cli.py | 389 ++++++++++++++++++++++++ mergify_cli/tests/freeze/__init__.py | 0 mergify_cli/tests/freeze/test_cli.py | 425 +++++++++++++++++++++++++++ 6 files changed, 915 insertions(+) create mode 100644 mergify_cli/freeze/__init__.py create mode 100644 mergify_cli/freeze/api.py create mode 100644 mergify_cli/freeze/cli.py create mode 100644 mergify_cli/tests/freeze/__init__.py create mode 100644 mergify_cli/tests/freeze/test_cli.py diff --git a/mergify_cli/cli.py b/mergify_cli/cli.py index 75f4e528..0395f39c 100644 --- a/mergify_cli/cli.py +++ b/mergify_cli/cli.py @@ -23,6 +23,7 @@ from mergify_cli import VERSION from mergify_cli.ci import cli as ci_cli_mod +from mergify_cli.freeze import cli as freeze_cli_mod from mergify_cli.stack import cli as stack_cli_mod @@ -44,6 +45,7 @@ def cli( cli.add_command(stack_cli_mod.stack) cli.add_command(ci_cli_mod.ci) +cli.add_command(freeze_cli_mod.freeze) def main() -> None: diff --git a/mergify_cli/freeze/__init__.py b/mergify_cli/freeze/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/mergify_cli/freeze/api.py b/mergify_cli/freeze/api.py new file mode 100644 index 00000000..317ef13d --- /dev/null +++ b/mergify_cli/freeze/api.py @@ -0,0 +1,99 @@ +from __future__ import annotations + +import typing + + +if typing.TYPE_CHECKING: + import datetime + import uuid + + import httpx + + +async def list_freezes( + client: httpx.AsyncClient, + repository: str, +) -> list[dict[str, typing.Any]]: + response = await client.get( + f"/v1/repos/{repository}/scheduled_freeze", + ) + return response.json()["scheduled_freezes"] # type: ignore[no-any-return] + + +async def create_freeze( + client: httpx.AsyncClient, + repository: str, + *, + reason: str, + timezone: str, + matching_conditions: list[str], + start: datetime.datetime | None = None, + end: datetime.datetime | None = None, + exclude_conditions: list[str] | None = None, +) -> dict[str, typing.Any]: + payload: dict[str, typing.Any] = { + "reason": reason, + "timezone": timezone, + "matching_conditions": matching_conditions, + } + if start is not None: + payload["start"] = start.isoformat() + if end is not None: + payload["end"] = end.isoformat() + if exclude_conditions: + payload["exclude_conditions"] = exclude_conditions + + response = await client.post( + f"/v1/repos/{repository}/scheduled_freeze", + json=payload, + ) + return response.json() # type: ignore[no-any-return] + + +async def update_freeze( + client: httpx.AsyncClient, + repository: str, + freeze_id: uuid.UUID, + *, + reason: str, + timezone: str, + matching_conditions: list[str], + start: datetime.datetime | None = None, + end: datetime.datetime | None = None, + exclude_conditions: list[str] | None = None, +) -> dict[str, typing.Any]: + payload: dict[str, typing.Any] = { + "reason": reason, + "timezone": timezone, + "matching_conditions": matching_conditions, + } + if start is not None: + payload["start"] = start.isoformat() + if end is not None: + payload["end"] = end.isoformat() + if exclude_conditions is not None: + payload["exclude_conditions"] = exclude_conditions + + response = await client.patch( + f"/v1/repos/{repository}/scheduled_freeze/{freeze_id}", + json=payload, + ) + return response.json() # type: ignore[no-any-return] + + +async def delete_freeze( + client: httpx.AsyncClient, + repository: str, + freeze_id: uuid.UUID, + *, + delete_reason: str | None = None, +) -> None: + url = f"/v1/repos/{repository}/scheduled_freeze/{freeze_id}" + if delete_reason is not None: + await client.request( + "DELETE", + url, + json={"delete_reason": delete_reason}, + ) + else: + await client.delete(url) diff --git a/mergify_cli/freeze/cli.py b/mergify_cli/freeze/cli.py new file mode 100644 index 00000000..b97ac37b --- /dev/null +++ b/mergify_cli/freeze/cli.py @@ -0,0 +1,389 @@ +from __future__ import annotations + +import asyncio +import datetime +import os +import typing + +import click +from rich.table import Table + +from mergify_cli import console +from mergify_cli import utils +from mergify_cli.freeze import api as freeze_api + + +if typing.TYPE_CHECKING: + import uuid + + +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 utils.run_command("gh", "auth", "token") + except utils.CommandError: + console.print( + "error: please set the 'MERGIFY_TOKEN' or 'GITHUB_TOKEN' environment variable, " + "or make sure that gh client is installed and you are authenticated", + style="red", + ) + 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 utils.git( + "config", + "--get", + "remote.origin.url", + ) + except utils.CommandError: + return None + + try: + user, repo_name = utils.get_slug(remote_url) + except (ValueError, IndexError): + return None + + return f"{user}/{repo_name}" + + +def _parse_naive_datetime(value: str) -> datetime.datetime: + try: + return datetime.datetime.fromisoformat(value) + except ValueError: + msg = f"Invalid datetime format: {value!r}. Use ISO 8601 format (e.g. 2024-06-19T08:00:00)" + raise click.BadParameter(msg) from None + + +class NaiveDateTimeType(click.ParamType): + name = "DATETIME" + + def convert( + self, + value: str | datetime.datetime, + param: click.Parameter | None, # noqa: ARG002 + ctx: click.Context | None, # noqa: ARG002 + ) -> datetime.datetime: + if isinstance(value, datetime.datetime): + return value + return _parse_naive_datetime(value) + + +NAIVE_DATETIME = NaiveDateTimeType() + + +def _format_datetime( + value: str | None, + timezone: str, +) -> str: + if value is None: + return "-" + return f"{value} ({timezone})" + + +def _is_active(freeze: dict[str, typing.Any]) -> bool: + start = freeze.get("start") + if start is None: + return False + start_dt = datetime.datetime.fromisoformat(str(start)) + # NOTE: start is naive in the freeze's timezone, but we don't know the + # server's current time in that timezone. We display the status based on + # UTC as a best-effort approximation. + return start_dt <= datetime.datetime.now(tz=datetime.UTC).replace(tzinfo=None) + + +def _print_freeze_table(freezes: list[dict[str, typing.Any]]) -> None: + if not freezes: + console.print("No scheduled freezes found.") + return + + table = Table(title="Scheduled Freezes") + table.add_column("ID", style="dim") + table.add_column("Reason") + table.add_column("Start") + table.add_column("End") + table.add_column("Conditions") + table.add_column("Status") + + for freeze in freezes: + timezone = str(freeze.get("timezone", "")) + active = _is_active(freeze) + conditions = ", ".join( + str(c) for c in (freeze.get("matching_conditions") or []) + ) + exclude = freeze.get("exclude_conditions") or [] + if exclude: + conditions += f" (exclude: {', '.join(str(c) for c in exclude)})" + + table.add_row( + str(freeze.get("id", "")), + str(freeze.get("reason", "")), + _format_datetime( + str(freeze["start"]) if freeze.get("start") else None, + timezone, + ), + _format_datetime( + str(freeze["end"]) if freeze.get("end") else None, + timezone, + ), + conditions, + "[green]active[/]" if active else "[yellow]scheduled[/]", + ) + + console.print(table) + + +def _print_freeze(freeze: dict[str, typing.Any]) -> None: + timezone = str(freeze.get("timezone", "")) + console.print(f" ID: {freeze.get('id')}") + console.print(f" Reason: {freeze.get('reason')}") + console.print( + f" Start: {_format_datetime(str(freeze['start']) if freeze.get('start') else None, timezone)}", + ) + console.print( + f" End: {_format_datetime(str(freeze['end']) if freeze.get('end') else None, timezone)}", + ) + conditions = ", ".join(str(c) for c in (freeze.get("matching_conditions") or [])) + console.print(f" Conditions: {conditions}") + exclude = freeze.get("exclude_conditions") or [] + if exclude: + console.print( + f" Exclude: {', '.join(str(c) for c in exclude)}", + ) + + +@click.group( + invoke_without_command=True, + help="Manage scheduled freezes", +) +@click.option( + "--token", + "-t", + help="Mergify or GitHub token", + envvar=["MERGIFY_TOKEN", "GITHUB_TOKEN"], + required=True, + default=lambda: asyncio.run(_get_default_token()), +) +@click.option( + "--api-url", + "-u", + help="URL of the Mergify API", + envvar="MERGIFY_API_URL", + default="https://api.mergify.com", + show_default=True, +) +@click.option( + "--repository", + "-r", + help="Repository full name (owner/repo)", + required=True, + default=lambda: asyncio.run(_get_default_repository()), +) +@click.pass_context +def freeze( + ctx: click.Context, + *, + token: str, + api_url: str, + repository: str, +) -> None: + ctx.ensure_object(dict) + ctx.obj["token"] = token + ctx.obj["api_url"] = api_url + ctx.obj["repository"] = repository + if ctx.invoked_subcommand is None: + click.echo(ctx.get_help()) + + +@freeze.command(name="list", help="List scheduled freezes for a repository") +@click.option( + "--json", + "output_json", + is_flag=True, + help="Output in JSON format", +) +@click.pass_context +@utils.run_with_asyncio +async def list_cmd(ctx: click.Context, *, output_json: bool) -> None: + import json + + async with utils.get_mergify_http_client( + ctx.obj["api_url"], + ctx.obj["token"], + ) as client: + freezes = await freeze_api.list_freezes(client, ctx.obj["repository"]) + + if output_json: + click.echo(json.dumps(freezes, indent=2)) + else: + _print_freeze_table(freezes) + + +@freeze.command(help="Create a new scheduled freeze") +@click.option("--reason", required=True, help="Reason for the freeze") +@click.option( + "--timezone", + required=True, + help="IANA timezone name (e.g. Europe/Paris, US/Eastern)", +) +@click.option( + "--condition", + "-c", + "conditions", + multiple=True, + required=True, + help="Matching condition (repeatable, e.g. -c 'base=main')", +) +@click.option( + "--start", + type=NAIVE_DATETIME, + default=None, + help="Start time in ISO 8601 format (default: now)", +) +@click.option( + "--end", + type=NAIVE_DATETIME, + default=None, + help="End time in ISO 8601 format (default: no end / emergency freeze)", +) +@click.option( + "--exclude", + "-e", + "excludes", + multiple=True, + help="Exclude condition (repeatable, e.g. -e 'label=hotfix')", +) +@click.pass_context +@utils.run_with_asyncio +async def create( + ctx: click.Context, + *, + reason: str, + timezone: str, + conditions: tuple[str, ...], + start: datetime.datetime | None, + end: datetime.datetime | None, + excludes: tuple[str, ...], +) -> None: + async with utils.get_mergify_http_client( + ctx.obj["api_url"], + ctx.obj["token"], + ) as client: + result = await freeze_api.create_freeze( + client, + ctx.obj["repository"], + reason=reason, + timezone=timezone, + matching_conditions=list(conditions), + start=start, + end=end, + exclude_conditions=list(excludes) if excludes else None, + ) + + console.print("[green]Freeze created successfully:[/]") + _print_freeze(result) + + +@freeze.command(help="Update an existing scheduled freeze") +@click.argument("freeze_id", type=click.UUID) +@click.option("--reason", required=True, help="Reason for the freeze") +@click.option( + "--timezone", + required=True, + help="IANA timezone name (e.g. Europe/Paris, US/Eastern)", +) +@click.option( + "--condition", + "-c", + "conditions", + multiple=True, + required=True, + help="Matching condition (repeatable, e.g. -c 'base=main')", +) +@click.option( + "--start", + type=NAIVE_DATETIME, + default=None, + help="Start time in ISO 8601 format", +) +@click.option( + "--end", + type=NAIVE_DATETIME, + default=None, + help="End time in ISO 8601 format", +) +@click.option( + "--exclude", + "-e", + "excludes", + multiple=True, + help="Exclude condition (repeatable, e.g. -e 'label=hotfix')", +) +@click.pass_context +@utils.run_with_asyncio +async def update( + ctx: click.Context, + *, + freeze_id: uuid.UUID, + reason: str, + timezone: str, + conditions: tuple[str, ...], + start: datetime.datetime | None, + end: datetime.datetime | None, + excludes: tuple[str, ...], +) -> None: + async with utils.get_mergify_http_client( + ctx.obj["api_url"], + ctx.obj["token"], + ) as client: + result = await freeze_api.update_freeze( + client, + ctx.obj["repository"], + freeze_id, + reason=reason, + timezone=timezone, + matching_conditions=list(conditions), + start=start, + end=end, + exclude_conditions=list(excludes) if excludes else None, + ) + + console.print("[green]Freeze updated successfully:[/]") + _print_freeze(result) + + +@freeze.command(help="Delete a scheduled freeze") +@click.argument("freeze_id", type=click.UUID) +@click.option( + "--reason", + "delete_reason", + default=None, + help="Reason for deleting the freeze (required if freeze is active)", +) +@click.pass_context +@utils.run_with_asyncio +async def delete( + ctx: click.Context, + *, + freeze_id: uuid.UUID, + delete_reason: str | None, +) -> None: + async with utils.get_mergify_http_client( + ctx.obj["api_url"], + ctx.obj["token"], + ) as client: + await freeze_api.delete_freeze( + client, + ctx.obj["repository"], + freeze_id, + delete_reason=delete_reason, + ) + + console.print("[green]Freeze deleted successfully.[/]") diff --git a/mergify_cli/tests/freeze/__init__.py b/mergify_cli/tests/freeze/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/mergify_cli/tests/freeze/test_cli.py b/mergify_cli/tests/freeze/test_cli.py new file mode 100644 index 00000000..c4aed7d9 --- /dev/null +++ b/mergify_cli/tests/freeze/test_cli.py @@ -0,0 +1,425 @@ +from __future__ import annotations + +import datetime +import json + +import click +from click.testing import CliRunner +from httpx import Response +import pytest +import respx + +from mergify_cli.freeze.cli import freeze + + +FAKE_FREEZE = { + "id": "550e8400-e29b-41d4-a716-446655440000", + "reason": "Release prep", + "start": "2099-06-19T08:00:00", + "end": "2099-06-20T17:00:00", + "timezone": "Europe/Paris", + "matching_conditions": ["base=main"], + "exclude_conditions": [], +} + +FAKE_FREEZE_WITH_EXCLUDE = { + **FAKE_FREEZE, + "exclude_conditions": ["label=hotfix"], +} + +BASE_ARGS = [ + "--token", + "test-token", + "--api-url", + "https://api.mergify.com", + "--repository", + "owner/repo", +] + + +def test_list_empty() -> None: + with respx.mock(base_url="https://api.mergify.com") as mock: + mock.get("/v1/repos/owner/repo/scheduled_freeze").mock( + return_value=Response(200, json={"scheduled_freezes": []}), + ) + + runner = CliRunner() + result = runner.invoke(freeze, [*BASE_ARGS, "list"]) + assert result.exit_code == 0 + assert "No scheduled freezes found" in result.output + + +def test_list_with_freezes() -> None: + with respx.mock(base_url="https://api.mergify.com") as mock: + mock.get("/v1/repos/owner/repo/scheduled_freeze").mock( + return_value=Response( + 200, + json={"scheduled_freezes": [FAKE_FREEZE]}, + ), + ) + + runner = CliRunner() + result = runner.invoke(freeze, [*BASE_ARGS, "list"]) + assert result.exit_code == 0 + assert "Release" in result.output + assert "base=main" in result.output + assert "scheduled" in result.output + + +def test_list_json() -> None: + with respx.mock(base_url="https://api.mergify.com") as mock: + mock.get("/v1/repos/owner/repo/scheduled_freeze").mock( + return_value=Response( + 200, + json={"scheduled_freezes": [FAKE_FREEZE]}, + ), + ) + + runner = CliRunner() + result = runner.invoke(freeze, [*BASE_ARGS, "list", "--json"]) + assert result.exit_code == 0 + data = json.loads(result.output) + assert len(data) == 1 + assert data[0]["reason"] == "Release prep" + + +def test_list_with_exclude_conditions() -> None: + with respx.mock(base_url="https://api.mergify.com") as mock: + mock.get("/v1/repos/owner/repo/scheduled_freeze").mock( + return_value=Response( + 200, + json={"scheduled_freezes": [FAKE_FREEZE_WITH_EXCLUDE]}, + ), + ) + + runner = CliRunner() + result = runner.invoke(freeze, [*BASE_ARGS, "list"]) + assert result.exit_code == 0 + assert "exclude" in result.output + + +def test_list_json_with_exclude_conditions() -> None: + with respx.mock(base_url="https://api.mergify.com") as mock: + mock.get("/v1/repos/owner/repo/scheduled_freeze").mock( + return_value=Response( + 200, + json={"scheduled_freezes": [FAKE_FREEZE_WITH_EXCLUDE]}, + ), + ) + + runner = CliRunner() + result = runner.invoke(freeze, [*BASE_ARGS, "list", "--json"]) + assert result.exit_code == 0 + data = json.loads(result.output) + assert data[0]["exclude_conditions"] == ["label=hotfix"] + + +def test_create_minimal() -> None: + with respx.mock(base_url="https://api.mergify.com") as mock: + mock.post("/v1/repos/owner/repo/scheduled_freeze").mock( + return_value=Response(201, json=FAKE_FREEZE), + ) + + runner = CliRunner() + result = runner.invoke( + freeze, + [ + *BASE_ARGS, + "create", + "--reason", + "Release prep", + "--timezone", + "Europe/Paris", + "-c", + "base=main", + ], + ) + assert result.exit_code == 0, result.output + assert "Freeze created successfully" in result.output + assert "Release prep" in result.output + + request = mock.calls.last.request + body = json.loads(request.content) + assert body["reason"] == "Release prep" + assert body["timezone"] == "Europe/Paris" + assert body["matching_conditions"] == ["base=main"] + assert "start" not in body + assert "end" not in body + + +def test_create_with_all_options() -> None: + with respx.mock(base_url="https://api.mergify.com") as mock: + mock.post("/v1/repos/owner/repo/scheduled_freeze").mock( + return_value=Response(201, json=FAKE_FREEZE_WITH_EXCLUDE), + ) + + runner = CliRunner() + result = runner.invoke( + freeze, + [ + *BASE_ARGS, + "create", + "--reason", + "Release prep", + "--timezone", + "Europe/Paris", + "-c", + "base=main", + "--start", + "2099-06-19T08:00:00", + "--end", + "2099-06-20T17:00:00", + "-e", + "label=hotfix", + ], + ) + assert result.exit_code == 0, result.output + + request = mock.calls.last.request + body = json.loads(request.content) + assert body["start"] == "2099-06-19T08:00:00" + assert body["end"] == "2099-06-20T17:00:00" + assert body["exclude_conditions"] == ["label=hotfix"] + + +def test_create_multiple_conditions() -> None: + with respx.mock(base_url="https://api.mergify.com") as mock: + mock.post("/v1/repos/owner/repo/scheduled_freeze").mock( + return_value=Response(201, json=FAKE_FREEZE), + ) + + runner = CliRunner() + result = runner.invoke( + freeze, + [ + *BASE_ARGS, + "create", + "--reason", + "Multi-branch freeze", + "--timezone", + "UTC", + "-c", + "base=main", + "-c", + "base=release", + ], + ) + assert result.exit_code == 0, result.output + + request = mock.calls.last.request + body = json.loads(request.content) + assert body["matching_conditions"] == ["base=main", "base=release"] + + +def test_create_missing_required() -> None: + runner = CliRunner() + result = runner.invoke( + freeze, + [*BASE_ARGS, "create", "--reason", "test"], + ) + assert result.exit_code != 0 + + +def test_update() -> None: + freeze_id = "550e8400-e29b-41d4-a716-446655440000" + updated_freeze = {**FAKE_FREEZE, "reason": "Updated reason"} + + with respx.mock(base_url="https://api.mergify.com") as mock: + mock.patch( + f"/v1/repos/owner/repo/scheduled_freeze/{freeze_id}", + ).mock( + return_value=Response(200, json=updated_freeze), + ) + + runner = CliRunner() + result = runner.invoke( + freeze, + [ + *BASE_ARGS, + "update", + freeze_id, + "--reason", + "Updated reason", + "--timezone", + "Europe/Paris", + "-c", + "base=main", + ], + ) + assert result.exit_code == 0, result.output + assert "Freeze updated successfully" in result.output + assert "Updated reason" in result.output + + +def test_update_with_end() -> None: + freeze_id = "550e8400-e29b-41d4-a716-446655440000" + + with respx.mock(base_url="https://api.mergify.com") as mock: + mock.patch( + f"/v1/repos/owner/repo/scheduled_freeze/{freeze_id}", + ).mock( + return_value=Response(200, json=FAKE_FREEZE), + ) + + runner = CliRunner() + result = runner.invoke( + freeze, + [ + *BASE_ARGS, + "update", + freeze_id, + "--reason", + "Release prep", + "--timezone", + "Europe/Paris", + "-c", + "base=main", + "--end", + "2099-12-31T23:59:59", + ], + ) + assert result.exit_code == 0, result.output + + request = mock.calls.last.request + body = json.loads(request.content) + assert body["end"] == "2099-12-31T23:59:59" + + +def test_delete() -> None: + freeze_id = "550e8400-e29b-41d4-a716-446655440000" + + with respx.mock(base_url="https://api.mergify.com") as mock: + mock.delete( + f"/v1/repos/owner/repo/scheduled_freeze/{freeze_id}", + ).mock( + return_value=Response(204), + ) + + runner = CliRunner() + result = runner.invoke( + freeze, + [*BASE_ARGS, "delete", freeze_id], + ) + assert result.exit_code == 0, result.output + assert "Freeze deleted successfully" in result.output + + +def test_delete_with_reason() -> None: + freeze_id = "550e8400-e29b-41d4-a716-446655440000" + + with respx.mock(base_url="https://api.mergify.com") as mock: + mock.delete( + f"/v1/repos/owner/repo/scheduled_freeze/{freeze_id}", + ).mock( + return_value=Response(204), + ) + + runner = CliRunner() + result = runner.invoke( + freeze, + [ + *BASE_ARGS, + "delete", + freeze_id, + "--reason", + "Emergency rollback completed", + ], + ) + assert result.exit_code == 0, result.output + + request = mock.calls.last.request + body = json.loads(request.content) + assert body["delete_reason"] == "Emergency rollback completed" + + +def test_delete_without_reason_sends_no_body() -> None: + freeze_id = "550e8400-e29b-41d4-a716-446655440000" + + with respx.mock(base_url="https://api.mergify.com") as mock: + mock.delete( + f"/v1/repos/owner/repo/scheduled_freeze/{freeze_id}", + ).mock( + return_value=Response(204), + ) + + runner = CliRunner() + result = runner.invoke( + freeze, + [*BASE_ARGS, "delete", freeze_id], + ) + assert result.exit_code == 0, result.output + + request = mock.calls.last.request + assert request.content == b"" + + +def test_list_api_error() -> None: + with respx.mock(base_url="https://api.mergify.com") as mock: + mock.get("/v1/repos/owner/repo/scheduled_freeze").mock( + return_value=Response(403, json={"message": "Forbidden"}), + ) + + runner = CliRunner() + result = runner.invoke(freeze, [*BASE_ARGS, "list"]) + assert result.exit_code != 0 + + +def test_create_api_error() -> None: + with respx.mock(base_url="https://api.mergify.com") as mock: + mock.post("/v1/repos/owner/repo/scheduled_freeze").mock( + return_value=Response( + 422, + json={"message": "end must be after start"}, + ), + ) + + runner = CliRunner() + result = runner.invoke( + freeze, + [ + *BASE_ARGS, + "create", + "--reason", + "test", + "--timezone", + "UTC", + "-c", + "base=main", + ], + ) + assert result.exit_code != 0 + + +def test_repository_from_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("GITHUB_REPOSITORY", "env-owner/env-repo") + + import asyncio + + from mergify_cli.freeze.cli import _get_default_repository + + result = asyncio.run(_get_default_repository()) + assert result == "env-owner/env-repo" + + +class TestNaiveDateTimeType: + def test_valid_datetime(self) -> None: + from mergify_cli.freeze.cli import NAIVE_DATETIME + + result = NAIVE_DATETIME.convert("2024-06-19T08:00:00", None, None) + assert result.year == 2024 + assert result.month == 6 + assert result.day == 19 + assert result.hour == 8 + + def test_invalid_datetime(self) -> None: + from mergify_cli.freeze.cli import NAIVE_DATETIME + + with pytest.raises(click.BadParameter, match="Invalid datetime format"): + NAIVE_DATETIME.convert("not-a-date", None, None) + + def test_passthrough_datetime(self) -> None: + from mergify_cli.freeze.cli import NAIVE_DATETIME + + dt = datetime.datetime(2024, 6, 19, 8, 0, 0, tzinfo=datetime.UTC) + result = NAIVE_DATETIME.convert(dt, None, None) + assert result is dt