Add sagemaker mock call: delete_pipeline (#5780)

This commit is contained in:
sist 2022-12-17 11:38:23 +01:00 committed by GitHub
parent 137f06b55e
commit f67abbe1f3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 48 additions and 1 deletions

View File

@ -108,7 +108,7 @@ sagemaker
- [ ] delete_monitoring_schedule
- [X] delete_notebook_instance
- [X] delete_notebook_instance_lifecycle_config
- [ ] delete_pipeline
- [X] delete_pipeline
- [ ] delete_project
- [ ] delete_studio_lifecycle_config
- [X] delete_tags

View File

@ -1774,6 +1774,19 @@ class SageMakerModelBackend(BaseBackend):
self.pipelines[pipeline_name] = pipeline
return pipeline
def delete_pipeline(
self,
pipeline_name,
):
try:
pipeline_arn = self.pipelines[pipeline_name].pipeline_arn
except KeyError:
raise ValidationError(
message=f"Could not find pipeline with name {pipeline_name}."
)
del self.pipelines[pipeline_name]
return pipeline_arn
def list_pipelines(
self,
pipeline_name_prefix,

View File

@ -485,6 +485,14 @@ class SageMakerResponse(BaseResponse):
return 200, {}, json.dumps(response)
@amzn_request_id
def delete_pipeline(self):
pipeline_arn = self.sagemaker_backend.delete_pipeline(
pipeline_name=self._get_param("PipelineName"),
)
response = {"PipelineArn": pipeline_arn}
return 200, {}, json.dumps(response)
@amzn_request_id
def list_pipelines(self):
max_results_range = range(1, 101)

View File

@ -178,3 +178,29 @@ def test_list_pipelines_created_before(sagemaker_client):
def test_list_pipelines_invalid_values(sagemaker_client, list_pipelines_kwargs):
with pytest.raises(botocore.exceptions.ClientError):
_ = sagemaker_client.list_pipelines(**list_pipelines_kwargs)
def test_delete_pipeline_exists(sagemaker_client):
fake_pipeline_names = ["APipelineName", "BPipelineName", "CPipelineName"]
_ = create_sagemaker_pipelines(sagemaker_client, fake_pipeline_names, 0.0)
pipeline_name_delete, pipeline_names_remain = (
fake_pipeline_names[0],
fake_pipeline_names[1:],
)
response = sagemaker_client.delete_pipeline(PipelineName=pipeline_name_delete)
assert response["PipelineArn"].endswith(pipeline_name_delete)
response = sagemaker_client.list_pipelines(PipelineNamePrefix=pipeline_name_delete)
assert response["PipelineSummaries"].should.be.empty
response = sagemaker_client.list_pipelines()
pipeline_names_exist = [
pipeline["PipelineName"] for pipeline in response["PipelineSummaries"]
]
assert pipeline_names_remain == pipeline_names_exist
def test_delete_pipeline_not_exists(sagemaker_client):
with pytest.raises(botocore.exceptions.ClientError):
_ = sagemaker_client.delete_pipeline(PipelineName="some-pipeline-name")