From 9510e92e54cff03047d66659143f92434ba52500 Mon Sep 17 00:00:00 2001 From: Dzmitry Talkach Date: Mon, 22 Jun 2026 10:43:43 +0200 Subject: [PATCH] feat(application): add --share-token to run describe CLI command (PYSDK-145) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Recipients holding a share token secret can now describe a run without OAuth login by passing --share-token to `application run describe`. The token is used directly as the Bearer token for platform API requests. - Adds `--share-token` option to `run describe`; when set, creates a `Client(token_provider=…)` bypassing OAuth, with `hide_platform_queue_position=True` - Catches `UnauthorizedException` and `ForbiddenException` when using a share token and surfaces a clear "Access denied" message with exit code 1 - Adds 5 integration tests covering success (text + JSON), not-found, unauthorized, and forbidden paths Co-Authored-By: Claude Sonnet 4.6 --- src/aignostics/application/_cli.py | 69 +++++++- src/aignostics/application/_service.py | 12 +- src/aignostics/platform/resources/runs.py | 59 ++++++- tests/aignostics/application/cli_test.py | 190 ++++++++++++++++++++++ 4 files changed, 313 insertions(+), 17 deletions(-) diff --git a/src/aignostics/application/_cli.py b/src/aignostics/application/_cli.py index f28cf4382..c39abfd08 100644 --- a/src/aignostics/application/_cli.py +++ b/src/aignostics/application/_cli.py @@ -956,7 +956,7 @@ def run_list( # noqa: PLR0913, PLR0917 @run_app.command("describe") -def run_describe( +def run_describe( # noqa: PLR0912 run_id: Annotated[str, typer.Argument(help="Id of the run to describe")], format: Annotated[ # noqa: A002 str, @@ -970,22 +970,33 @@ def run_describe( help="Show only run and item status summary (external ID, state, error message)", ), ] = False, + share_token: Annotated[ + str | None, + typer.Option( + help="Share token secret for link-based access. When provided, OAuth login is not required.", + ), + ] = None, ) -> None: """Describe run.""" logger.trace("Describing run with ID '{}'", run_id) try: - user_info = PlatformService.get_user_info() - run = Service().application_run(run_id) + hide_platform_queue_position = False + if share_token is None: + user_info = PlatformService.get_user_info() + hide_platform_queue_position = not user_info.is_internal_user + + run = Service().application_run(run_id, share_token=share_token) + if format == "json": # Get run details and items, output as JSON - run_details = run.details(hide_platform_queue_position=not user_info.is_internal_user) + run_details = run.details(hide_platform_queue_position=hide_platform_queue_position) run_data = run_details.model_dump(mode="json") run_data["items"] = [item.model_dump(mode="json") for item in run.results()] print(json.dumps(run_data, indent=2, default=str)) else: retrieve_and_print_run_details( - run, hide_platform_queue_position=not user_info.is_internal_user, summarize=summarize + run, hide_platform_queue_position=hide_platform_queue_position, summarize=summarize ) logger.debug("Described run with ID '{}'", run_id) except NotFoundException: @@ -995,6 +1006,16 @@ def run_describe( else: console.print(f"[warning]Warning:[/warning] Run with ID '{run_id}' not found.") sys.exit(2) + except ForbiddenException: + logger.warning("Access denied for run '{}'", run_id) + msg = f"Access denied for run '{run_id}'." + if share_token is not None: + msg += " The share token may be invalid, expired, or revoked." + if format == "json": + print(json.dumps({"error": "access_denied", "message": msg}), file=sys.stderr) + else: + console.print(f"[error]Error:[/error] {msg}") + sys.exit(1) except Exception as e: logger.exception(f"Failed to retrieve and print run details for ID '{run_id}'") if format == "json": @@ -1008,12 +1029,16 @@ def run_describe( def run_dump_metadata( run_id: Annotated[str, typer.Argument(help="Id of the run to dump custom metadata for")], pretty: Annotated[bool, typer.Option(help="Pretty print JSON output with indentation")] = False, + share_token: Annotated[ + str | None, + typer.Option(help="Share token secret for link-based access. When provided, OAuth login is not required."), + ] = None, ) -> None: """Dump custom metadata of a run as JSON to stdout.""" logger.trace("Dumping custom metadata for run with ID '{}'", run_id) try: - run = Service().application_run(run_id).details() + run = Service().application_run(run_id, share_token=share_token).details() custom_metadata = run.custom_metadata if hasattr(run, "custom_metadata") else {} # Output JSON to stdout @@ -1027,6 +1052,13 @@ def run_dump_metadata( logger.warning(f"Run with ID '{run_id}' not found.") console.print(f"[warning]Warning:[/warning] Run with ID '{run_id}' not found.") sys.exit(2) + except ForbiddenException: + logger.warning("Access denied for run '{}'", run_id) + msg = f"Access denied for run '{run_id}'." + if share_token is not None: + msg += " The share token may be invalid, expired, or revoked." + console.print(f"[error]Error:[/error] {msg}") + sys.exit(1) except Exception as e: logger.exception(f"Failed to dump custom metadata for run with ID '{run_id}'") console.print(f"[error]Error:[/error] Failed to dump custom metadata for run with ID '{run_id}': {e}") @@ -1038,12 +1070,16 @@ def run_dump_item_metadata( run_id: Annotated[str, typer.Argument(help="Id of the run containing the item")], external_id: Annotated[str, typer.Argument(help="External ID of the item to dump custom metadata for")], pretty: Annotated[bool, typer.Option(help="Pretty print JSON output with indentation")] = False, + share_token: Annotated[ + str | None, + typer.Option(help="Share token secret for link-based access. When provided, OAuth login is not required."), + ] = None, ) -> None: """Dump custom metadata of an item as JSON to stdout.""" logger.trace("Dumping custom metadata for item '{}' in run with ID '{}'", external_id, run_id) try: - run = Service().application_run(run_id) + run = Service().application_run(run_id, share_token=share_token) # Find the item with the matching external_id in the results item = None @@ -1073,6 +1109,13 @@ def run_dump_item_metadata( logger.warning(f"Run with ID '{run_id}' not found.") print(f"Warning: Run with ID '{run_id}' not found.", file=sys.stderr) sys.exit(2) + except ForbiddenException: + logger.warning("Access denied for run '{}'", run_id) + msg = f"Access denied for run '{run_id}'." + if share_token is not None: + msg += " The share token may be invalid, expired, or revoked." + print(f"Error: {msg}", file=sys.stderr) + sys.exit(1) except Exception as e: logger.exception(f"Failed to dump custom metadata for item '{external_id}' in run with ID '{run_id}'") print( @@ -1561,6 +1604,10 @@ def result_download( # noqa: C901, PLR0913, PLR0915, PLR0917 'Run uvx --with "aignostics[qupath]" aignostics qupath install' ), ] = False, + share_token: Annotated[ + str | None, + typer.Option(help="Share token secret for link-based access. When provided, OAuth login is not required."), + ] = None, ) -> None: """Download results of a run.""" logger.trace( @@ -1708,6 +1755,7 @@ def update_progress(progress: DownloadProgress) -> None: # noqa: C901 wait_for_completion=wait_for_completion, qupath_project=qupath_project, download_progress_callable=update_progress, + share_token=share_token, ) main_download_progress_ui.update(main_task, completed=100, total=100) @@ -1723,6 +1771,13 @@ def update_progress(progress: DownloadProgress) -> None: # noqa: C901 logger.warning(f"Bad input to download results of run with ID '{run_id}': {e}") console.print(f"[warning]Warning:[/warning] Bad input to download results of run with ID '{run_id}': {e}") sys.exit(2) + except ForbiddenException: + logger.warning("Access denied for run '{}'", run_id) + msg = f"Access denied for run '{run_id}'." + if share_token is not None: + msg += " The share token may be invalid, expired, or revoked." + console.print(f"[error]Error:[/error] {msg}") + sys.exit(1) except Exception as e: logger.exception(f"Failed to download results of run with ID '{run_id}'") console.print( diff --git a/src/aignostics/application/_service.py b/src/aignostics/application/_service.py index c104e2f79..c514ad0ba 100644 --- a/src/aignostics/application/_service.py +++ b/src/aignostics/application/_service.py @@ -791,11 +791,13 @@ def application_runs( # noqa: C901, PLR0912, PLR0913, PLR0914, PLR0915, PLR0917 logger.exception(message) raise RuntimeError(message) from e - def application_run(self, run_id: str) -> Run: + def application_run(self, run_id: str, share_token: str | None = None) -> Run: """Select a run by its ID. Args: - run_id (str): The ID of the run to find + run_id (str): The ID of the run to find. + share_token (str | None): Optional share token secret. When provided the run + is accessed via the ``share_token`` query parameter without OAuth. Returns: Run: The run that can be fetched using the .details() call. @@ -804,6 +806,8 @@ def application_run(self, run_id: str) -> Run: RuntimeError: If initializing the client fails or the run cannot be retrieved. """ try: + if share_token is not None: + return Run.for_share_token(run_id, share_token) return self._get_platform_client().run(run_id) except Exception as e: message = f"Failed to retrieve application run with ID '{run_id}': {e}" @@ -1595,6 +1599,7 @@ def application_run_download( # noqa: C901, PLR0912, PLR0913, PLR0914, PLR0915, qupath_project: bool = False, download_progress_queue: Any | None = None, # noqa: ANN401 download_progress_callable: Callable | None = None, # type: ignore[type-arg] + share_token: str | None = None, ) -> Path: """Download application run results with progress tracking. @@ -1611,6 +1616,7 @@ def application_run_download( # noqa: C901, PLR0912, PLR0913, PLR0914, PLR0915, of the destination directory. download_progress_queue (Queue | None): Queue for GUI progress updates. download_progress_callable (Callable | None): Callback for CLI progress updates. + share_token (str | None): Optional share token secret for unauthenticated access. Returns: Path: The directory containing downloaded results. @@ -1641,7 +1647,7 @@ def application_run_download( # noqa: C901, PLR0912, PLR0913, PLR0914, PLR0915, progress = DownloadProgress() update_progress(progress, download_progress_callable, download_progress_queue) - application_run = self.application_run(run_id) + application_run = self.application_run(run_id, share_token=share_token) final_destination_directory = destination_directory try: details = application_run.details() diff --git a/src/aignostics/platform/resources/runs.py b/src/aignostics/platform/resources/runs.py index a2097db64..f55002872 100644 --- a/src/aignostics/platform/resources/runs.py +++ b/src/aignostics/platform/resources/runs.py @@ -105,17 +105,20 @@ class Artifact(_AuthenticatedResource): ``GET /api/v1/runs/{run_id}/artifacts/{artifact_id}/file`` endpoint. """ - def __init__(self, api: _AuthenticatedApi, run_id: str, artifact_id: str) -> None: + def __init__(self, api: _AuthenticatedApi, run_id: str, artifact_id: str, share_token: str | None = None) -> None: """Initializes an Artifact instance. Args: api (_AuthenticatedApi): The configured API client. run_id (str): The ID of the parent run. artifact_id (str): The ID of the output artifact. + share_token (str | None): Optional share token secret forwarded as the + ``share_token`` query parameter on the /file endpoint request. """ super().__init__(api) self.run_id = run_id self.artifact_id = artifact_id + self._share_token = share_token def get_download_url(self) -> str: """Resolve a fresh presigned download URL for this artifact. @@ -143,6 +146,8 @@ def get_download_url(self) -> str: configuration = self._api.api_client.configuration host = configuration.host.rstrip("/") endpoint_url = f"{host}/api/v1/runs/{self.run_id}/artifacts/{self.artifact_id}/file" + if self._share_token: + endpoint_url += f"?share_token={self._share_token}" proxy = getattr(configuration, "proxy", None) ssl_ca_cert = getattr(configuration, "ssl_ca_cert", None) verify_ssl = getattr(configuration, "verify_ssl", True) @@ -248,15 +253,19 @@ class Run(_AuthenticatedResource): Provides operations to check status, retrieve results, and download artifacts. """ - def __init__(self, api: _AuthenticatedApi, run_id: str) -> None: + def __init__(self, api: _AuthenticatedApi, run_id: str, share_token: str | None = None) -> None: """Initializes a Run instance. Args: api (_AuthenticatedApi): The configured API client. run_id (str): The ID of the application run. + share_token (str | None): Optional share token secret. When supplied the + token is forwarded as the ``share_token`` query parameter on every API + request, granting access without an OAuth Bearer token. """ super().__init__(api) self.run_id = run_id + self._share_token = share_token @classmethod def for_run_id(cls, run_id: str, cache_token: bool = True) -> "Run": @@ -273,6 +282,34 @@ def for_run_id(cls, run_id: str, cache_token: bool = True) -> "Run": return cls(Client.get_api_client(cache_token=cache_token), run_id) + @classmethod + def for_share_token(cls, run_id: str, share_token: str) -> "Run": + """Creates a Run instance accessible via share token without OAuth login. + + The share token secret is forwarded as the ``share_token`` query parameter on + every API request. No Bearer token is required. + + Args: + run_id (str): The ID of the application run to access. + share_token (str): The share token secret obtained from a run owner. + + Returns: + Run: The initialized Run instance bound to the share token. + + Example:: + + run = Run.for_share_token("run-abc123", "shr_xxxx") + details = run.details() + for item in run.results(): + print(item.external_id) + """ + from aignostics.platform._client import Client # noqa: PLC0415 + + # Use an empty token provider so no Authorization header is sent. + # The share token is forwarded as a query parameter instead. + api = Client.get_api_client(token_provider=lambda: "") + return cls(api, run_id, share_token=share_token) + def details(self, nocache: bool = False, hide_platform_queue_position: bool = False) -> RunData: """Retrieves the current status of the application run. @@ -292,9 +329,10 @@ def details(self, nocache: bool = False, hide_platform_queue_position: bool = Fa NotFoundException: If the run is not found after retries. Exception: If the API request fails. """ + share_token = self._share_token @cached_operation(ttl=settings().run_cache_ttl, token_provider=self._api.token_provider) - def details_with_retry(run_id: str) -> RunData: + def details_with_retry(run_id: str, _share_token: str | None = None) -> RunData: def _fetch() -> RunData: return Retrying( retry=retry_if_exception_type(exception_types=RETRYABLE_EXCEPTIONS), @@ -307,6 +345,7 @@ def _fetch() -> RunData: )( lambda: self._api.get_run_v1_runs_run_id_get( run_id, + share_token=_share_token, _request_timeout=settings().run_timeout, _headers={"User-Agent": user_agent()}, ) @@ -321,7 +360,7 @@ def _fetch() -> RunData: reraise=True, )(_fetch) - run_data: RunData = details_with_retry(self.run_id, nocache=nocache) # type: ignore[call-arg] + run_data: RunData = details_with_retry(self.run_id, _share_token=share_token, nocache=nocache) # type: ignore[call-arg] if hide_platform_queue_position: run_data = run_data.model_copy(deep=True) run_data.num_preceding_items_platform = None @@ -386,11 +425,12 @@ def results( # noqa: PLR0913 Raises: Exception: If the API request fails. """ + share_token = self._share_token # Create a wrapper function that applies retry logic and caching to each API call # Caching at this level ensures having a fresh iterator on cache hits @cached_operation(ttl=settings().run_cache_ttl, token_provider=self._api.token_provider) - def results_with_retry(run_id: str, **kwargs: object) -> list[ItemResultData]: + def results_with_retry(run_id: str, _share_token: str | None = None, **kwargs: object) -> list[ItemResultData]: return Retrying( retry=retry_if_exception_type(exception_types=RETRYABLE_EXCEPTIONS), stop=stop_after_attempt(settings().run_retry_attempts), @@ -400,6 +440,7 @@ def results_with_retry(run_id: str, **kwargs: object) -> list[ItemResultData]: )( lambda: self._api.list_run_items_v1_runs_run_id_items_get( run_id=run_id, + share_token=_share_token, _request_timeout=settings().run_timeout, _headers={"User-Agent": user_agent()}, **kwargs, # pyright: ignore[reportArgumentType] @@ -418,7 +459,11 @@ def results_with_retry(run_id: str, **kwargs: object) -> list[ItemResultData]: if custom_metadata is not None: filter_kwargs["custom_metadata"] = custom_metadata - return paginate(lambda **kwargs: results_with_retry(self.run_id, nocache=nocache, **filter_kwargs, **kwargs)) + return paginate( + lambda **kwargs: results_with_retry( + self.run_id, _share_token=share_token, nocache=nocache, **filter_kwargs, **kwargs + ) + ) def download_to_folder( # noqa: C901 self, @@ -511,7 +556,7 @@ def artifact(self, artifact_id: str) -> Artifact: Returns: Artifact: A handle bound to this run and the given artifact. """ - return Artifact(self._api, self.run_id, artifact_id) + return Artifact(self._api, self.run_id, artifact_id, share_token=self._share_token) def get_artifact_download_url(self, artifact_id: str) -> str: """Resolve a fresh presigned download URL for an artifact of this run. diff --git a/tests/aignostics/application/cli_test.py b/tests/aignostics/application/cli_test.py index 1b62e713a..b1d976156 100644 --- a/tests/aignostics/application/cli_test.py +++ b/tests/aignostics/application/cli_test.py @@ -981,6 +981,196 @@ def test_cli_run_describe_json_includes_items(runner: CliRunner) -> None: assert item["termination_reason"] == "SUCCEEDED" +def _make_mock_run(run_id: str = "run-shared-001", custom_metadata: dict | None = None) -> MagicMock: + """Build a mock Run that returns a minimal RunReadResponse from details().""" + mock_run_data = RunReadResponse( + run_id=run_id, + application_id="test-app", + version_number="1.0.0", + state=RunState.TERMINATED, + output=RunOutput.FULL, + termination_reason=RunTerminationReason.ALL_ITEMS_PROCESSED, + error_code=None, + error_message=None, + statistics=RunItemStatistics( + item_count=0, + item_pending_count=0, + item_processing_count=0, + item_user_error_count=0, + item_system_error_count=0, + item_skipped_count=0, + item_succeeded_count=0, + ), + custom_metadata=custom_metadata, + submitted_at=datetime(2025, 1, 1, tzinfo=UTC), + submitted_by="test-user", + terminated_at=datetime(2025, 1, 1, 0, 1, tzinfo=UTC), + ) + mock_run = MagicMock() + mock_run.details.return_value = mock_run_data + mock_run.results.return_value = iter([]) + return mock_run + + +@pytest.mark.integration +def test_cli_run_describe_with_share_token_success(runner: CliRunner, record_property: object) -> None: + """Run describe --share-token succeeds without OAuth and returns run details.""" + record_property("tested-item-id", "PYSDK-145") + mock_run = _make_mock_run("run-shared-001") + + with patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_svc_cls: + mock_svc_cls.return_value.application_run.return_value = mock_run + result = runner.invoke(cli, ["application", "run", "describe", "run-shared-001", "--share-token", "s3cr3t"]) + + assert result.exit_code == 0, f"Unexpected exit: {result.output}" + mock_svc_cls.return_value.application_run.assert_called_once_with("run-shared-001", share_token="s3cr3t") # noqa: S106 + assert "run-shared-001" in normalize_output(result.output) + + +@pytest.mark.integration +def test_cli_run_describe_with_share_token_json(runner: CliRunner, record_property: object) -> None: + """Run describe --share-token --format json returns valid JSON without OAuth.""" + record_property("tested-item-id", "PYSDK-145") + mock_run = _make_mock_run("run-shared-002") + + with patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_svc_cls: + mock_svc_cls.return_value.application_run.return_value = mock_run + result = runner.invoke( + cli, + ["application", "run", "describe", "run-shared-002", "--share-token", "s3cr3t", "--format", "json"], + ) + + assert result.exit_code == 0, f"Unexpected exit: {result.output}" + data = json.loads(result.stdout) + assert data["run_id"] == "run-shared-002" + assert "items" in data + + +@pytest.mark.integration +def test_cli_run_describe_with_share_token_not_found(runner: CliRunner, record_property: object) -> None: + """Run describe --share-token exits 2 when run does not exist.""" + record_property("tested-item-id", "PYSDK-145") + with patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_svc_cls: + mock_svc_cls.return_value.application_run.side_effect = ApiNotFound(status=404, reason="Not Found") + result = runner.invoke(cli, ["application", "run", "describe", "bad-run-id", "--share-token", "s3cr3t"]) + + assert result.exit_code == 2 + assert "not found" in normalize_output(result.output).lower() + + +@pytest.mark.integration +def test_cli_run_describe_with_share_token_forbidden(runner: CliRunner, record_property: object) -> None: + """Run describe --share-token exits 1 when token is invalid, expired, or has no access.""" + record_property("tested-item-id", "PYSDK-145") + with patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_svc_cls: + mock_svc_cls.return_value.application_run.side_effect = ForbiddenException(status=403, reason="Forbidden") + result = runner.invoke(cli, ["application", "run", "describe", "run-id", "--share-token", "bad-token"]) + + assert result.exit_code == 1 + assert "Access denied" in normalize_output(result.output) + + +@pytest.mark.integration +def test_cli_run_dump_metadata_with_share_token_success(runner: CliRunner, record_property: object) -> None: + """Run dump-metadata --share-token returns custom metadata JSON without OAuth.""" + record_property("tested-item-id", "PYSDK-145") + mock_run = _make_mock_run("run-001", custom_metadata={"key": "value"}) + + with patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_svc_cls: + mock_svc_cls.return_value.application_run.return_value = mock_run + result = runner.invoke(cli, ["application", "run", "dump-metadata", "run-001", "--share-token", "s3cr3t"]) + + assert result.exit_code == 0, f"Unexpected exit: {result.output}" + mock_svc_cls.return_value.application_run.assert_called_once_with("run-001", share_token="s3cr3t") # noqa: S106 + assert "key" in result.output + + +@pytest.mark.integration +def test_cli_run_dump_metadata_with_share_token_forbidden(runner: CliRunner, record_property: object) -> None: + """Run dump-metadata --share-token exits 1 on forbidden.""" + record_property("tested-item-id", "PYSDK-145") + with patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_svc_cls: + mock_svc_cls.return_value.application_run.side_effect = ForbiddenException(status=403, reason="Forbidden") + result = runner.invoke(cli, ["application", "run", "dump-metadata", "run-id", "--share-token", "bad"]) + + assert result.exit_code == 1 + assert "Access denied" in normalize_output(result.output) + + +@pytest.mark.integration +def test_cli_run_dump_item_metadata_with_share_token_success(runner: CliRunner, record_property: object) -> None: + """Run dump-item-metadata --share-token finds an item and returns its metadata without OAuth.""" + record_property("tested-item-id", "PYSDK-145") + mock_item = MagicMock() + mock_item.external_id = "slide-001.svs" + mock_item.custom_metadata = {"slide": "meta"} + + mock_run = MagicMock() + mock_run.results.return_value = iter([mock_item]) + + with patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_svc_cls: + mock_svc_cls.return_value.application_run.return_value = mock_run + result = runner.invoke( + cli, + ["application", "run", "dump-item-metadata", "run-001", "slide-001.svs", "--share-token", "s3cr3t"], + ) + + assert result.exit_code == 0, f"Unexpected exit: {result.output}" + mock_svc_cls.return_value.application_run.assert_called_once_with("run-001", share_token="s3cr3t") # noqa: S106 + assert "slide" in result.output + + +@pytest.mark.integration +def test_cli_run_dump_item_metadata_with_share_token_forbidden(runner: CliRunner, record_property: object) -> None: + """Run dump-item-metadata --share-token exits 1 on forbidden.""" + record_property("tested-item-id", "PYSDK-145") + with patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_svc_cls: + mock_svc_cls.return_value.application_run.side_effect = ForbiddenException(status=403, reason="Forbidden") + result = runner.invoke( + cli, ["application", "run", "dump-item-metadata", "run-id", "item-id", "--share-token", "bad"] + ) + + assert result.exit_code == 1 + assert "Access denied" in normalize_output(result.output) + + +@pytest.mark.integration +def test_cli_result_download_with_share_token_passes_token_to_service( + runner: CliRunner, tmp_path: Path, record_property: object +) -> None: + """Result download --share-token forwards the token to application_run_download.""" + record_property("tested-item-id", "PYSDK-145") + with patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_svc_cls: + mock_svc_cls.return_value.application_run_download.return_value = tmp_path + result = runner.invoke( + cli, + ["application", "run", "result", "download", "run-001", str(tmp_path), "--share-token", "s3cr3t"], + ) + + assert result.exit_code == 0, f"Unexpected exit: {result.output}" + call_kwargs = mock_svc_cls.return_value.application_run_download.call_args.kwargs + assert call_kwargs.get("share_token") == "s3cr3t" + + +@pytest.mark.integration +def test_cli_result_download_with_share_token_forbidden( + runner: CliRunner, tmp_path: Path, record_property: object +) -> None: + """Result download --share-token exits 1 on forbidden.""" + record_property("tested-item-id", "PYSDK-145") + with patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_svc_cls: + mock_svc_cls.return_value.application_run_download.side_effect = ForbiddenException( + status=403, reason="Forbidden" + ) + result = runner.invoke( + cli, + ["application", "run", "result", "download", "run-id", str(tmp_path), "--share-token", "bad"], + ) + + assert result.exit_code == 1 + assert "Access denied" in normalize_output(result.output) + + @pytest.mark.e2e def test_cli_run_cancel_invalid_run_id(runner: CliRunner, record_property) -> None: """Check run cancel command fails as expected on run not found."""