Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 48 additions & 2 deletions lib/cli/src/crewai_cli/deploy/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
console = Console()
_MISSING_LOCKFILE_ERROR_CODES = {"missing_lockfile"}
_DEPLOYMENT_ID_KEYS = ("deployment_id", "deploymentId")
_DEPLOYMENT_FALLBACK_IDENTIFIER_KEYS = ("id", "uuid")
_DEPLOYMENT_FALLBACK_IDENTIFIER_KEYS = ("id",)


def _run_predeploy_validation(
Expand Down Expand Up @@ -223,7 +223,7 @@ def _open_deployment_page(self, json_response: dict[str, Any]) -> None:
getattr(self.plus_api_client, "base_url", None)
or DEFAULT_CREWAI_ENTERPRISE_URL
)
deployment_url = _deployment_page_url(base_url, json_response)
deployment_url = self._deployment_page_url(json_response, base_url)
if not deployment_url:
return

Expand All @@ -239,6 +239,52 @@ def _open_deployment_page(self, json_response: dict[str, Any]) -> None:
style="yellow",
)

def _deployment_page_url(
self,
json_response: dict[str, Any],
base_url: str,
) -> str | None:
"""Build the deployment show URL, resolving UUID-only responses if needed."""
deployment_url = _deployment_page_url(base_url, json_response)
if deployment_url:
return deployment_url

identifier = self._deployment_identifier_from_status(json_response)
if not identifier:
return None

return (
f"{base_url.rstrip('/')}/crewai_plus/deployments/"
f"{quote(identifier, safe='')}"
)

def _deployment_identifier_from_status(
self,
json_response: dict[str, Any],
) -> str | None:
"""Resolve the deployment page id from status without failing the command."""
crew_uuid = json_response.get("uuid")
if not crew_uuid:
return None

try:
response = self.plus_api_client.crew_status_by_uuid(str(crew_uuid))
except Exception:
return None

if not getattr(response, "is_success", False):
return None

try:
status_response = response.json()
except ValueError:
return None

if not isinstance(status_response, dict):
return None

return _deployment_identifier(status_response)

def deploy(self, uuid: str | None = None, skip_validate: bool = False) -> None:
"""
Deploy a crew using either UUID or project name.
Expand Down
51 changes: 48 additions & 3 deletions lib/cli/tests/deploy/test_deploy_main.py
Original file line number Diff line number Diff line change
Expand Up @@ -187,13 +187,13 @@ def test_deployment_page_url_prefers_nested_deployment_id_over_crew_uuid():
)


def test_deployment_page_url_falls_back_to_nested_uuid():
def test_deployment_page_url_does_not_use_uuid_as_deployment_id():
assert (
deploy_main._deployment_page_url(
"https://app.crewai.com/",
{"deployment": {"uuid": "deployment-uuid"}},
{"uuid": "crew-uuid", "deployment": {"uuid": "deployment-uuid"}},
)
== "https://app.crewai.com/crewai_plus/deployments/deployment-uuid"
is None
)


Expand Down Expand Up @@ -353,6 +353,51 @@ def test_display_deployment_info_warns_when_browser_open_raises(self):
"https://app.crewai.com/crewai_plus/deployments/128687"
)

def test_display_creation_success_resolves_deployment_page_id_from_status(self):
status_response = MagicMock()
status_response.is_success = True
status_response.json.return_value = {
"uuid": "new-uuid",
"id": 128774,
"status": "created",
}
self.mock_client.crew_status_by_uuid.return_value = status_response

with patch("sys.stdout", new=StringIO()) as fake_out:
self.deploy_command._display_creation_success(
{"uuid": "new-uuid", "status": "created"}
)
output = fake_out.getvalue()

self.assertIn("crewai deploy push --uuid new-uuid", output)
self.assertIn(
"https://app.crewai.com/crewai_plus/deployments/128774",
output,
)
self.assertNotIn(
"https://app.crewai.com/crewai_plus/deployments/new-uuid",
output,
)
self.mock_client.crew_status_by_uuid.assert_called_once_with("new-uuid")
self.mock_browser_open.assert_called_once_with(
"https://app.crewai.com/crewai_plus/deployments/128774"
)

def test_open_deployment_page_does_not_open_uuid_url_when_status_lookup_fails(
self,
):
status_response = MagicMock()
status_response.is_success = False
self.mock_client.crew_status_by_uuid.return_value = status_response

with patch("sys.stdout", new=StringIO()) as fake_out:
self.deploy_command._open_deployment_page({"uuid": "new-uuid"})
output = fake_out.getvalue()

self.mock_client.crew_status_by_uuid.assert_called_once_with("new-uuid")
self.mock_browser_open.assert_not_called()
self.assertNotIn("deployments/new-uuid", output)

def test_display_logs(self):
with patch("sys.stdout", new=StringIO()) as fake_out:
self.deploy_command._display_logs(
Expand Down
Loading