diff --git a/src/fastapi_cloud_cli/commands/integrations/resources/__init__.py b/src/fastapi_cloud_cli/commands/integrations/resources/__init__.py index b556c200..2be51f49 100644 --- a/src/fastapi_cloud_cli/commands/integrations/resources/__init__.py +++ b/src/fastapi_cloud_cli/commands/integrations/resources/__init__.py @@ -1,5 +1,6 @@ import typer +from fastapi_cloud_cli.commands.integrations.resources.connect import connect_resource from fastapi_cloud_cli.commands.integrations.resources.get import get_resource from fastapi_cloud_cli.commands.integrations.resources.list import list_resources @@ -7,6 +8,7 @@ no_args_is_help=True, help="Manage resources connected to an app.", ) +resources_app.command("connect")(connect_resource) resources_app.command("get")(get_resource) resources_app.command("list")(list_resources) diff --git a/src/fastapi_cloud_cli/commands/integrations/resources/connect.py b/src/fastapi_cloud_cli/commands/integrations/resources/connect.py new file mode 100644 index 00000000..98963950 --- /dev/null +++ b/src/fastapi_cloud_cli/commands/integrations/resources/connect.py @@ -0,0 +1,155 @@ +import logging +from typing import Annotated, Any + +import typer +from pydantic import BaseModel, Field +from rich.text import Text +from rich_toolkit import RichToolkit + +from fastapi_cloud_cli.commands.apps.list import ( + App, + _get_app, + _get_app_dashboard_url, + _get_team, +) +from fastapi_cloud_cli.config import Settings +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 get_rich_toolkit +from fastapi_cloud_cli.utils.execution import JsonOutputOption + +logger = logging.getLogger(__name__) + + +class ResourceConnectOutput(BaseModel): + app_id: str + connect_url: str + app_name: Annotated[str, Field(exclude=True)] + browser_opened: Annotated[bool, Field(exclude=True)] + app_id_was_provided: Annotated[bool, Field(exclude=True)] + + +def _get_resource_connect_url( + app: App, + *, + team_slug: str, + settings: Settings, +) -> str: + app_dashboard_url = _get_app_dashboard_url( + app, + team_slug=team_slug, + settings=settings, + ) + return f"{app_dashboard_url}/integrations/connect" + + +def _render_resource_connect_output( + data: ResourceConnectOutput, + toolkit: RichToolkit, +) -> None: + toolkit.print_title("connect resource") + toolkit.print_line() + + if data.browser_opened: + toolkit.print( + f"Opened the integration setup for [bold]{data.app_name}[/bold] in your browser.", + emoji="🔌", + ) + else: + toolkit.print( + f"Open the integration setup for [bold]{data.app_name}[/bold] in your browser:", + emoji="🔌", + ) + + toolkit.print(Text(data.connect_url, style=f"link {data.connect_url}")) + toolkit.print_line() + + list_command = "fastapi cloud integrations resources list" + if data.app_id_was_provided: + list_command = f"{list_command} --app-id {data.app_id}" + + toolkit.print( + f"When the connection is complete, run [bold]{list_command}[/bold] to view it.", + emoji="💡", + ) + + +def connect_resource( + app_id: Annotated[ + str | None, + typer.Option( + "--app-id", + help="ID of the app to connect a provider resource to.", + ), + ] = None, + no_open: Annotated[ + bool, + typer.Option( + "--no-open", + help="Do not open the browser automatically.", + ), + ] = False, + json_output: JsonOutputOption = False, +) -> Any: + """ + Open FastAPI Cloud to connect a provider resource to 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.", + ) + + app_id_was_provided = app_id is not None + app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) + + with APIClient() as client: + with ( + toolkit.progress(title="Fetching app", transient=True) as progress, + client.handle_http_errors( + progress, + default_message="Error fetching app. Please try again later.", + not_found_message="App not found.", + toolkit=toolkit, + ), + ): + app = _get_app(client, app_id) + + with ( + toolkit.progress(title="Fetching team", transient=True) as progress, + client.handle_http_errors( + progress, + default_message="Error fetching team. Please try again later.", + not_found_message="Team not found.", + toolkit=toolkit, + ), + ): + team = _get_team(client, app.team_id) + + connect_url = _get_resource_connect_url( + app, + team_slug=team.slug, + settings=Settings.get(), + ) + browser_opened = False + + if not json_output and not no_open: + launch_result = typer.launch(connect_url) + logger.debug("Launch command result: %s", launch_result) + browser_opened = launch_result == 0 + + toolkit.success( + ResourceConnectOutput( + app_id=app.id, + app_name=app.name, + connect_url=connect_url, + browser_opened=browser_opened, + app_id_was_provided=app_id_was_provided, + ), + render_output=_render_resource_connect_output, + ) diff --git a/tests/integrations/resources/test_connect.py b/tests/integrations/resources/test_connect.py new file mode 100644 index 00000000..b4e4fe85 --- /dev/null +++ b/tests/integrations/resources/test_connect.py @@ -0,0 +1,221 @@ +import json +from pathlib import Path +from unittest.mock import patch + +import pytest +import respx +from httpx import Response +from typer.testing import CliRunner + +from fastapi_cloud_cli.cli import cloud_app as app +from fastapi_cloud_cli.config import Settings +from fastapi_cloud_cli.utils.apps import AppConfig, write_app_config +from tests.utils import changing_dir + +runner = CliRunner() + +APP_ID = "00000000-0000-4000-8000-000000000001" +LINKED_APP_ID = "00000000-0000-4000-8000-000000000002" +TEAM_ID = "00000000-0000-4000-8000-000000000003" +TEAM_SLUG = "acme" +APP_SLUG = "api" +CONNECT_URL = ( + f"{Settings().dashboard_base_url}/{TEAM_SLUG}/apps/{APP_SLUG}/integrations/connect" +) + + +def _app(*, app_id: str = APP_ID) -> dict[str, object]: + return { + "id": app_id, + "team_id": TEAM_ID, + "slug": APP_SLUG, + "name": "API", + "directory": None, + } + + +def _team() -> dict[str, str]: + return { + "id": TEAM_ID, + "slug": TEAM_SLUG, + "name": "Acme", + } + + +def _mock_app_and_team( + respx_mock: respx.MockRouter, + *, + app_id: str = APP_ID, +) -> None: + respx_mock.get(f"/apps/{app_id}").mock( + return_value=Response(200, json=_app(app_id=app_id)) + ) + respx_mock.get(f"/teams/{TEAM_ID}").mock(return_value=Response(200, json=_team())) + + +def test_connect_resource_json_returns_not_logged_in_when_logged_out( + logged_out_cli: None, +) -> None: + result = runner.invoke( + app, + ["integrations", "resources", "connect", "--app-id", APP_ID, "--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 == "" + + +@pytest.mark.respx +def test_connect_resource_json_returns_url_without_opening_browser( + logged_in_cli: None, + respx_mock: respx.MockRouter, + tmp_path: Path, +) -> None: + write_app_config( + tmp_path, + AppConfig(app_id=LINKED_APP_ID, team_id=TEAM_ID), + ) + _mock_app_and_team(respx_mock) + + with ( + changing_dir(tmp_path), + patch( + "fastapi_cloud_cli.commands.integrations.resources.connect.typer.launch" + ) as mock_launch, + ): + result = runner.invoke( + app, + [ + "integrations", + "resources", + "connect", + "--app-id", + APP_ID, + "--json", + ], + ) + + assert result.exit_code == 0 + assert json.loads(result.stdout) == { + "data": { + "app_id": APP_ID, + "connect_url": CONNECT_URL, + } + } + assert result.stderr == "" + mock_launch.assert_not_called() + + +@pytest.mark.respx +def test_connect_resource_opens_browser_for_linked_app( + logged_in_cli: None, + respx_mock: respx.MockRouter, + tmp_path: Path, +) -> None: + write_app_config( + tmp_path, + AppConfig(app_id=APP_ID, team_id=TEAM_ID), + ) + _mock_app_and_team(respx_mock) + + with ( + changing_dir(tmp_path), + patch( + "fastapi_cloud_cli.commands.integrations.resources.connect.typer.launch", + return_value=0, + ) as mock_launch, + ): + result = runner.invoke(app, ["integrations", "resources", "connect"]) + + assert result.exit_code == 0 + mock_launch.assert_called_once_with(CONNECT_URL) + assert "connect resource" in result.output + assert "Opened the integration setup for API in your browser." in result.output + assert CONNECT_URL in result.output + assert "When the connection is complete" in result.output + assert "fastapi cloud integrations resources list" in " ".join( + result.output.split() + ) + assert APP_ID not in result.output + + +@pytest.mark.respx +def test_connect_resource_no_open_prints_url_without_opening_browser( + logged_in_cli: None, + respx_mock: respx.MockRouter, +) -> None: + _mock_app_and_team(respx_mock) + + with patch( + "fastapi_cloud_cli.commands.integrations.resources.connect.typer.launch" + ) as mock_launch: + result = runner.invoke( + app, + [ + "integrations", + "resources", + "connect", + "--app-id", + APP_ID, + "--no-open", + ], + ) + + assert result.exit_code == 0 + mock_launch.assert_not_called() + assert "Open the integration setup for API in your browser:" in result.output + assert CONNECT_URL in result.output + assert f"fastapi cloud integrations resources list --app-id {APP_ID}" in " ".join( + result.output.split() + ) + + +def test_connect_resource_json_requires_app_without_linked_app( + logged_in_cli: None, + tmp_path: Path, +) -> None: + with changing_dir(tmp_path): + result = runner.invoke( + app, + ["integrations", "resources", "connect", "--json"], + ) + + assert result.exit_code == 1 + assert json.loads(result.stdout) == { + "error": { + "code": "missing_required_input", + "message": "App ID is required.", + "hint": "Pass --app-id or run `fastapi cloud apps create --link` first.", + } + } + assert result.stderr == "" + + +@pytest.mark.respx +def test_connect_resource_json_returns_not_found_for_unknown_app( + logged_in_cli: None, + respx_mock: respx.MockRouter, +) -> None: + respx_mock.get(f"/apps/{APP_ID}").mock(return_value=Response(404)) + + result = runner.invoke( + app, + ["integrations", "resources", "connect", "--app-id", APP_ID, "--json"], + ) + + assert result.exit_code == 1 + assert json.loads(result.stdout) == { + "error": { + "code": "not_found", + "message": "App not found.", + "hint": None, + } + } + assert result.stderr == ""