Skip to content
Open
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
15 changes: 15 additions & 0 deletions plane/api/workspaces.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
from ..models.workspaces import (
PaginatedWorkspaceMemberResponse,
ProjectRoleDistribution,
WorkspaceAssetDownload,
WorkspaceFeature,
WorkspaceMember,
)
Expand Down Expand Up @@ -68,6 +69,20 @@ def get_project_role_distribution(self, workspace_slug: str) -> ProjectRoleDistr
response = self._get(f"{workspace_slug}/project-role-distribution")
return ProjectRoleDistribution.model_validate(response)

def get_asset(self, workspace_slug: str, asset_id: str) -> WorkspaceAssetDownload:
"""Get a presigned download URL for a workspace asset.

Covers any uploaded asset in the workspace, including images embedded in
work item descriptions, which are not listed as work item attachments.
The URL is time-limited and needs no Plane auth to fetch.

Args:
workspace_slug: The workspace slug identifier
asset_id: UUID of the asset (the ``src`` of an ``image-component`` tag)
"""
response = self._get(f"{workspace_slug}/assets/{asset_id}")
return WorkspaceAssetDownload.model_validate(response)

def get_features(self, workspace_slug: str) -> WorkspaceFeature:
"""Get features of a workspace.

Expand Down
16 changes: 16 additions & 0 deletions plane/models/workspaces.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,22 @@ class PaginatedWorkspaceMemberResponse(PaginatedResponse):
results: list[WorkspaceMember]


class WorkspaceAssetDownload(BaseModel):
"""Presigned download details for a workspace asset.

Returned by Workspaces.get_asset(). ``asset_url`` is time-limited and can be
fetched without Plane authentication. It is None when the server could not
presign the stored file.
"""

model_config = ConfigDict(extra="allow", populate_by_name=True)

asset_id: str
asset_url: str | None = None
asset_name: str | None = None
asset_type: str | None = None


class WorkspaceFeature(BaseModel):
"""Workspace feature model.

Expand Down
92 changes: 91 additions & 1 deletion tests/unit/test_workspaces.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,18 @@
"""Unit tests for Workspaces API resource (smoke tests with real HTTP requests)."""

import json
from unittest.mock import MagicMock

import pytest

from plane.client import PlaneClient
from plane.errors.errors import HttpError
from plane.models.query_params import MemberListQueryParams, MemberQueryParams
from plane.models.workspaces import ProjectRoleDistributionEntry, WorkspaceMember
from plane.models.workspaces import (
ProjectRoleDistributionEntry,
WorkspaceAssetDownload,
WorkspaceMember,
)


class TestWorkspacesAPI:
Expand Down Expand Up @@ -91,3 +101,83 @@ def test_update_features(self, client: PlaneClient, workspace_slug: str) -> None
assert updated is not None
assert hasattr(updated, "initiatives")
assert updated.initiatives is True


class TestWorkspaceAssetsOffline:
"""Workspaces.get_asset against a stubbed HTTP session (no network, no credentials)."""

ASSET_ID = "0b1f6d1e-6f0e-4f0a-9d63-2f3c1f6f2a11"

@pytest.fixture
def offline_client(self) -> PlaneClient:
return PlaneClient(base_url="https://plane.test", api_key="test-key")

@staticmethod
def _response(status_code: int, body: dict | None = None, reason: str = "OK") -> MagicMock:
resp = MagicMock()
resp.status_code = status_code
resp.reason = reason
resp.content = json.dumps(body).encode() if body is not None else b""
resp.text = json.dumps(body) if body is not None else ""
resp.headers = {"content-type": "application/json"}
resp.json.return_value = body
return resp

def test_get_asset_returns_download_details(self, offline_client: PlaneClient) -> None:
body = {
"asset_id": self.ASSET_ID,
"asset_url": "https://bucket.s3.example/signed?X-Amz-Signature=abc",
"asset_name": "screenshot.png",
"asset_type": "image/png",
}
session_get = MagicMock(return_value=self._response(200, body))
offline_client.workspaces.session.get = session_get

asset = offline_client.workspaces.get_asset("acme", self.ASSET_ID)

assert isinstance(asset, WorkspaceAssetDownload)
assert asset.asset_id == self.ASSET_ID
assert asset.asset_url == body["asset_url"]
assert asset.asset_name == "screenshot.png"
assert asset.asset_type == "image/png"

def test_get_asset_hits_workspace_asset_detail_route(self, offline_client: PlaneClient) -> None:
body = {"asset_id": self.ASSET_ID, "asset_url": "https://signed.example/x"}
session_get = MagicMock(return_value=self._response(200, body))
offline_client.workspaces.session.get = session_get

offline_client.workspaces.get_asset("acme", self.ASSET_ID)

(url,), kwargs = session_get.call_args
assert url.endswith(f"/api/v1/workspaces/acme/assets/{self.ASSET_ID}/")
assert kwargs["headers"]["X-Api-Key"] == "test-key"

def test_get_asset_tolerates_a_null_url(self, offline_client: PlaneClient) -> None:
"""The server answers 200 with a null url when presigning fails."""
body = {"asset_id": self.ASSET_ID, "asset_url": None}
offline_client.workspaces.session.get = MagicMock(return_value=self._response(200, body))

assert offline_client.workspaces.get_asset("acme", self.ASSET_ID).asset_url is None

def test_get_asset_optional_fields_default_to_none(self, offline_client: PlaneClient) -> None:
body = {"asset_id": self.ASSET_ID, "asset_url": "https://signed.example/x"}
offline_client.workspaces.session.get = MagicMock(return_value=self._response(200, body))

asset = offline_client.workspaces.get_asset("acme", self.ASSET_ID)

assert asset.asset_name is None
assert asset.asset_type is None

@pytest.mark.parametrize("status_code, reason", [(400, "Bad Request"), (404, "Not Found")])
def test_get_asset_error_raises_http_error(
self, offline_client: PlaneClient, status_code: int, reason: str
) -> None:
body = {"error": "Asset not yet uploaded"}
offline_client.workspaces.session.get = MagicMock(
return_value=self._response(status_code, body, reason)
)

with pytest.raises(HttpError) as exc_info:
offline_client.workspaces.get_asset("acme", self.ASSET_ID)

assert exc_info.value.status_code == status_code
Loading