67 lines
1.9 KiB
Python
67 lines
1.9 KiB
Python
import json
|
|
|
|
from moto.core.responses import BaseResponse
|
|
|
|
from .models import CodePipelineBackend, codepipeline_backends
|
|
|
|
|
|
class CodePipelineResponse(BaseResponse):
|
|
def __init__(self) -> None:
|
|
super().__init__(service_name="codepipeline")
|
|
|
|
@property
|
|
def codepipeline_backend(self) -> CodePipelineBackend:
|
|
return codepipeline_backends[self.current_account][self.region]
|
|
|
|
def create_pipeline(self) -> str:
|
|
pipeline, tags = self.codepipeline_backend.create_pipeline(
|
|
self._get_param("pipeline"), self._get_param("tags")
|
|
)
|
|
|
|
return json.dumps({"pipeline": pipeline, "tags": tags})
|
|
|
|
def get_pipeline(self) -> str:
|
|
pipeline, metadata = self.codepipeline_backend.get_pipeline(
|
|
self._get_param("name")
|
|
)
|
|
|
|
return json.dumps({"pipeline": pipeline, "metadata": metadata})
|
|
|
|
def update_pipeline(self) -> str:
|
|
pipeline = self.codepipeline_backend.update_pipeline(
|
|
self._get_param("pipeline")
|
|
)
|
|
|
|
return json.dumps({"pipeline": pipeline})
|
|
|
|
def list_pipelines(self) -> str:
|
|
pipelines = self.codepipeline_backend.list_pipelines()
|
|
|
|
return json.dumps({"pipelines": pipelines})
|
|
|
|
def delete_pipeline(self) -> str:
|
|
self.codepipeline_backend.delete_pipeline(self._get_param("name"))
|
|
|
|
return ""
|
|
|
|
def list_tags_for_resource(self) -> str:
|
|
tags = self.codepipeline_backend.list_tags_for_resource(
|
|
self._get_param("resourceArn")
|
|
)
|
|
|
|
return json.dumps({"tags": tags})
|
|
|
|
def tag_resource(self) -> str:
|
|
self.codepipeline_backend.tag_resource(
|
|
self._get_param("resourceArn"), self._get_param("tags")
|
|
)
|
|
|
|
return ""
|
|
|
|
def untag_resource(self) -> str:
|
|
self.codepipeline_backend.untag_resource(
|
|
self._get_param("resourceArn"), self._get_param("tagKeys")
|
|
)
|
|
|
|
return ""
|