moto/moto/ecr/responses.py

from __future__ import unicode_literals
import json
from base64 import b64encode
from datetime import datetime
import time

from moto.core.responses import BaseResponse
from .models import ecr_backends, DEFAULT_REGISTRY_ID


class ECRResponse(BaseResponse):
    @property
    def ecr_backend(self):
        return ecr_backends[self.region]

    @property
    def request_params(self):
        try:
            return json.loads(self.body)
        except ValueError:
            return {}

    def _get_param(self, param):
        return self.request_params.get(param, None)

    def create_repository(self):
        repository_name = self._get_param('repositoryName')
        if repository_name is None:
            repository_name = 'default'
        repository = self.ecr_backend.create_repository(repository_name)
        return json.dumps({
            'repository': repository.response_object
        })

    def describe_repositories(self):
        describe_repositories_name = self._get_param('repositoryNames')
        registry_id = self._get_param('registryId')

        repositories = self.ecr_backend.describe_repositories(
            repository_names=describe_repositories_name, registry_id=registry_id)
        return json.dumps({
            'repositories': repositories,
            'failures': []
        })

    def delete_repository(self):
        repository_str = self._get_param('repositoryName')
        registry_id = self._get_param('registryId')
        repository = self.ecr_backend.delete_repository(repository_str, registry_id)
        return json.dumps({
            'repository': repository.response_object
        })

    def put_image(self):
        repository_str = self._get_param('repositoryName')
        image_manifest = self._get_param('imageManifest')
        image_tag = self._get_param('imageTag')
        image = self.ecr_backend.put_image(repository_str, image_manifest, image_tag)

        return json.dumps({
            'image': image.response_object
        })

    def list_images(self):
        repository_str = self._get_param('repositoryName')
        registry_id = self._get_param('registryId')
        images = self.ecr_backend.list_images(repository_str, registry_id)
        return json.dumps({
            'imageIds': [image.response_list_object for image in images],
        })

    def describe_images(self):
        repository_str = self._get_param('repositoryName')
        registry_id = self._get_param('registryId')
        image_ids = self._get_param('imageIds')
        images = self.ecr_backend.describe_images(repository_str, registry_id, image_ids)
        return json.dumps({
            'imageDetails': [image.response_describe_object for image in images],
        })

    def batch_check_layer_availability(self):
        if self.is_not_dryrun('BatchCheckLayerAvailability'):
            raise NotImplementedError(
                'ECR.batch_check_layer_availability is not yet implemented')

    def batch_delete_image(self):
        repository_str = self._get_param('repositoryName')
        registry_id = self._get_param('registryId')
        image_ids = self._get_param('imageIds')

        response = self.ecr_backend.batch_delete_image(repository_str, registry_id, image_ids)
        return json.dumps(response)

    def batch_get_image(self):
        repository_str = self._get_param('repositoryName')
        registry_id = self._get_param('registryId')
        image_ids = self._get_param('imageIds')
        accepted_media_types = self._get_param('acceptedMediaTypes')

        response = self.ecr_backend.batch_get_image(repository_str, registry_id, image_ids, accepted_media_types)
        return json.dumps(response)

    def can_paginate(self):
        if self.is_not_dryrun('CanPaginate'):
            raise NotImplementedError(
                'ECR.can_paginate is not yet implemented')

    def complete_layer_upload(self):
        if self.is_not_dryrun('CompleteLayerUpload'):
            raise NotImplementedError(
                'ECR.complete_layer_upload is not yet implemented')

    def delete_repository_policy(self):
        if self.is_not_dryrun('DeleteRepositoryPolicy'):
            raise NotImplementedError(
                'ECR.delete_repository_policy is not yet implemented')

    def generate_presigned_url(self):
        if self.is_not_dryrun('GeneratePresignedUrl'):
            raise NotImplementedError(
                'ECR.generate_presigned_url is not yet implemented')

    def get_authorization_token(self):
        registry_ids = self._get_param('registryIds')
        if not registry_ids:
            registry_ids = [DEFAULT_REGISTRY_ID]
        auth_data = []
        for registry_id in registry_ids:
            password = '{}-auth-token'.format(registry_id)
            auth_token = b64encode("AWS:{}".format(password).encode('ascii')).decode()
            auth_data.append({
                'authorizationToken': auth_token,
                'expiresAt': time.mktime(datetime(2015, 1, 1).timetuple()),
                'proxyEndpoint': 'https://{}.dkr.ecr.{}.amazonaws.com'.format(registry_id, self.region)
            })
        return json.dumps({'authorizationData': auth_data})

    def get_download_url_for_layer(self):
        if self.is_not_dryrun('GetDownloadUrlForLayer'):
            raise NotImplementedError(
                'ECR.get_download_url_for_layer is not yet implemented')

    def get_paginator(self):
        if self.is_not_dryrun('GetPaginator'):
            raise NotImplementedError(
                'ECR.get_paginator is not yet implemented')

    def get_repository_policy(self):
        if self.is_not_dryrun('GetRepositoryPolicy'):
            raise NotImplementedError(
                'ECR.get_repository_policy is not yet implemented')

    def get_waiter(self):
        if self.is_not_dryrun('GetWaiter'):
            raise NotImplementedError(
                'ECR.get_waiter is not yet implemented')

    def initiate_layer_upload(self):
        if self.is_not_dryrun('InitiateLayerUpload'):
            raise NotImplementedError(
                'ECR.initiate_layer_upload is not yet implemented')

    def set_repository_policy(self):
        if self.is_not_dryrun('SetRepositoryPolicy'):
            raise NotImplementedError(
                'ECR.set_repository_policy is not yet implemented')

    def upload_layer_part(self):
        if self.is_not_dryrun('UploadLayerPart'):
            raise NotImplementedError(
                'ECR.upload_layer_part is not yet implemented')