diff --git a/src/fastapi_cloud_cli/commands/tokens/__init__.py b/src/fastapi_cloud_cli/commands/tokens/__init__.py index 3e5357d..f69c6c4 100644 --- a/src/fastapi_cloud_cli/commands/tokens/__init__.py +++ b/src/fastapi_cloud_cli/commands/tokens/__init__.py @@ -1,11 +1,13 @@ import typer +from fastapi_cloud_cli.commands.tokens.create import create_token from fastapi_cloud_cli.commands.tokens.list import list_tokens tokens_app = typer.Typer( no_args_is_help=True, help="Manage deploy tokens for your app.", ) +tokens_app.command("create")(create_token) tokens_app.command("list")(list_tokens) __all__ = ["tokens_app"] diff --git a/src/fastapi_cloud_cli/commands/tokens/create.py b/src/fastapi_cloud_cli/commands/tokens/create.py new file mode 100644 index 0000000..16bfadc --- /dev/null +++ b/src/fastapi_cloud_cli/commands/tokens/create.py @@ -0,0 +1,182 @@ +from pathlib import Path +from typing import Annotated, Any, Literal + +import typer +from pydantic import BaseModel, Field +from rich_toolkit import RichToolkit + +from fastapi_cloud_cli.utils.api import APIClient +from fastapi_cloud_cli.utils.apps import resolve_app_id_or_fail +from fastapi_cloud_cli.utils.auth import Identity +from fastapi_cloud_cli.utils.cli import FastAPIRichToolkit, get_rich_toolkit +from fastapi_cloud_cli.utils.execution import JsonOutputOption + +DEFAULT_EXPIRES_IN_DAYS = 365 + + +class CreatedDeployToken(BaseModel): + id: str + name: str + expired_at: str + + +class DeployTokenCreateAPIResponse(CreatedDeployToken): + value: str + + +class StoredDeployTokenSecret(BaseModel): + provider: Literal["file"] = "file" + path: Path + + +class DeployTokenCreateOutput(BaseModel): + app_id: str + token: CreatedDeployToken + stored_secret: StoredDeployTokenSecret + output_file: Annotated[Path, Field(exclude=True)] + + +def _resolve_token_name(toolkit: FastAPIRichToolkit, *, name: str | None) -> str: + if name is not None: + return name + + if toolkit.mode == "json": + toolkit.fail( + "missing_required_input", + "Deploy token name is required.", + hint="Pass --name to choose a deploy token name.", + ) + + return toolkit.input( + "What's the deploy token name?", + default="Deploy token", + bullet=False, + ) + + +def _resolve_output_file( + toolkit: FastAPIRichToolkit, *, output_file: Path | None +) -> Path: + if output_file is not None: + return output_file + + toolkit.fail( + "missing_required_input", + "Output file is required.", + hint="Pass --output-file to store the deploy token value.", + ) + + +def _create_deploy_token( + client: APIClient, *, app_id: str, name: str, expires_in_days: int +) -> DeployTokenCreateAPIResponse: + response = client.post( + f"/apps/{app_id}/tokens", + json={"name": name, "expires_in_days": expires_in_days}, + ) + response.raise_for_status() + + return DeployTokenCreateAPIResponse.model_validate(response.json()) + + +def _write_token_value(output_file: Path, value: str) -> None: + output_file.parent.mkdir(parents=True, exist_ok=True) + output_file.write_text(value, encoding="utf-8") + output_file.chmod(0o600) + + +def _render_deploy_token_create_output( + data: DeployTokenCreateOutput, toolkit: RichToolkit +) -> None: + toolkit.print(f"Created deploy token [bold]{data.token.name}[/bold]", bullet=False) + toolkit.print( + f"Stored deploy token value in [bold]{data.output_file}[/bold]", + bullet=False, + ) + + +def create_token( + app_id: Annotated[ + str | None, + typer.Option( + "--app-id", + help="ID of the app whose deploy token should be created.", + ), + ] = None, + name: Annotated[ + str | None, + typer.Option( + "--name", + help="Name of the deploy token to create.", + ), + ] = None, + expires_in_days: Annotated[ + int, + typer.Option( + "--expires-in-days", + help="Number of days before the deploy token expires.", + min=1, + ), + ] = DEFAULT_EXPIRES_IN_DAYS, + output_file: Annotated[ + Path | None, + typer.Option( + "--output-file", + help="File path where the deploy token value should be stored.", + ), + ] = None, + json_output: JsonOutputOption = False, +) -> Any: + """ + Create a deploy token for an app. + """ + identity = Identity() + + with get_rich_toolkit(json_output=json_output) as toolkit: + if not identity.is_logged_in(): + toolkit.fail( + "not_logged_in", + "No credentials found.", + hint="Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.", + ) + + target_app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) + + toolkit.print_title("deploy tokens") + toolkit.print_line() + + output_file = _resolve_output_file(toolkit, output_file=output_file) + name_needs_prompt = name is None + name = _resolve_token_name(toolkit, name=name) + if name_needs_prompt: + toolkit.print_line() + + with APIClient() as client: + with toolkit.progress( + title="Creating deploy token", + transient=True, + ) as progress: + with client.handle_http_errors( + progress, + default_message="Error creating deploy token. Please try again later.", + not_found_message="App not found.", + toolkit=toolkit, + ): + token = _create_deploy_token( + client, + app_id=target_app_id, + name=name, + expires_in_days=expires_in_days, + ) + + _write_token_value(output_file, token.value) + + toolkit.success( + DeployTokenCreateOutput( + app_id=target_app_id, + token=CreatedDeployToken.model_validate(token), + stored_secret=StoredDeployTokenSecret(path=output_file), + output_file=output_file, + ), + render_output=_render_deploy_token_create_output, + ) diff --git a/src/fastapi_cloud_cli/utils/api.py b/src/fastapi_cloud_cli/utils/api.py index 385042b..3a4a5e2 100644 --- a/src/fastapi_cloud_cli/utils/api.py +++ b/src/fastapi_cloud_cli/utils/api.py @@ -241,6 +241,9 @@ def handle_http_error( elif status_code == 400: message = _get_response_error_message(error.response) + elif status_code == 409: + message = _get_response_error_message(error.response) + elif status_code == 401: message = _handle_unauthorized(auth_mode=auth_mode) @@ -273,7 +276,7 @@ def get_http_error_code(error: httpx.HTTPError) -> ErrorCode: if isinstance(error, httpx.HTTPStatusError): status_code = error.response.status_code - if status_code == 400: + if status_code in {400, 409}: return "invalid_input" if status_code == 401: diff --git a/src/fastapi_cloud_cli/utils/cli.py b/src/fastapi_cloud_cli/utils/cli.py index 4b81b88..0c3f5f1 100644 --- a/src/fastapi_cloud_cli/utils/cli.py +++ b/src/fastapi_cloud_cli/utils/cli.py @@ -253,9 +253,12 @@ def _animate_title_sweep(self, text: str) -> None: self.console.show_cursor(True) def _get_progress_status_emoji(self, element: Progress, done: bool) -> str: - if element._cancelled or element.is_error: + if element._cancelled: return "🟡" + if element.is_error: + return ERROR_BULLET + if done: return cast(str, element.metadata.get("done_emoji", "🐔")) diff --git a/tests/test_cli.py b/tests/test_cli.py index 97981ef..ef28234 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -206,7 +206,8 @@ def test_fastapi_style_progress_status_emoji_states() -> None: progress._cancelled = False progress.is_error = True - assert style._get_progress_status_emoji(progress, done=False) == "🟡" + assert style._get_progress_status_emoji(progress, done=False) == ERROR_BULLET + assert render_plain(style.render_element(progress)) == " ✗ Working\n" def test_fastapi_style_gets_cursor_offset_with_and_without_bullet_column() -> None: diff --git a/tests/test_cli_tokens.py b/tests/test_cli_tokens.py index 8ef942e..9b68af0 100644 --- a/tests/test_cli_tokens.py +++ b/tests/test_cli_tokens.py @@ -1,4 +1,6 @@ import json +from pathlib import Path +from unittest.mock import patch import pytest import respx @@ -7,11 +9,398 @@ from fastapi_cloud_cli.cli import cloud_app as app from tests.conftest import ConfiguredApp -from tests.utils import changing_dir +from tests.utils import Keys, changing_dir runner = CliRunner() +def test_creates_token_json_returns_not_logged_in_when_logged_out( + logged_out_cli: None, +) -> None: + result = runner.invoke( + app, + [ + "tokens", + "create", + "--app-id", + "00000000-0000-4000-8000-000000000002", + "--name", + "GitHub Actions", + "--output-file", + "deploy-token.txt", + "--json", + ], + ) + + assert result.exit_code == 1 + assert json.loads(result.stdout) == { + "error": { + "code": "not_logged_in", + "message": "No credentials found.", + "hint": "Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.", + } + } + assert result.stderr == "" + + +def test_creates_token_json_returns_missing_required_input_without_name( + logged_in_cli: None, +) -> None: + result = runner.invoke( + app, + [ + "tokens", + "create", + "--app-id", + "00000000-0000-4000-8000-000000000002", + "--output-file", + "deploy-token.txt", + "--json", + ], + ) + + assert result.exit_code == 1 + assert json.loads(result.stdout) == { + "error": { + "code": "missing_required_input", + "message": "Deploy token name is required.", + "hint": "Pass --name to choose a deploy token name.", + } + } + assert result.stderr == "" + + +def test_creates_token_json_returns_missing_required_input_without_output_file( + logged_in_cli: None, +) -> None: + result = runner.invoke( + app, + [ + "tokens", + "create", + "--app-id", + "00000000-0000-4000-8000-000000000002", + "--name", + "GitHub Actions", + "--json", + ], + ) + + assert result.exit_code == 1 + assert json.loads(result.stdout) == { + "error": { + "code": "missing_required_input", + "message": "Output file is required.", + "hint": "Pass --output-file to store the deploy token value.", + } + } + assert result.stderr == "" + + +def test_create_token_requires_output_file_before_prompting_for_name( + logged_in_cli: None, +) -> None: + result = runner.invoke( + app, + [ + "tokens", + "create", + "--app-id", + "00000000-0000-4000-8000-000000000002", + ], + ) + + assert result.exit_code == 1 + assert "deploy tokens" in result.output + assert "Output file is required." in result.output + assert "Pass --output-file to store the deploy token value." in result.output + assert "What's the deploy token name?" not in result.output + + +@pytest.mark.respx +def test_creates_token_as_json_with_app_id_and_writes_value_to_file( + logged_in_cli: None, + respx_mock: respx.MockRouter, + tmp_path: Path, +) -> None: + app_id = "00000000-0000-4000-8000-000000000002" + token_id = "00000000-0000-4000-8000-000000000004" + token_value = "fcp_secret_token_value" + output_file = tmp_path / "deploy-token" + respx_mock.post( + f"/apps/{app_id}/tokens", + json={"name": "GitHub Actions", "expires_in_days": 365}, + ).mock( + return_value=Response( + 201, + json={ + "id": token_id, + "name": "GitHub Actions", + "expired_at": "2027-05-22T10:00:00Z", + "value": token_value, + }, + ) + ) + + result = runner.invoke( + app, + [ + "tokens", + "create", + "--app-id", + app_id, + "--name", + "GitHub Actions", + "--output-file", + str(output_file), + "--json", + ], + ) + + assert result.exit_code == 0 + assert json.loads(result.stdout) == { + "data": { + "app_id": app_id, + "token": { + "id": token_id, + "name": "GitHub Actions", + "expired_at": "2027-05-22T10:00:00Z", + }, + "stored_secret": { + "provider": "file", + "path": str(output_file), + }, + } + } + assert token_value not in result.stdout + assert output_file.read_text(encoding="utf-8") == token_value + assert result.stderr == "" + + +@pytest.mark.respx +def test_creates_token_uses_linked_app_and_custom_expiration( + logged_in_cli: None, + respx_mock: respx.MockRouter, + configured_app: ConfiguredApp, + tmp_path: Path, +) -> None: + output_file = tmp_path / "deploy-token" + respx_mock.post( + f"/apps/{configured_app.app_id}/tokens", + json={"name": "Release Bot", "expires_in_days": 30}, + ).mock( + return_value=Response( + 201, + json={ + "id": "00000000-0000-4000-8000-000000000004", + "name": "Release Bot", + "expired_at": "2026-06-21T10:00:00Z", + "value": "fcp_secret_token_value", + }, + ) + ) + + with changing_dir(configured_app.path): + result = runner.invoke( + app, + [ + "tokens", + "create", + "--name", + "Release Bot", + "--expires-in-days", + "30", + "--output-file", + str(output_file), + "--json", + ], + ) + + assert result.exit_code == 0 + assert json.loads(result.stdout)["data"]["app_id"] == configured_app.app_id + assert output_file.read_text(encoding="utf-8") == "fcp_secret_token_value" + assert result.stderr == "" + + +@pytest.mark.respx +def test_create_token_prompts_for_name_under_deploy_tokens_heading( + logged_in_cli: None, + respx_mock: respx.MockRouter, + tmp_path: Path, +) -> None: + app_id = "00000000-0000-4000-8000-000000000002" + output_file = tmp_path / "deploy-token" + respx_mock.post( + f"/apps/{app_id}/tokens", + json={"name": "GitHub Actions", "expires_in_days": 365}, + ).mock( + return_value=Response( + 201, + json={ + "id": "00000000-0000-4000-8000-000000000004", + "name": "GitHub Actions", + "expired_at": "2027-05-22T10:00:00Z", + "value": "fcp_secret_token_value", + }, + ) + ) + + with patch( + "rich_toolkit.container.getchar", + side_effect=[*"GitHub Actions", Keys.ENTER], + ): + result = runner.invoke( + app, + [ + "tokens", + "create", + "--app-id", + app_id, + "--output-file", + str(output_file), + ], + ) + + assert result.exit_code == 0 + assert "deploy tokens" in result.output + assert "What's the deploy token name?" in result.output + assert "Created deploy token GitHub Actions" in result.output + assert output_file.read_text(encoding="utf-8") == "fcp_secret_token_value" + + +@pytest.mark.respx +def test_create_token_separates_prompt_from_http_error( + logged_in_cli: None, + respx_mock: respx.MockRouter, + tmp_path: Path, +) -> None: + app_id = "00000000-0000-4000-8000-000000000002" + output_file = tmp_path / "deploy-token" + respx_mock.post( + f"/apps/{app_id}/tokens", + json={"name": "GitHub Actions", "expires_in_days": 365}, + ).mock( + return_value=Response( + 409, + json={"detail": "Deploy token name already exists."}, + ) + ) + + with patch( + "rich_toolkit.container.getchar", + side_effect=[*"GitHub Actions", Keys.ENTER], + ): + result = runner.invoke( + app, + [ + "tokens", + "create", + "--app-id", + app_id, + "--output-file", + str(output_file), + ], + ) + + assert result.exit_code == 1 + assert "What's the deploy token name?" in result.output + assert "Deploy token name already exists." in result.output + lines = result.output.splitlines() + error_index = next( + index + for index, line in enumerate(lines) + if "Deploy token name already exists." in line + ) + assert lines[error_index - 1].replace("\u200b", "").strip() == "" + assert not output_file.exists() + + +@pytest.mark.respx +def test_create_token_human_output_does_not_print_token_value( + logged_in_cli: None, + respx_mock: respx.MockRouter, + tmp_path: Path, +) -> None: + app_id = "00000000-0000-4000-8000-000000000002" + token_value = "fcp_secret_token_value" + output_file = tmp_path / "deploy-token" + respx_mock.post(f"/apps/{app_id}/tokens").mock( + return_value=Response( + 201, + json={ + "id": "00000000-0000-4000-8000-000000000004", + "name": "GitHub Actions", + "expired_at": "2027-05-22T10:00:00Z", + "value": token_value, + }, + ) + ) + + result = runner.invoke( + app, + [ + "tokens", + "create", + "--app-id", + app_id, + "--name", + "GitHub Actions", + "--output-file", + str(output_file), + ], + ) + + assert result.exit_code == 0 + assert "deploy tokens" in result.output + assert "Created deploy token GitHub Actions" in result.output + assert "Stored deploy token value in" in result.output + assert output_file.name in result.output + assert token_value not in result.output + assert output_file.read_text(encoding="utf-8") == token_value + + +@pytest.mark.respx +def test_create_token_json_maps_duplicate_name_to_invalid_input( + logged_in_cli: None, + respx_mock: respx.MockRouter, + tmp_path: Path, +) -> None: + app_id = "00000000-0000-4000-8000-000000000002" + output_file = tmp_path / "deploy-token" + respx_mock.post(f"/apps/{app_id}/tokens").mock( + return_value=Response( + 409, + json={"detail": "Deploy token name already exists."}, + ) + ) + + result = runner.invoke( + app, + [ + "tokens", + "create", + "--app-id", + app_id, + "--name", + "GitHub Actions", + "--output-file", + str(output_file), + "--json", + ], + ) + + assert result.exit_code == 1 + assert json.loads(result.stdout) == { + "error": { + "code": "invalid_input", + "message": "Deploy token name already exists.", + "hint": None, + } + } + assert not output_file.exists() + assert result.stderr == "" + + def test_lists_tokens_json_returns_not_logged_in_when_logged_out( logged_out_cli: None, ) -> None: