diff --git a/plane/api/workspaces.py b/plane/api/workspaces.py index 3b5f9fc..0cf5a43 100644 --- a/plane/api/workspaces.py +++ b/plane/api/workspaces.py @@ -6,6 +6,7 @@ from ..models.workspaces import ( PaginatedWorkspaceMemberResponse, ProjectRoleDistribution, + WorkspaceAssetDownload, WorkspaceFeature, WorkspaceMember, ) @@ -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. diff --git a/plane/models/workspaces.py b/plane/models/workspaces.py index e7aabf1..f084d98 100644 --- a/plane/models/workspaces.py +++ b/plane/models/workspaces.py @@ -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. diff --git a/tests/unit/test_workspaces.py b/tests/unit/test_workspaces.py index e787209..bea5d11 100644 --- a/tests/unit/test_workspaces.py +++ b/tests/unit/test_workspaces.py @@ -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: @@ -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