From 94916ff54f991fdff4f6c1db475aa24fe3d7ad1e Mon Sep 17 00:00:00 2001 From: Patrick Arminio Date: Sun, 30 Aug 2026 20:27:21 +0200 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Add=20`fastapi=20cloud=20domains=20?= =?UTF-8?q?get`?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Shortcake-Parent: main --- .../commands/domains/__init__.py | 2 + .../commands/domains/_setup.py | 211 ++++++++ .../commands/domains/_shared.py | 40 ++ src/fastapi_cloud_cli/commands/domains/get.py | 128 +++++ .../commands/domains/rendering.py | 137 +++++- tests/domains/__init__.py | 0 tests/domains/test_cli.py | 462 ++++++++++++++++++ tests/domains/test_setup.py | 190 +++++++ tests/test_cli_domains.py | 170 ------- 9 files changed, 1169 insertions(+), 171 deletions(-) create mode 100644 src/fastapi_cloud_cli/commands/domains/_setup.py create mode 100644 src/fastapi_cloud_cli/commands/domains/_shared.py create mode 100644 src/fastapi_cloud_cli/commands/domains/get.py create mode 100644 tests/domains/__init__.py create mode 100644 tests/domains/test_cli.py create mode 100644 tests/domains/test_setup.py delete mode 100644 tests/test_cli_domains.py diff --git a/src/fastapi_cloud_cli/commands/domains/__init__.py b/src/fastapi_cloud_cli/commands/domains/__init__.py index 925700f1..29898eae 100644 --- a/src/fastapi_cloud_cli/commands/domains/__init__.py +++ b/src/fastapi_cloud_cli/commands/domains/__init__.py @@ -1,11 +1,13 @@ import typer +from fastapi_cloud_cli.commands.domains.get import get_domain from fastapi_cloud_cli.commands.domains.list import list_domains domains_app = typer.Typer( no_args_is_help=True, help="Manage the custom domains of your app.", ) +domains_app.command("get")(get_domain) domains_app.command("list")(list_domains) __all__ = ["domains_app"] diff --git a/src/fastapi_cloud_cli/commands/domains/_setup.py b/src/fastapi_cloud_cli/commands/domains/_setup.py new file mode 100644 index 00000000..c7ae0187 --- /dev/null +++ b/src/fastapi_cloud_cli/commands/domains/_setup.py @@ -0,0 +1,211 @@ +from dataclasses import dataclass +from typing import Literal + +from fastapi_cloud_cli.api import ( + CustomDomain, + CustomDomainRecord, + CustomDomainStatus, +) + +StepStatus = Literal["verified", "in_progress", "attention", "locked", "failed"] +StepID = Literal["ownership", "certificate", "traffic", "combined"] +PhasedStepID = Literal["ownership", "certificate", "traffic"] +RecordGroup = PhasedStepID +Phase = Literal["internal", "external", "origin"] + + +@dataclass(frozen=True) +class SetupStep: + id: StepID + title: str + description: str + status: StepStatus + records: list[CustomDomainRecord] + + +ATTENTION_STATUSES = frozenset( + { + CustomDomainStatus.internal_dcv_invalid, + CustomDomainStatus.origin_setup_invalid, + } +) + +FAILED_STATUSES = frozenset( + { + CustomDomainStatus.internal_dcv_timeout, + CustomDomainStatus.internal_dcv_revoked, + CustomDomainStatus.external_dcv_blocked, + CustomDomainStatus.external_dcv_timeout, + CustomDomainStatus.origin_setup_timeout, + CustomDomainStatus.origin_setup_removed, + } +) + +PHASE_ORDER: dict[Phase, int] = { + "internal": 0, + "external": 1, + "origin": 2, +} + + +def _phase_from_status(status: CustomDomainStatus) -> Phase: + if status.value.startswith("internal_dcv"): + return "internal" + if status.value.startswith("external_dcv"): + return "external" + return "origin" + + +def _record_group(record: CustomDomainRecord) -> RecordGroup: + if record.type == "A": + return "traffic" + if record.type == "TXT": + return "ownership" if "_fc-dcv" in (record.name or "") else "certificate" + return "certificate" if "_acme-challenge" in (record.name or "") else "traffic" + + +def _step_status(status: CustomDomainStatus, phase: Phase) -> StepStatus: + current_phase = PHASE_ORDER[_phase_from_status(status)] + step_phase = PHASE_ORDER[phase] + if current_phase > step_phase: + return "verified" + if current_phase < step_phase: + return "locked" + if status == CustomDomainStatus.origin_setup_success: + return "verified" + if status in ATTENTION_STATUSES: + return "attention" + if status in FAILED_STATUSES: + return "failed" + return "in_progress" + + +def _describe_step( + step_id: StepID, + status: StepStatus, + *, + is_apex: bool = False, +) -> str: + if step_id == "ownership": + if status == "attention": + return ( + "We found this record, but its value doesn't match. Update it to " + "the value below. We re-check every minute, no restart needed." + ) + if status == "failed": + return ( + "We couldn't confirm ownership. Re-check the value below, then " + "restart verification." + ) + if status == "verified": + return "Ownership confirmed." + return ( + "Add this TXT record. We check every minute and unlock the next step " + "automatically." + ) + + if step_id == "certificate": + if status == "locked": + return "Unlocks once ownership is verified." + if status == "failed": + return ( + "We couldn't secure your domain. Re-check the records below, then " + "restart verification." + ) + if status == "verified": + return "Domain verified and TLS certificate issued." + return ( + "Add both records so we can issue your TLS certificate. Your live site " + "doesn't change yet." + ) + + if step_id == "traffic": + if status == "locked": + return ( + "Unlocks once your domain is secured. Your live site won't change " + "until you add the records here." + ) + if status == "attention": + return ( + "We found these records, but they don't match. Update them to the " + "values below. We re-check automatically." + ) + if status == "failed": + return ( + "We couldn't confirm your traffic records. Re-check them below, then " + "restart verification." + ) + if status == "verified": + return "Your domain is live on FastAPI Cloud." + return "Add these records to move traffic to FastAPI Cloud with no downtime." + + if status == "attention": + return ( + "We found your records, but some values don't match. Update them to the " + "values below. We re-check automatically." + ) + if status == "failed": + return ( + "We couldn't complete setup. Re-check the records below, then restart " + "verification." + ) + if status == "verified": + return "Your domain is live on FastAPI Cloud." + if is_apex: + return ( + "Add the records below. We'll verify ownership, issue your TLS " + "certificate, and route traffic automatically." + ) + return ( + "Add the record below. We'll verify ownership, issue your TLS certificate, " + "and route traffic automatically." + ) + + +def get_setup_steps(domain: CustomDomain) -> list[SetupStep]: + is_apex = any(record.type == "A" for record in domain.dns_records) + + if not domain.is_using_pre_validation: + if domain.setup_successful: + status: StepStatus = "verified" + elif domain.setup_failed: + status = "failed" + elif domain.status in ATTENTION_STATUSES: + status = "attention" + else: + status = "in_progress" + + return [ + SetupStep( + id="combined", + title="Verify ownership and route traffic", + description=_describe_step("combined", status, is_apex=is_apex), + status=status, + records=domain.dns_records, + ) + ] + + def records_for(group: RecordGroup) -> list[CustomDomainRecord]: + return [ + record for record in domain.dns_records if _record_group(record) == group + ] + + def build_step( + step_id: PhasedStepID, + title: str, + phase: Phase, + ) -> SetupStep: + status = _step_status(domain.status, phase) + return SetupStep( + id=step_id, + title=title, + description=_describe_step(step_id, status), + status=status, + records=records_for(step_id), + ) + + return [ + build_step("ownership", "Prove ownership", "internal"), + build_step("certificate", "Secure your domain", "external"), + build_step("traffic", "Switch traffic", "origin"), + ] diff --git a/src/fastapi_cloud_cli/commands/domains/_shared.py b/src/fastapi_cloud_cli/commands/domains/_shared.py new file mode 100644 index 00000000..6d86b252 --- /dev/null +++ b/src/fastapi_cloud_cli/commands/domains/_shared.py @@ -0,0 +1,40 @@ +import string + +from rich_toolkit.menu import Option + +from fastapi_cloud_cli.api import CustomDomain +from fastapi_cloud_cli.utils.cli import FastAPIRichToolkit + + +def _normalize_domain_name(name: str) -> str: + return name.strip(string.whitespace + ".").lower() + + +def _find_custom_domain( + domains: list[CustomDomain], + name_or_id: str, +) -> CustomDomain | None: + normalized = _normalize_domain_name(name_or_id) + + return next( + ( + domain + for domain in domains + if domain.id.lower() == normalized + or _normalize_domain_name(domain.name) == normalized + ), + None, + ) + + +def _select_custom_domain( + toolkit: FastAPIRichToolkit, + domains: list[CustomDomain], + *, + prompt: str, +) -> CustomDomain: + return toolkit.ask( + prompt, + options=[Option({"name": domain.name, "value": domain}) for domain in domains], + bullet=False, + ) diff --git a/src/fastapi_cloud_cli/commands/domains/get.py b/src/fastapi_cloud_cli/commands/domains/get.py new file mode 100644 index 00000000..053b4e4f --- /dev/null +++ b/src/fastapi_cloud_cli/commands/domains/get.py @@ -0,0 +1,128 @@ +from typing import Annotated, Any + +import typer +from pydantic import BaseModel, Field +from rich_toolkit import RichToolkit + +from fastapi_cloud_cli.api import APIClient, CustomDomain +from fastapi_cloud_cli.commands.domains._shared import ( + _find_custom_domain, + _select_custom_domain, +) +from fastapi_cloud_cli.commands.domains.rendering import render_custom_domain_details +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 + + +class CustomDomainGetOutput(BaseModel): + app_id: str + domain: CustomDomain + show_title: Annotated[bool, Field(exclude=True)] = True + + +def _render_custom_domain_get_output( + data: CustomDomainGetOutput, + toolkit: RichToolkit, +) -> None: + if data.show_title: + toolkit.print_title("custom domains") + toolkit.print_line() + + render_custom_domain_details(data.domain, toolkit) + + +def get_domain( + domain: Annotated[ + str | None, + typer.Argument( + help="Hostname or ID of the custom domain to return.", + ), + ] = None, + app_id: Annotated[ + str | None, + typer.Option( + "--app-id", + help="ID of the app that owns the custom domain.", + ), + ] = None, + json_output: JsonOutputOption = False, +) -> Any: + """ + Get a custom domain 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`.", + ) + + app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) + domain_was_provided = domain is not None + + if domain is None and toolkit.mode == "json": + toolkit.fail( + "missing_required_input", + "Custom domain is required.", + hint="Pass DOMAIN to choose a custom domain.", + ) + + with APIClient() as client: + with ( + toolkit.progress( + title="Fetching custom domains", + transient=True, + ) as progress, + client.handle_http_errors( + progress, + default_message=( + "Error fetching custom domains. Please try again later." + ), + not_found_message="App not found.", + toolkit=toolkit, + ), + ): + domains = client.get_custom_domains(app_id=app_id).data + + selected_domain: CustomDomain | None + + if domain is None: + toolkit.print_title("custom domains") + toolkit.print_line() + + if not domains: + toolkit.print("No custom domains found.", bullet=False) + return + + selected_domain = _select_custom_domain( + toolkit, + domains, + prompt="Select the custom domain to get:", + ) + toolkit.print_line() + else: + if (selected_domain := _find_custom_domain(domains, domain)) is None: + toolkit.fail( + "not_found", + f"Custom domain {domain} not found.", + hint=( + "Run `fastapi cloud domains list` to see available custom " + "domains." + ), + ) + + assert selected_domain is not None + + toolkit.success( + CustomDomainGetOutput( + app_id=app_id, + domain=selected_domain, + show_title=domain_was_provided, + ), + render_output=_render_custom_domain_get_output, + ) diff --git a/src/fastapi_cloud_cli/commands/domains/rendering.py b/src/fastapi_cloud_cli/commands/domains/rendering.py index 2b16d983..cbb8f893 100644 --- a/src/fastapi_cloud_cli/commands/domains/rendering.py +++ b/src/fastapi_cloud_cli/commands/domains/rendering.py @@ -1,9 +1,22 @@ from dataclasses import dataclass +from rich.console import RenderableType from rich.table import Table from rich.text import Text +from rich_toolkit import RichToolkit -from fastapi_cloud_cli.api import CustomDomain, CustomDomainStatus +from fastapi_cloud_cli.api import ( + CustomDomain, + CustomDomainRecord, + CustomDomainStatus, +) +from fastapi_cloud_cli.commands.domains._setup import ( + ATTENTION_STATUSES, + SetupStep, + StepStatus, + get_setup_steps, +) +from fastapi_cloud_cli.utils.cli import get_details_table from fastapi_cloud_cli.utils.dates import format_last_updated @@ -173,3 +186,125 @@ def get_custom_domains_table(domains: list[CustomDomain]) -> Table: ) return table + + +STEP_LABELS: dict[StepStatus, str] = { + "verified": "Verified", + "in_progress": "In progress", + "attention": "Needs attention", + "locked": "Locked", + "failed": "Action needed", +} + + +def _get_dns_records_table(records: list[CustomDomainRecord]) -> Table: + table = Table.grid(padding=(0, 2), pad_edge=False) + table.add_column("Type", no_wrap=True) + table.add_column("Name", overflow="fold") + table.add_column("Value", overflow="fold") + table.add_row( + Text("Type", style="bold"), + Text("Name", style="bold"), + Text("Value", style="bold"), + ) + + generating = Text("Generating; check again shortly", style="dim italic") + for record in records: + table.add_row( + Text(record.type), + Text(record.name) if record.name is not None else generating, + Text(record.value) if record.value is not None else generating, + ) + + return table + + +def _render_setup_step( + step: SetupStep, + toolkit: RichToolkit, + *, + number: int, +) -> None: + toolkit.print( + f"[bold]{number} {step.title}[/bold] [dim]{STEP_LABELS[step.status]}[/dim]", + bullet=False, + ) + toolkit.print(step.description, bullet=False) + + records = step.records if step.status != "locked" else [] + if not records: + return + + toolkit.print_line() + toolkit.print(_get_dns_records_table(records), bullet=False) + + if any( + record.type == "CNAME" and (record.value or "").endswith(".") + for record in records + ): + toolkit.print_line() + toolkit.print( + "Copy CNAME values as shown. If your DNS provider rejects the trailing " + "dot, remove it and try again.", + emoji="💡", + ) + + if step.status != "verified" and any( + record.type in {"CNAME", "A"} for record in records + ): + toolkit.print_line() + toolkit.print( + "Using Cloudflare? Set these records to DNS only (gray cloud), not " + "Proxied (orange cloud).", + emoji="💡", + ) + + +def _get_next_action(domain: CustomDomain) -> str: + if domain.setup_successful: + return "No action needed. Your domain is live." + if domain.setup_failed: + return ( + "Correct the DNS records, then run " + f"`fastapi cloud domains restart {domain.name}`." + ) + if domain.status in ATTENTION_STATUSES: + return "Correct the DNS records shown. We'll check again automatically." + return ( + "Wait for automatic verification. DNS changes can take up to 48 hours " + "to propagate." + ) + + +def render_custom_domain_details( + domain: CustomDomain, + toolkit: RichToolkit, +) -> None: + metadata = DOMAIN_STATUS[domain.status] + rows: list[tuple[str, RenderableType]] = [ + ("hostname", domain.name), + ("status", metadata.label), + ("raw status", domain.status.value), + ("setup mode", get_setup_mode_label(domain)), + ("id", domain.id), + ("created", format_last_updated(domain.created_at)), + ("updated", format_last_updated(domain.updated_at)), + ("setup started", format_last_updated(domain.setup_started_at)), + ("last checked", format_last_updated(domain.setup_checked_at)), + ] + if domain.setup_successful: + url = f"https://{domain.name}" + rows.insert(1, ("url", Text(url, style=f"link {url}"))) + + toolkit.print(Text(domain.name, style="bold"), emoji="🌐") + toolkit.print_line() + toolkit.print(get_details_table(rows)) + toolkit.print_line() + toolkit.print(f"[bold]{metadata.title}[/bold]\n{metadata.description}") + + for number, step in enumerate(get_setup_steps(domain), start=1): + toolkit.print_line() + _render_setup_step(step, toolkit, number=number) + + toolkit.print_line() + toolkit.print(_get_next_action(domain), emoji="⏳") diff --git a/tests/domains/__init__.py b/tests/domains/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/domains/test_cli.py b/tests/domains/test_cli.py new file mode 100644 index 00000000..0664a7f6 --- /dev/null +++ b/tests/domains/test_cli.py @@ -0,0 +1,462 @@ +import json +from datetime import datetime, timezone +from textwrap import dedent +from typing import Any +from unittest.mock import patch + +import pytest +import respx +import time_machine +from httpx import Response +from typer.testing import CliRunner + +from fastapi_cloud_cli.api import CustomDomain, CustomDomainStatus +from fastapi_cloud_cli.cli import cloud_app as app +from fastapi_cloud_cli.commands.domains.rendering import ( + DOMAIN_STATUS, + _get_next_action, +) +from tests.conftest import ConfiguredApp +from tests.utils import Keys, changing_dir + +runner = CliRunner() + +APP_ID = "00000000-0000-4000-8000-000000000002" +DOMAIN_ID = "00000000-0000-4000-8000-000000000003" + + +def _normalize_output(output: str) -> str: + output = dedent(output.replace("\u200b", "")).strip() + return "\n".join(line.rstrip() for line in output.splitlines()) + + +def custom_domain(**overrides: Any) -> dict[str, Any]: + return { + "id": DOMAIN_ID, + "name": "api.example.com", + "status": "internal_dcv_pending", + "setup_in_progress": True, + "setup_failed": False, + "setup_successful": False, + "is_using_pre_validation": False, + "dns_records": [ + { + "type": "CNAME", + "name": "api", + "value": f"{DOMAIN_ID}.endpoints.fastapicloud.dev.", + } + ], + "created_at": "2026-08-28T10:00:00Z", + "updated_at": "2026-08-28T10:01:00Z", + "setup_started_at": "2026-08-28T10:00:00Z", + "setup_checked_at": "2026-08-28T10:01:00Z", + "app_id": APP_ID, + **overrides, + } + + +@pytest.mark.parametrize( + "command", + [["list"], ["get", "api.example.com"]], + ids=["list", "get"], +) +def test_domains_commands_require_user_session( + command: list[str], + logged_out_cli: None, +) -> None: + result = runner.invoke( + app, + ["domains", *command, "--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`.", + } + } + + +def test_domains_list_requires_app_context(logged_in_cli: None) -> None: + result = runner.invoke(app, ["domains", "list", "--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.", + } + } + + +@pytest.mark.respx +@time_machine.travel(datetime(2026, 9, 1, 10, 1, tzinfo=timezone.utc), tick=False) +def test_domains_list_uses_linked_app_and_renders_rows( + logged_in_cli: None, + respx_mock: respx.MockRouter, + configured_app: ConfiguredApp, +) -> None: + newest = custom_domain( + id="00000000-0000-4000-8000-000000000004", + name="new.example.com", + status="origin_setup_success", + setup_in_progress=False, + setup_successful=True, + is_using_pre_validation=True, + app_id=configured_app.app_id, + ) + oldest = custom_domain(name="old.example.com", app_id=configured_app.app_id) + respx_mock.get(f"/apps/{configured_app.app_id}/custom-domains").mock( + return_value=Response(200, json={"data": [newest, oldest], "count": 2}) + ) + + with changing_dir(configured_app.path): + result = runner.invoke(app, ["domains", "list"]) + + assert result.exit_code == 0 + assert _normalize_output(result.output) == _normalize_output( + """ + custom domains + + Domain Status Setup mode Last check + + new.example.com Live Zero-downtime 4 days ago + old.example.com Pending Standard 4 days ago + """ + ) + + +@pytest.mark.respx +def test_domains_list_returns_json( + logged_in_cli: None, + respx_mock: respx.MockRouter, +) -> None: + domain = custom_domain() + respx_mock.get(f"/apps/{APP_ID}/custom-domains").mock( + return_value=Response(200, json={"data": [domain], "count": 1}) + ) + + result = runner.invoke(app, ["domains", "list", "--app-id", APP_ID, "--json"]) + + assert result.exit_code == 0 + assert json.loads(result.stdout) == { + "data": { + "app_id": APP_ID, + "domains": [domain], + "total_count": 1, + } + } + + +@pytest.mark.respx +def test_domains_list_renders_empty_state( + logged_in_cli: None, + respx_mock: respx.MockRouter, +) -> None: + respx_mock.get(f"/apps/{APP_ID}/custom-domains").mock( + return_value=Response(200, json={"data": [], "count": 0}) + ) + + result = runner.invoke(app, ["domains", "list", "--app-id", APP_ID]) + + assert result.exit_code == 0 + assert "No custom domains found." in result.output + + +@pytest.mark.respx +def test_domains_list_surfaces_app_not_found_as_json( + logged_in_cli: None, + respx_mock: respx.MockRouter, +) -> None: + respx_mock.get(f"/apps/{APP_ID}/custom-domains").mock( + return_value=Response(404, json={"detail": "App not found"}) + ) + + result = runner.invoke(app, ["domains", "list", "--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, + } + } + + +def test_domain_status_metadata_is_exhaustive() -> None: + assert set(DOMAIN_STATUS) == set(CustomDomainStatus) + + +@pytest.mark.parametrize( + ("overrides", "expected"), + [ + ( + { + "status": "internal_dcv_timeout", + "setup_in_progress": False, + "setup_failed": True, + }, + ( + "Correct the DNS records, then run " + "`fastapi cloud domains restart api.example.com`." + ), + ), + ( + {"status": "internal_dcv_invalid"}, + "Correct the DNS records shown. We'll check again automatically.", + ), + ], +) +def test_domain_next_action_when_user_intervention_is_required( + overrides: dict[str, Any], + expected: str, +) -> None: + domain = CustomDomain.model_validate(custom_domain(**overrides)) + + assert _get_next_action(domain) == expected + + +def test_domains_get_json_requires_domain(logged_in_cli: None) -> None: + result = runner.invoke( + app, + ["domains", "get", "--app-id", APP_ID, "--json"], + ) + + assert result.exit_code == 1 + assert json.loads(result.stdout) == { + "error": { + "code": "missing_required_input", + "message": "Custom domain is required.", + "hint": "Pass DOMAIN to choose a custom domain.", + } + } + + +@pytest.mark.respx +@pytest.mark.parametrize("identifier", [" API.Example.COM. ", DOMAIN_ID]) +def test_domains_get_resolves_normalized_hostname_or_id_as_json( + identifier: str, + logged_in_cli: None, + respx_mock: respx.MockRouter, +) -> None: + domain = custom_domain() + respx_mock.get(f"/apps/{APP_ID}/custom-domains").mock( + return_value=Response(200, json={"data": [domain], "count": 1}) + ) + + result = runner.invoke( + app, + ["domains", "get", identifier, "--app-id", APP_ID, "--json"], + ) + + assert result.exit_code == 0 + assert json.loads(result.stdout) == {"data": {"app_id": APP_ID, "domain": domain}} + + +@pytest.mark.respx +@time_machine.travel(datetime(2026, 9, 1, 10, 1, tzinfo=timezone.utc), tick=False) +def test_domains_get_prompts_with_selector_and_renders_details( + logged_in_cli: None, + respx_mock: respx.MockRouter, +) -> None: + domain = custom_domain() + respx_mock.get(f"/apps/{APP_ID}/custom-domains").mock( + return_value=Response(200, json={"data": [domain], "count": 1}) + ) + + with patch("rich_toolkit.container.getchar", side_effect=[Keys.ENTER]): + result = runner.invoke(app, ["domains", "get", "--app-id", APP_ID]) + + assert result.exit_code == 0 + assert _normalize_output(result.output) == _normalize_output( + """ + custom domains + + Select the custom domain to get: + ● api.example.com + + Select the custom domain to get: api.example.com + + 🌐 api.example.com + + hostname api.example.com + status Pending + raw status internal_dcv_pending + setup mode Standard + id 00000000-0000-4000-8000-000000000003 + created 4 days ago + updated 4 days ago + setup started 4 days ago + last checked 4 days ago + + Waiting domain verification + We are checking your DNS configuration to confirm domain ownership. This + usually takes a few minutes. + + 1 Verify ownership and route traffic In progress + Add the record below. We'll verify ownership, issue your TLS certificate, and + route traffic automatically. + + Type Name Value + CNAME api 00000000-0000-4000-8000-000000000003.endpoints.fastapicloud.dev. + + 💡 Copy CNAME values as shown. If your DNS provider rejects the trailing dot, + remove it and try again. + + 💡 Using Cloudflare? Set these records to DNS only (gray cloud), not Proxied + (orange cloud). + + ⏳ Wait for automatic verification. DNS changes can take up to 48 hours to + propagate. + """ + ) + + +@pytest.mark.respx +def test_domains_get_reports_not_found( + logged_in_cli: None, + respx_mock: respx.MockRouter, +) -> None: + respx_mock.get(f"/apps/{APP_ID}/custom-domains").mock( + return_value=Response(200, json={"data": [custom_domain()], "count": 1}) + ) + + result = runner.invoke( + app, + ["domains", "get", "missing.example.com", "--app-id", APP_ID, "--json"], + ) + + assert result.exit_code == 1 + assert json.loads(result.stdout) == { + "error": { + "code": "not_found", + "message": "Custom domain missing.example.com not found.", + "hint": ( + "Run `fastapi cloud domains list` to see available custom domains." + ), + } + } + + +@pytest.mark.respx +def test_domains_get_selector_handles_empty_collection( + logged_in_cli: None, + respx_mock: respx.MockRouter, +) -> None: + respx_mock.get(f"/apps/{APP_ID}/custom-domains").mock( + return_value=Response(200, json={"data": [], "count": 0}) + ) + + result = runner.invoke(app, ["domains", "get", "--app-id", APP_ID]) + + assert result.exit_code == 0 + assert "No custom domains found." in result.output + + +PREVALIDATION_RECORDS = [ + {"type": "TXT", "name": "_fc-dcv.api", "value": "ownership-value"}, + {"type": "TXT", "name": None, "value": None}, + { + "type": "CNAME", + "name": "_acme-challenge.api", + "value": "certificate-value.", + }, + {"type": "CNAME", "name": "api", "value": "traffic-value."}, +] + + +@pytest.mark.respx +@pytest.mark.parametrize( + ("status", "shows_certificate", "shows_traffic"), + [ + ("internal_dcv_pending", False, False), + ("external_dcv_pending", True, False), + ("origin_setup_pending", True, True), + ], +) +def test_domains_get_gates_zero_downtime_records_by_phase( + status: str, + shows_certificate: bool, + shows_traffic: bool, + logged_in_cli: None, + respx_mock: respx.MockRouter, +) -> None: + domain = custom_domain( + status=status, + is_using_pre_validation=True, + dns_records=PREVALIDATION_RECORDS, + ) + respx_mock.get(f"/apps/{APP_ID}/custom-domains").mock( + return_value=Response(200, json={"data": [domain], "count": 1}) + ) + + result = runner.invoke( + app, + ["domains", "get", "api.example.com", "--app-id", APP_ID], + ) + + assert result.exit_code == 0 + assert "ownership-value" in result.output + assert ("certificate-value" in result.output) is shows_certificate + assert ("Generating; check again shortly" in result.output) is shows_certificate + assert ("traffic-value" in result.output) is shows_traffic + + +@pytest.mark.respx +@time_machine.travel(datetime(2026, 9, 1, 10, 1, tzinfo=timezone.utc), tick=False) +def test_domains_get_shows_url_and_no_action_when_live( + logged_in_cli: None, + respx_mock: respx.MockRouter, +) -> None: + domain = custom_domain( + status="origin_setup_success", + setup_in_progress=False, + setup_successful=True, + ) + respx_mock.get(f"/apps/{APP_ID}/custom-domains").mock( + return_value=Response(200, json={"data": [domain], "count": 1}) + ) + + result = runner.invoke( + app, + ["domains", "get", domain["name"], "--app-id", APP_ID], + ) + + assert result.exit_code == 0 + assert _normalize_output(result.output) == _normalize_output( + """ + custom domains + + 🌐 api.example.com + + hostname api.example.com + url https://api.example.com + status Live + raw status origin_setup_success + setup mode Standard + id 00000000-0000-4000-8000-000000000003 + created 4 days ago + updated 4 days ago + setup started 4 days ago + last checked 4 days ago + + Live + Your domain is fully configured, secured with TLS, and set up to route + traffic to your app. + + 1 Verify ownership and route traffic Verified + Your domain is live on FastAPI Cloud. + + Type Name Value + CNAME api 00000000-0000-4000-8000-000000000003.endpoints.fastapicloud.dev. + + 💡 Copy CNAME values as shown. If your DNS provider rejects the trailing dot, + remove it and try again. + + ⏳ No action needed. Your domain is live. + """ + ) diff --git a/tests/domains/test_setup.py b/tests/domains/test_setup.py new file mode 100644 index 00000000..8ab791bf --- /dev/null +++ b/tests/domains/test_setup.py @@ -0,0 +1,190 @@ +from typing import Any + +import pytest + +from fastapi_cloud_cli.api import CustomDomain, CustomDomainStatus +from fastapi_cloud_cli.commands.domains._setup import get_setup_steps + +APP_ID = "00000000-0000-4000-8000-000000000002" +DOMAIN_ID = "00000000-0000-4000-8000-000000000003" + +STANDARD_CNAME_RECORD = { + "type": "CNAME", + "name": "www", + "value": f"{DOMAIN_ID}.endpoints.fastapicloud.dev.", +} +PRE_VALIDATION_RECORDS = [ + {"type": "TXT", "name": "_fc-dcv.www", "value": "ownership-value"}, + {"type": "TXT", "name": None, "value": None}, + { + "type": "CNAME", + "name": "_acme-challenge.www", + "value": "certificate-value", + }, + {"type": "A", "name": "@", "value": "192.0.2.10"}, + {"type": "CNAME", "name": "www", "value": "traffic-value"}, +] + + +def _custom_domain(**overrides: Any) -> CustomDomain: + return CustomDomain.model_validate( + { + "id": DOMAIN_ID, + "name": "www.example.com", + "status": CustomDomainStatus.internal_dcv_pending, + "setup_in_progress": True, + "setup_failed": False, + "setup_successful": False, + "is_using_pre_validation": False, + "dns_records": [STANDARD_CNAME_RECORD], + "created_at": "2026-08-28T10:00:00Z", + "updated_at": "2026-08-28T10:01:00Z", + "setup_started_at": "2026-08-28T10:00:00Z", + "setup_checked_at": "2026-08-28T10:01:00Z", + "app_id": APP_ID, + **overrides, + } + ) + + +@pytest.mark.parametrize( + ("statuses", "expected_statuses"), + [ + ( + ( + CustomDomainStatus.internal_dcv_pending, + CustomDomainStatus.internal_dcv_missing, + ), + ("in_progress", "locked", "locked"), + ), + ( + (CustomDomainStatus.internal_dcv_invalid,), + ("attention", "locked", "locked"), + ), + ( + ( + CustomDomainStatus.internal_dcv_timeout, + CustomDomainStatus.internal_dcv_revoked, + ), + ("failed", "locked", "locked"), + ), + ( + ( + CustomDomainStatus.external_dcv_pending, + CustomDomainStatus.external_dcv_proxied, + CustomDomainStatus.external_dcv_secured, + ), + ("verified", "in_progress", "locked"), + ), + ( + ( + CustomDomainStatus.external_dcv_blocked, + CustomDomainStatus.external_dcv_timeout, + ), + ("verified", "failed", "locked"), + ), + ( + ( + CustomDomainStatus.origin_setup_pending, + CustomDomainStatus.origin_setup_missing, + ), + ("verified", "verified", "in_progress"), + ), + ( + (CustomDomainStatus.origin_setup_invalid,), + ("verified", "verified", "attention"), + ), + ( + ( + CustomDomainStatus.origin_setup_timeout, + CustomDomainStatus.origin_setup_removed, + ), + ("verified", "verified", "failed"), + ), + ( + (CustomDomainStatus.origin_setup_success,), + ("verified", "verified", "verified"), + ), + ], +) +def test_pre_validation_setup_steps_follow_domain_phase( + statuses: tuple[CustomDomainStatus, ...], + expected_statuses: tuple[str, str, str], +) -> None: + for status in statuses: + steps = get_setup_steps( + _custom_domain( + status=status, + is_using_pre_validation=True, + dns_records=PRE_VALIDATION_RECORDS, + ) + ) + + assert tuple(step.status for step in steps) == expected_statuses + assert [[record.name for record in step.records] for step in steps] == [ + ["_fc-dcv.www"], + [None, "_acme-challenge.www"], + ["@", "www"], + ] + + +@pytest.mark.parametrize( + ("overrides", "expected_status", "expected_description"), + [ + ( + {}, + "in_progress", + ( + "Add the record below. We'll verify ownership, issue your TLS " + "certificate, and route traffic automatically." + ), + ), + ( + {"dns_records": [{"type": "A", "name": "@", "value": "192.0.2.10"}]}, + "in_progress", + ( + "Add the records below. We'll verify ownership, issue your TLS " + "certificate, and route traffic automatically." + ), + ), + ( + {"status": CustomDomainStatus.internal_dcv_invalid}, + "attention", + ( + "We found your records, but some values don't match. Update them to " + "the values below. We re-check automatically." + ), + ), + ( + { + "status": CustomDomainStatus.internal_dcv_timeout, + "setup_in_progress": False, + "setup_failed": True, + }, + "failed", + ( + "We couldn't complete setup. Re-check the records below, then restart " + "verification." + ), + ), + ( + { + "status": CustomDomainStatus.origin_setup_success, + "setup_in_progress": False, + "setup_successful": True, + }, + "verified", + "Your domain is live on FastAPI Cloud.", + ), + ], +) +def test_standard_setup_step_reflects_domain_state( + overrides: dict[str, Any], + expected_status: str, + expected_description: str, +) -> None: + [step] = get_setup_steps(_custom_domain(**overrides)) + + assert step.id == "combined" + assert step.status == expected_status + assert step.description == expected_description diff --git a/tests/test_cli_domains.py b/tests/test_cli_domains.py deleted file mode 100644 index 870463f3..00000000 --- a/tests/test_cli_domains.py +++ /dev/null @@ -1,170 +0,0 @@ -import json -from datetime import datetime, timezone -from textwrap import dedent -from typing import Any - -import pytest -import respx -import time_machine -from httpx import Response -from typer.testing import CliRunner - -from fastapi_cloud_cli.api import CustomDomainStatus -from fastapi_cloud_cli.cli import cloud_app as app -from fastapi_cloud_cli.commands.domains.rendering import DOMAIN_STATUS -from tests.conftest import ConfiguredApp -from tests.utils import changing_dir - -runner = CliRunner() - -APP_ID = "00000000-0000-4000-8000-000000000002" -DOMAIN_ID = "00000000-0000-4000-8000-000000000003" - - -def _normalize_output(output: str) -> str: - return "\n".join(line.rstrip() for line in dedent(output).strip().splitlines()) - - -def custom_domain(**overrides: Any) -> dict[str, Any]: - return { - "id": DOMAIN_ID, - "name": "api.example.com", - "status": "internal_dcv_pending", - "setup_in_progress": True, - "setup_failed": False, - "setup_successful": False, - "is_using_pre_validation": False, - "dns_records": [ - { - "type": "CNAME", - "name": "api", - "value": f"{DOMAIN_ID}.endpoints.fastapicloud.dev.", - } - ], - "created_at": "2026-08-28T10:00:00Z", - "updated_at": "2026-08-28T10:01:00Z", - "setup_started_at": "2026-08-28T10:00:00Z", - "setup_checked_at": "2026-08-28T10:01:00Z", - "app_id": APP_ID, - **overrides, - } - - -def test_domains_list_requires_user_session(logged_out_cli: None) -> None: - result = runner.invoke(app, ["domains", "list", "--app-id", APP_ID]) - - assert result.exit_code == 1 - assert "No credentials found." in result.output - assert "FASTAPI_CLOUD_TOKEN" not in result.output - - -def test_domains_list_requires_app_context(logged_in_cli: None) -> None: - result = runner.invoke(app, ["domains", "list", "--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.", - } - } - - -@pytest.mark.respx -@time_machine.travel(datetime(2026, 9, 1, 10, 1, tzinfo=timezone.utc), tick=False) -def test_domains_list_uses_linked_app_and_renders_rows( - logged_in_cli: None, - respx_mock: respx.MockRouter, - configured_app: ConfiguredApp, -) -> None: - newest = custom_domain( - id="00000000-0000-4000-8000-000000000004", - name="new.example.com", - status="origin_setup_success", - setup_in_progress=False, - setup_successful=True, - is_using_pre_validation=True, - app_id=configured_app.app_id, - ) - oldest = custom_domain(name="old.example.com", app_id=configured_app.app_id) - respx_mock.get(f"/apps/{configured_app.app_id}/custom-domains").mock( - return_value=Response(200, json={"data": [newest, oldest], "count": 2}) - ) - - with changing_dir(configured_app.path): - result = runner.invoke(app, ["domains", "list"]) - - assert result.exit_code == 0 - assert _normalize_output(result.output) == _normalize_output( - """ - custom domains - - Domain Status Setup mode Last check - - new.example.com Live Zero-downtime 4 days ago - old.example.com Pending Standard 4 days ago - """ - ) - - -@pytest.mark.respx -def test_domains_list_returns_json( - logged_in_cli: None, - respx_mock: respx.MockRouter, -) -> None: - domain = custom_domain() - respx_mock.get(f"/apps/{APP_ID}/custom-domains").mock( - return_value=Response(200, json={"data": [domain], "count": 1}) - ) - - result = runner.invoke(app, ["domains", "list", "--app-id", APP_ID, "--json"]) - - assert result.exit_code == 0 - assert json.loads(result.stdout) == { - "data": { - "app_id": APP_ID, - "domains": [domain], - "total_count": 1, - } - } - - -@pytest.mark.respx -def test_domains_list_renders_empty_state( - logged_in_cli: None, - respx_mock: respx.MockRouter, -) -> None: - respx_mock.get(f"/apps/{APP_ID}/custom-domains").mock( - return_value=Response(200, json={"data": [], "count": 0}) - ) - - result = runner.invoke(app, ["domains", "list", "--app-id", APP_ID]) - - assert result.exit_code == 0 - assert "No custom domains found." in result.output - - -@pytest.mark.respx -def test_domains_list_surfaces_app_not_found_as_json( - logged_in_cli: None, - respx_mock: respx.MockRouter, -) -> None: - respx_mock.get(f"/apps/{APP_ID}/custom-domains").mock( - return_value=Response(404, json={"detail": "App not found"}) - ) - - result = runner.invoke(app, ["domains", "list", "--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, - } - } - - -def test_domain_status_metadata_is_exhaustive() -> None: - assert set(DOMAIN_STATUS) == set(CustomDomainStatus)