diff --git a/sagemaker-serve/src/sagemaker/serve/bedrock_model_builder.py b/sagemaker-serve/src/sagemaker/serve/bedrock_model_builder.py index 902b6dcfd1..dfa1ebbfc4 100644 --- a/sagemaker-serve/src/sagemaker/serve/bedrock_model_builder.py +++ b/sagemaker-serve/src/sagemaker/serve/bedrock_model_builder.py @@ -26,6 +26,8 @@ build_source_tag, find_active_bedrock_deployment_for_model, find_existing_bedrock_model, + find_existing_imported_model, + find_existing_model_import_job, ) from sagemaker.core.training.utils import ( build_nova_manifest_s3_uri, @@ -167,7 +169,7 @@ def _get_sagemaker_client(self): self._sagemaker_client = self.boto_session.client("sagemaker") return self._sagemaker_client - def _resolve_model_source_id(self) -> Optional[str]: + def _resolve_nova_model_source_id(self) -> Optional[str]: """Determine the model source identifier for reuse lookups. Resolution order: @@ -349,7 +351,7 @@ def deploy( sagemaker_session=self.sagemaker_session, ) - source_id = self._resolve_model_source_id() + source_id = self._resolve_nova_model_source_id() if source_id and reuse_resources: existing_arn = find_existing_bedrock_model( @@ -435,23 +437,104 @@ def deploy( role_type="bedrock", sagemaker_session=self.sagemaker_session, ) - model_data_source = {"s3DataSource": {"s3Uri": self.s3_model_artifacts}} + + # Resolve model source identifier for reuse tagging. + # Priority: model package ARN > S3 artifact URI > None (with warning). + oss_source_id = None + if self.model_package: + mp_arn = getattr(self.model_package, "model_package_arn", None) + if mp_arn and isinstance(mp_arn, str): + oss_source_id = mp_arn + if not oss_source_id and self.s3_model_artifacts and isinstance(self.s3_model_artifacts, str): + oss_source_id = self.s3_model_artifacts + if not oss_source_id: + logger.warning( + "Cannot determine model source identifier for OSS model resource reuse. " + "Neither Model package ARN nor model artifacts S3 URI is available. " + ) + + # Reuse: first look for an already-completed imported model, then + # fall back to an in-progress import job for the same source. + if oss_source_id and reuse_resources: + # 1. A completed imported model can be reused directly; there is + # no import job to wait on. + model_arn = find_existing_imported_model( + self._get_bedrock_client(), + oss_source_id, + ) + if model_arn: + logger.info( + "Reusing existing imported model %s (matched model-source tag). " + "No new import job was created. Pass reuse_resources=False to " + "force a new import.", + model_arn, + ) + model_details = self._get_bedrock_client().get_imported_model( + modelIdentifier=model_arn + ) + self._imported_model_id = model_details.get("modelName") + return model_details + + # 2. Otherwise, an import job may already be running for this + # source; wait for it to complete instead of starting a new one. + job_arn = find_existing_model_import_job( + self._get_bedrock_client(), + oss_source_id, + ) + if job_arn: + logger.info( + "Reusing in-progress import job %s (matched model-source tag). " + "No new import job was created. Pass reuse_resources=False to " + "force a new import.", + job_arn, + ) + self._wait_for_import_job_complete(job_arn) + job_details = self._get_bedrock_client().get_model_import_job( + jobIdentifier=job_arn + ) + self._imported_model_id = job_details.get("importedModelName") + return job_details + + # If artifacts are a tar.gz, extract to S3 first (Bedrock requires uncompressed format) if self.s3_model_artifacts.endswith(".tar.gz") or self.s3_model_artifacts.endswith(".tar.gz/"): extracted_uri = self._extract_tar_gz_to_s3(self.s3_model_artifacts.rstrip("/")) resolved_uri = self._resolve_hf_model_path(extracted_uri) model_data_source = {"s3DataSource": {"s3Uri": resolved_uri}} + else: + resolved_uri = self._resolve_hf_model_path(self.s3_model_artifacts) + model_data_source = {"s3DataSource": {"s3Uri": resolved_uri}} + # Auto-generate job_name if not provided if not job_name: import time job_name = f"{imported_model_name or 'import'}-{int(time.time())}" + + # Inject the source tag into both the imported model tags and the + # import job tags. The model tags let a completed model be reused; + # the job tags let an in-progress import job be discovered and reused + # (reuse discovery matches the tag on the job ARN while the model + # does not yet exist). + merged_imported_tags = list(imported_model_tags) if imported_model_tags else [] + merged_job_tags = list(job_tags) if job_tags else [] + if oss_source_id: + source_tag = build_source_tag(oss_source_id) + merged_imported_tags = [ + t for t in merged_imported_tags if t.get("key") != source_tag["key"] + ] + merged_imported_tags.append(source_tag) + merged_job_tags = [ + t for t in merged_job_tags if t.get("key") != source_tag["key"] + ] + merged_job_tags.append(source_tag) + params = { "jobName": job_name, "importedModelName": imported_model_name, "roleArn": role_arn, "modelDataSource": model_data_source, - "jobTags": job_tags, - "importedModelTags": imported_model_tags, + "jobTags": merged_job_tags if merged_job_tags else None, + "importedModelTags": merged_imported_tags if merged_imported_tags else None, "clientRequestToken": client_request_token, "importedModelKmsKeyId": imported_model_kms_key_id, } @@ -902,6 +985,18 @@ def _resolve_hf_model_path(self, s3_uri: str) -> str: print(f"[BedrockModelBuilder] Base s3_uri from model package: {s3_uri}") + # Idempotency guard: if the given URI already points directly at a + # resolved model directory (contains config.json), it is already + # correct. Return it as-is instead of appending another checkpoints/ + # prefix, so repeated calls are a no-op. + base_config_key = parsed_base.path.lstrip("/") + "config.json" + try: + s3_client.head_object(Bucket=bucket, Key=base_config_key) + logger.info("s3_uri already resolved (config.json present) at %s", s3_uri) + return s3_uri.rstrip("/") + except Exception as e: + logger.debug(f"[BedrockModelBuilder]{s3_uri} Not a resolved dir, continuing: {e}") + hf_merged_uri = s3_uri + "checkpoints/hf_merged/" merged_config_key = urlparse(hf_merged_uri).path.lstrip("/") + "config.json" print(f"[BedrockModelBuilder] Probing for hf_merged: s3://{bucket}/{merged_config_key}") diff --git a/sagemaker-serve/src/sagemaker/serve/model_reuse.py b/sagemaker-serve/src/sagemaker/serve/model_reuse.py index a0f5560f11..e3c0a1efa7 100644 --- a/sagemaker-serve/src/sagemaker/serve/model_reuse.py +++ b/sagemaker-serve/src/sagemaker/serve/model_reuse.py @@ -153,6 +153,75 @@ def find_active_bedrock_deployment_for_model(bedrock_client, model_arn: str) -> return None +def find_existing_imported_model( + bedrock_client, + source_id: str, +) -> Optional[str]: + """Find an existing completed Bedrock imported model matching a source id. + + Enumerates imported models (via ``list_imported_models``) and matches on the + ``sagemaker.amazonaws.com/model-source`` tag. + + Args: + bedrock_client: A boto3 Bedrock client. + source_id: Raw source identifier (will be normalized). + + Returns: + The imported-model ARN (``.../imported-model/...``) if a match is found, + None otherwise. + """ + tag_value = normalize_tag_value(source_id) + + try: + resource_arn = _find_imported_model_arn_by_tag(bedrock_client, tag_value) + except ClientError as e: + _reraise_if_access_denied(e, "bedrock:ListTagsForResource") + logger.warning("Could not list Bedrock imported models: %s. Proceeding without.", e) + return None + except Exception as e: + logger.warning("Could not list Bedrock imported models: %s. Proceeding without.", e) + return None + + return resource_arn + + +def find_existing_model_import_job( + bedrock_client, + source_id: str, +) -> Optional[str]: + """Find an in-progress Bedrock model import job matching a source id. + + Enumerates in-progress import jobs (via ``list_model_import_jobs``) and + matches on the ``sagemaker.amazonaws.com/model-source`` tag. Use this when + ``find_existing_imported_model`` returns None to detect an import that is + already running for the same source. + + Args: + bedrock_client: A boto3 Bedrock client. + source_id: Raw source identifier (will be normalized). + + Returns: + The import-job ARN (``.../model-import-job/...``) if a matching + in-progress job is found, None otherwise. + """ + tag_value = normalize_tag_value(source_id) + + try: + job_arn = _find_in_progress_import_job_by_tag(bedrock_client, tag_value) + except ClientError as e: + _reraise_if_access_denied(e, "bedrock:ListTagsForResource") + logger.warning("Could not list Bedrock import jobs: %s. Proceeding without.", e) + return None + except Exception as e: + logger.warning("Could not list Bedrock import jobs: %s. Proceeding without.", e) + return None + + if job_arn: + logger.info("Found in-progress import job %s with matching model-source tag.", job_arn) + + return job_arn + + def find_existing_sagemaker_endpoint( sagemaker_client, source_id: str, @@ -211,6 +280,47 @@ def _find_bedrock_model_arn_by_tag(bedrock_client, tag_value: str) -> Optional[s return None +def _find_imported_model_arn_by_tag(bedrock_client, tag_value: str) -> Optional[str]: + """Return the ARN of the first Bedrock imported model carrying the source tag.""" + next_token = None + while True: + kwargs = {"nextToken": next_token} if next_token else {} + response = bedrock_client.list_imported_models(**kwargs) + for summary in response.get("modelSummaries", []): + arn = summary.get("modelArn") + if arn and _bedrock_resource_has_tag(bedrock_client, arn, tag_value): + return arn + next_token = response.get("nextToken") + if not next_token: + return None + + +# The Bedrock ListModelImportJobs API only accepts the enum values +# {Completed, InProgress, Failed} for statusEquals. +_IMPORT_JOB_IN_PROGRESS_STATUSES = {"InProgress"} + +def _find_in_progress_import_job_by_tag(bedrock_client, tag_value: str) -> Optional[str]: + """Return the job ARN of an in-progress import job carrying the source tag. + + Searches jobs in the InProgress state. + """ + for status_filter in _IMPORT_JOB_IN_PROGRESS_STATUSES: + next_token = None + while True: + kwargs = {"statusEquals": status_filter} + if next_token: + kwargs["nextToken"] = next_token + response = bedrock_client.list_model_import_jobs(**kwargs) + for summary in response.get("modelImportJobSummaries", []): + job_arn = summary.get("jobArn") + if job_arn and _bedrock_resource_has_tag(bedrock_client, job_arn, tag_value): + return job_arn + next_token = response.get("nextToken") + if not next_token: + break + return None + + def _bedrock_resource_has_tag(bedrock_client, resource_arn: str, tag_value: str) -> bool: """Return True if the Bedrock resource carries the source tag with tag_value.""" tags = bedrock_client.list_tags_for_resource(resourceARN=resource_arn).get("tags", []) diff --git a/sagemaker-serve/tests/unit/test_bedrock_model_builder.py b/sagemaker-serve/tests/unit/test_bedrock_model_builder.py index 36a757ceb0..a244770491 100644 --- a/sagemaker-serve/tests/unit/test_bedrock_model_builder.py +++ b/sagemaker-serve/tests/unit/test_bedrock_model_builder.py @@ -780,7 +780,9 @@ def test_s3_uri_string_without_custom_model_name_uses_oss_path(self): b._bedrock_client.create_model_import_job.assert_called_once() kw = b._bedrock_client.create_model_import_job.call_args[1] - assert kw["modelDataSource"] == {"s3DataSource": {"s3Uri": "s3://my-bucket/my-checkpoint/"}} + # _resolve_hf_model_path strips the trailing slash when no + # hf_merged/hf checkpoint is found under the base path. + assert kw["modelDataSource"] == {"s3DataSource": {"s3Uri": "s3://my-bucket/my-checkpoint"}} def test_s3_uri_string_invalid_raises(self): """Non-S3 string as model raises ValueError.""" @@ -1240,7 +1242,7 @@ def test_training_job_manifest_json(self): b.boto_session = session with patch(f"{MODULE}.TrainingJob", type(mock_job)): - result = b._resolve_model_source_id() + result = b._resolve_nova_model_source_id() assert result == "s3://bucket/ckpt/step_100" @@ -1286,7 +1288,7 @@ def get_object_side_effect(Bucket, Key): b.boto_session = session with patch(f"{MODULE}.TrainingJob", type(mock_job)): - result = b._resolve_model_source_id() + result = b._resolve_nova_model_source_id() assert result == "s3://bucket/ckpt/step_50" @@ -1301,7 +1303,7 @@ def test_model_package_arn_for_rmp(self): with patch(f"{MODULE}.TrainingJob", _SentinelA), \ patch(f"{MODULE}.ModelTrainer", _SentinelB), \ patch(f"{MODULE}.BaseTrainer", _SentinelC): - result = b._resolve_model_source_id() + result = b._resolve_nova_model_source_id() assert result == "arn:aws:sagemaker:us-west-2:123456789012:model-package/my-pkg" @@ -1315,7 +1317,7 @@ def test_s3_model_artifacts_direct(self): with patch(f"{MODULE}.TrainingJob", _SentinelA), \ patch(f"{MODULE}.ModelTrainer", _SentinelB), \ patch(f"{MODULE}.BaseTrainer", _SentinelC): - result = b._resolve_model_source_id() + result = b._resolve_nova_model_source_id() assert result == "s3://my-bucket/checkpoints/" @@ -1326,7 +1328,7 @@ def test_returns_none_when_no_source(self): b.model_package = None b.s3_model_artifacts = None - result = b._resolve_model_source_id() + result = b._resolve_nova_model_source_id() assert result is None @@ -1454,3 +1456,159 @@ def test_deploy_without_target_still_applies_source_tag(self): } assert source_tag in kw["modelTags"] assert result["customModelDeploymentArn"] == "arn:dep" + + +class TestOSSModelReuseDeploy: + """Tests for OSS imported model reuse in deploy().""" + + @pytest.fixture(autouse=True) + def _stub_role_resolver(self): + with patch( + f"{MODULE}.resolve_and_validate_role", + side_effect=lambda provided_role, **kwargs: provided_role or "auto-role", + ): + yield + + def test_oss_reuse_existing_imported_model(self): + """When reuse_resources=True and a completed imported model exists, skip import.""" + b = BedrockModelBuilder(model="s3://bucket/artifacts/") + b._bedrock_client = Mock() + b._bedrock_client.get_imported_model.return_value = { + "modelArn": "arn:aws:bedrock:us-west-2:123:imported-model/existing", + "modelName": "existing", + } + + with patch( + f"{MODULE}.find_existing_imported_model", + return_value="arn:aws:bedrock:us-west-2:123:imported-model/existing", + ), patch(f"{MODULE}.find_existing_model_import_job") as mock_find_job: + result = b.deploy( + job_name="j", imported_model_name="m", role_arn="r", reuse_resources=True + ) + + b._bedrock_client.create_model_import_job.assert_not_called() + # Completed model reuse resolves via get_imported_model, not the job API. + b._bedrock_client.get_imported_model.assert_called_once_with( + modelIdentifier="arn:aws:bedrock:us-west-2:123:imported-model/existing" + ) + mock_find_job.assert_not_called() + assert result["modelArn"] == "arn:aws:bedrock:us-west-2:123:imported-model/existing" + assert b._imported_model_id == "existing" + + def test_oss_reuse_existing_in_progress_job(self): + """When no completed model but an in-progress job matches, wait on it.""" + b = BedrockModelBuilder(model="s3://bucket/artifacts/") + b._bedrock_client = Mock() + job_arn = "arn:aws:bedrock:us-west-2:123:model-import-job/abcd1234wxyz" + b._bedrock_client.get_model_import_job.return_value = { + "status": "Completed", + "importedModelName": "reused-model", + } + + with patch(f"{MODULE}.find_existing_imported_model", return_value=None), \ + patch(f"{MODULE}.find_existing_model_import_job", return_value=job_arn), \ + patch(f"{MODULE}.time.sleep"): + result = b.deploy( + job_name="j", imported_model_name="m", role_arn="r", reuse_resources=True + ) + + b._bedrock_client.create_model_import_job.assert_not_called() + b._bedrock_client.get_model_import_job.assert_called_with(jobIdentifier=job_arn) + assert result["status"] == "Completed" + assert b._imported_model_id == "reused-model" + + def test_oss_reuse_not_found_creates_new_import(self): + """When reuse_resources=True but nothing exists, create a new import.""" + b = BedrockModelBuilder(model="s3://bucket/artifacts/") + b._bedrock_client = Mock() + b._bedrock_client.create_model_import_job.return_value = {"jobArn": "arn:job"} + b._bedrock_client.get_model_import_job.return_value = { + "status": "Completed", + "importedModelName": "new-model", + } + + with patch(f"{MODULE}.find_existing_imported_model", return_value=None), \ + patch(f"{MODULE}.find_existing_model_import_job", return_value=None), \ + patch(f"{MODULE}.time.sleep"): + result = b.deploy( + job_name="j", imported_model_name="m", role_arn="r", reuse_resources=True + ) + + b._bedrock_client.create_model_import_job.assert_called_once() + assert result["status"] == "Completed" + + def test_oss_reuse_false_skips_lookup_but_tags(self): + """Default reuse_resources=False: no lookup, but source tag is applied.""" + b = BedrockModelBuilder(model="s3://bucket/artifacts/") + b._bedrock_client = Mock() + b._bedrock_client.create_model_import_job.return_value = {"jobArn": "arn:job"} + b._bedrock_client.get_model_import_job.return_value = { + "status": "Completed", + "importedModelName": "m", + } + + with patch(f"{MODULE}.find_existing_imported_model") as mock_find, \ + patch(f"{MODULE}.time.sleep"): + b.deploy(job_name="j", imported_model_name="m", role_arn="r") + + mock_find.assert_not_called() + # Source tag should still be applied to new imports + kw = b._bedrock_client.create_model_import_job.call_args[1] + source_tag = { + "key": "sagemaker.amazonaws.com/model-source", + "value": "s3://bucket/artifacts/", + } + assert source_tag in kw["importedModelTags"] + + def test_oss_reuse_uses_model_package_arn_as_source(self): + """Source ID prefers model package ARN over S3 artifacts.""" + c = _make_container(s3_uri="s3://b/m/") + b = _builder() + b.model_package = _make_model_package(c) + b.model_package.model_package_arn = "arn:aws:sagemaker:us-west-2:123:model-package/mp/1" + b.s3_model_artifacts = "s3://b/m/" + b._bedrock_client = Mock() + b._bedrock_client.create_model_import_job.return_value = {"jobArn": "arn:job"} + b._bedrock_client.get_model_import_job.return_value = { + "status": "Completed", + "importedModelName": "m", + } + + with patch(f"{MODULE}.find_existing_imported_model") as mock_find, \ + patch(f"{MODULE}.find_existing_model_import_job", return_value=None), \ + patch(f"{MODULE}.time.sleep"): + mock_find.return_value = None + b.deploy(job_name="j", imported_model_name="m", role_arn="r", reuse_resources=True) + + # Should pass model package ARN as source_id, not s3 artifacts + mock_find.assert_called_once() + call_source_id = mock_find.call_args[0][1] + assert call_source_id == "arn:aws:sagemaker:us-west-2:123:model-package/mp/1" + + def test_oss_reuse_preserves_user_tags(self): + """User-provided imported_model_tags are preserved alongside the source tag.""" + b = BedrockModelBuilder(model="s3://bucket/path/") + b._bedrock_client = Mock() + b._bedrock_client.create_model_import_job.return_value = {"jobArn": "arn:job"} + b._bedrock_client.get_model_import_job.return_value = { + "status": "Completed", + "importedModelName": "m", + } + user_tag = {"key": "team", "value": "ml-platform"} + + with patch(f"{MODULE}.find_existing_imported_model", return_value=None), \ + patch(f"{MODULE}.find_existing_model_import_job", return_value=None), \ + patch(f"{MODULE}.time.sleep"): + b.deploy( + job_name="j", + imported_model_name="m", + role_arn="r", + imported_model_tags=[user_tag], + reuse_resources=True, + ) + + kw = b._bedrock_client.create_model_import_job.call_args[1] + tags = kw["importedModelTags"] + assert user_tag in tags + source_tag = {"key": "sagemaker.amazonaws.com/model-source", "value": "s3://bucket/path/"} + assert source_tag in tags diff --git a/sagemaker-serve/tests/unit/test_model_reuse.py b/sagemaker-serve/tests/unit/test_model_reuse.py index 7cc6bffcab..37dec16acc 100644 --- a/sagemaker-serve/tests/unit/test_model_reuse.py +++ b/sagemaker-serve/tests/unit/test_model_reuse.py @@ -23,6 +23,8 @@ normalize_tag_value, find_active_bedrock_deployment_for_model, find_existing_bedrock_model, + find_existing_imported_model, + find_existing_model_import_job, find_existing_sagemaker_endpoint, build_source_tag, check_bedrock_model_status, @@ -358,3 +360,106 @@ def test_check_sagemaker_endpoint_status_raises_on_failure(): with pytest.raises(Exception, match="Endpoint not found"): check_sagemaker_endpoint_status(sm_client, ENDPOINT_ARN) + + +# ── find_existing_imported_model tests ────────────────────────────────────── + +IMPORTED_MODEL_ARN = "arn:aws:bedrock:us-east-1:123456789012:imported-model/my-model" + + +def test_find_existing_imported_model_returns_arn_when_found(bedrock_client): + bedrock_client.list_imported_models.return_value = { + "modelSummaries": [{"modelArn": IMPORTED_MODEL_ARN}] + } + bedrock_client.list_tags_for_resource.return_value = { + "tags": [{"key": MODEL_SOURCE_TAG_KEY, "value": "s3://bucket/path/"}] + } + + result = find_existing_imported_model(bedrock_client, "s3://bucket/path/") + + assert result == IMPORTED_MODEL_ARN + + +def test_find_existing_imported_model_returns_none_when_no_match(bedrock_client): + bedrock_client.list_imported_models.return_value = { + "modelSummaries": [{"modelArn": IMPORTED_MODEL_ARN}] + } + bedrock_client.list_tags_for_resource.return_value = { + "tags": [{"key": MODEL_SOURCE_TAG_KEY, "value": "s3://other/path/"}] + } + + result = find_existing_imported_model(bedrock_client, "s3://bucket/path/") + + assert result is None + + +def test_find_existing_imported_model_ignores_import_jobs(bedrock_client): + """find_existing_imported_model only inspects completed models, not jobs.""" + bedrock_client.list_imported_models.return_value = {"modelSummaries": []} + bedrock_client.list_tags_for_resource.return_value = { + "tags": [{"key": MODEL_SOURCE_TAG_KEY, "value": "s3://bucket/path/"}] + } + + result = find_existing_imported_model(bedrock_client, "s3://bucket/path/") + + assert result is None + bedrock_client.list_model_import_jobs.assert_not_called() + + +def test_find_existing_imported_model_access_denied_raises_permission_error(bedrock_client): + """AccessDeniedException is surfaced as PermissionError.""" + bedrock_client.list_imported_models.return_value = { + "modelSummaries": [{"modelArn": IMPORTED_MODEL_ARN}] + } + bedrock_client.list_tags_for_resource.side_effect = _access_denied_error( + "ListTagsForResource", "bedrock:ListTagsForResource" + ) + + with pytest.raises(PermissionError, match="bedrock:ListTagsForResource"): + find_existing_imported_model(bedrock_client, "s3://bucket/path/") + + +# ── find_existing_model_import_job tests ───────────────────────────────────── + +IMPORT_JOB_ARN = "arn:aws:bedrock:us-east-1:123456789012:model-import-job/abcd1234wxyz" + + +def test_find_existing_model_import_job_returns_arn_when_found(bedrock_client): + """Returns the in-progress import job ARN when its source tag matches.""" + bedrock_client.list_model_import_jobs.return_value = { + "modelImportJobSummaries": [{"jobArn": IMPORT_JOB_ARN}] + } + bedrock_client.list_tags_for_resource.return_value = { + "tags": [{"key": MODEL_SOURCE_TAG_KEY, "value": "s3://bucket/path/"}] + } + + result = find_existing_model_import_job(bedrock_client, "s3://bucket/path/") + + assert result == IMPORT_JOB_ARN + + +def test_find_existing_model_import_job_returns_none_when_no_match(bedrock_client): + """Returns None when no in-progress job carries a matching source tag.""" + bedrock_client.list_model_import_jobs.return_value = { + "modelImportJobSummaries": [{"jobArn": IMPORT_JOB_ARN}] + } + bedrock_client.list_tags_for_resource.return_value = { + "tags": [{"key": MODEL_SOURCE_TAG_KEY, "value": "s3://other/path/"}] + } + + result = find_existing_model_import_job(bedrock_client, "s3://bucket/path/") + + assert result is None + + +def test_find_existing_model_import_job_access_denied_raises_permission_error(bedrock_client): + """AccessDeniedException is surfaced as PermissionError.""" + bedrock_client.list_model_import_jobs.return_value = { + "modelImportJobSummaries": [{"jobArn": IMPORT_JOB_ARN}] + } + bedrock_client.list_tags_for_resource.side_effect = _access_denied_error( + "ListTagsForResource", "bedrock:ListTagsForResource" + ) + + with pytest.raises(PermissionError, match="bedrock:ListTagsForResource"): + find_existing_model_import_job(bedrock_client, "s3://bucket/path/")