catalin/moto
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Add support for setting tags on ecs task definitions

Browse source 
This also implements the ecs.list_tags_for_resources, although the
resources it checks for are currently only the task definitions
This commit is contained in:
Michael van Tellingen 2019-07-26 15:11:25 +02:00
commit 64e2a74e8c
4 changed files with 99 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

32
moto/ecs/models.py
View file

@ -1,4 +1,5 @@
from __future__ import unicode_literals
import re
import uuid
from datetime import datetime
from random import random, randint
@ -10,7 +11,10 @@ from moto.core import BaseBackend, BaseModel
from moto.ec2 import ec2_backends
from copy import copy
from .exceptions import ServiceNotFoundException
from .exceptions import (
ServiceNotFoundException,
TaskDefinitionNotFoundException
)
class BaseObject(BaseModel):
@ -103,12 +107,13 @@ class Cluster(BaseObject):
class TaskDefinition(BaseObject):
def __init__(self, family, revision, container_definitions, volumes=None):
def __init__(self, family, revision, container_definitions, volumes=None, tags=None):
self.family = family
self.revision = revision
self.arn = 'arn:aws:ecs:us-east-1:012345678910:task-definition/{0}:{1}'.format(
family, revision)
self.container_definitions = container_definitions
self.tags = tags if tags is not None else []
if volumes is None:
self.volumes = []
else:
@ -119,6 +124,7 @@ class TaskDefinition(BaseObject):
response_object = self.gen_response_object()
response_object['taskDefinitionArn'] = response_object['arn']
del response_object['arn']
del response_object['tags']
return response_object
@property
@ -464,7 +470,7 @@ class EC2ContainerServiceBackend(BaseBackend):
else:
raise Exception("{0} is not a cluster".format(cluster_name))
def register_task_definition(self, family, container_definitions, volumes):
def register_task_definition(self, family, container_definitions, volumes, tags=None):
if family in self.task_definitions:
last_id = self._get_last_task_definition_revision_id(family)
revision = (last_id or 0) + 1
@ -472,7 +478,7 @@ class EC2ContainerServiceBackend(BaseBackend):
self.task_definitions[family] = {}
revision = 1
task_definition = TaskDefinition(
family, revision, container_definitions, volumes)
family, revision, container_definitions, volumes, tags)
self.task_definitions[family][revision] = task_definition
return task_definition
@ -951,6 +957,24 @@ class EC2ContainerServiceBackend(BaseBackend):
yield task_fam
def list_tags_for_resource(self, resource_arn):
"""Currently only implemented for task definitions"""
match = re.match(
"^arn:aws:ecs:(?P<region>[^:]+):(?P<account_id>[^:]+):(?P<service>[^:]+)/(?P<id>.*)$",
resource_arn)
if not match:
raise JsonRESTError('InvalidParameterException', 'The ARN provided is invalid.')
service = match.group("service")
if service == "task-definition":
for task_definition in self.task_definitions.values():
for revision in task_definition.values():
if revision.arn == resource_arn:
return revision.tags
else:
raise TaskDefinitionNotFoundException()
raise NotImplementedError()
def _get_last_task_definition_revision_id(self, family):
definitions = self.task_definitions.get(family, {})
if definitions: