Merge pull request #2435 from waynemetcalfe/issue-2432

Issue 2432

IMPLEMENTATION_COVERAGE.md

@@ -4080,7 +4080,7 @@
 - [ ] get_log_group_fields
 - [ ] get_log_record
 - [ ] get_query_results
-- [ ] list_tags_log_group
+- [X] list_tags_log_group
 - [ ] put_destination
 - [ ] put_destination_policy
 - [X] put_log_events
@@ -4090,9 +4090,9 @@
 - [ ] put_subscription_filter
 - [ ] start_query
 - [ ] stop_query
-- [ ] tag_log_group
+- [X] tag_log_group
 - [ ] test_metric_filter
-- [ ] untag_log_group
+- [X] untag_log_group
 
 ## machinelearning
 0% implemented

moto/logs/models.py

@@ -231,6 +231,19 @@ class LogGroup:
     def set_retention_policy(self, retention_in_days):
         self.retentionInDays = retention_in_days
 
+    def list_tags(self):
+        return self.tags if self.tags else {}
+
+    def tag(self, tags):
+        if self.tags:
+            self.tags.update(tags)
+        else:
+            self.tags = tags
+
+    def untag(self, tags_to_remove):
+        if self.tags:
+            self.tags = {k: v for (k, v) in self.tags.items() if k not in tags_to_remove}
+
 
 class LogsBackend(BaseBackend):
     def __init__(self, region_name):
@@ -322,5 +335,23 @@ class LogsBackend(BaseBackend):
         log_group = self.groups[log_group_name]
         return log_group.set_retention_policy(None)
 
+    def list_tags_log_group(self, log_group_name):
+        if log_group_name not in self.groups:
+            raise ResourceNotFoundException()
+        log_group = self.groups[log_group_name]
+        return log_group.list_tags()
+
+    def tag_log_group(self, log_group_name, tags):
+        if log_group_name not in self.groups:
+            raise ResourceNotFoundException()
+        log_group = self.groups[log_group_name]
+        log_group.tag(tags)
+
+    def untag_log_group(self, log_group_name, tags):
+        if log_group_name not in self.groups:
+            raise ResourceNotFoundException()
+        log_group = self.groups[log_group_name]
+        log_group.untag(tags)
+
 
 logs_backends = {region.name: LogsBackend(region.name) for region in boto.logs.regions()}

moto/logs/responses.py

@@ -134,3 +134,22 @@ class LogsResponse(BaseResponse):
         log_group_name = self._get_param('logGroupName')
         self.logs_backend.delete_retention_policy(log_group_name)
         return ''
+
+    def list_tags_log_group(self):
+        log_group_name = self._get_param('logGroupName')
+        tags = self.logs_backend.list_tags_log_group(log_group_name)
+        return json.dumps({
+            'tags': tags
+        })
+
+    def tag_log_group(self):
+        log_group_name = self._get_param('logGroupName')
+        tags = self._get_param('tags')
+        self.logs_backend.tag_log_group(log_group_name, tags)
+        return ''
+
+    def untag_log_group(self):
+        log_group_name = self._get_param('logGroupName')
+        tags = self._get_param('tags')
+        self.logs_backend.untag_log_group(log_group_name, tags)
+        return ''

tests/test_logs/test_logs.py

@@ -225,3 +225,65 @@ def test_get_log_events():
     for i in range(10):
         resp['events'][i]['timestamp'].should.equal(i)
         resp['events'][i]['message'].should.equal(str(i))
+
+
+@mock_logs
+def test_list_tags_log_group():
+    conn = boto3.client('logs', 'us-west-2')
+    log_group_name = 'dummy'
+    tags = {'tag_key_1': 'tag_value_1', 'tag_key_2': 'tag_value_2'}
+
+    response = conn.create_log_group(logGroupName=log_group_name)
+    response = conn.list_tags_log_group(logGroupName=log_group_name)
+    assert response['tags'] == {}
+
+    response = conn.delete_log_group(logGroupName=log_group_name)
+    response = conn.create_log_group(logGroupName=log_group_name, tags=tags)
+    response = conn.list_tags_log_group(logGroupName=log_group_name)
+    assert response['tags'] == tags
+
+    response = conn.delete_log_group(logGroupName=log_group_name)
+
+
+@mock_logs
+def test_tag_log_group():
+    conn = boto3.client('logs', 'us-west-2')
+    log_group_name = 'dummy'
+    tags = {'tag_key_1': 'tag_value_1'}
+    response = conn.create_log_group(logGroupName=log_group_name)
+
+    response = conn.tag_log_group(logGroupName=log_group_name, tags=tags)
+    response = conn.list_tags_log_group(logGroupName=log_group_name)
+    assert response['tags'] == tags
+
+    tags_with_added_value = {'tag_key_1': 'tag_value_1', 'tag_key_2': 'tag_value_2'}
+    response = conn.tag_log_group(logGroupName=log_group_name, tags={'tag_key_2': 'tag_value_2'})
+    response = conn.list_tags_log_group(logGroupName=log_group_name)
+    assert response['tags'] == tags_with_added_value
+
+    tags_with_updated_value = {'tag_key_1': 'tag_value_XX', 'tag_key_2': 'tag_value_2'}
+    response = conn.tag_log_group(logGroupName=log_group_name, tags={'tag_key_1': 'tag_value_XX'})
+    response = conn.list_tags_log_group(logGroupName=log_group_name)
+    assert response['tags'] == tags_with_updated_value
+
+    response = conn.delete_log_group(logGroupName=log_group_name)
+
+
+@mock_logs
+def test_untag_log_group():
+    conn = boto3.client('logs', 'us-west-2')
+    log_group_name = 'dummy'
+    response = conn.create_log_group(logGroupName=log_group_name)
+
+    tags = {'tag_key_1': 'tag_value_1', 'tag_key_2': 'tag_value_2'}
+    response = conn.tag_log_group(logGroupName=log_group_name, tags=tags)
+    response = conn.list_tags_log_group(logGroupName=log_group_name)
+    assert response['tags'] == tags
+
+    tags_to_remove = ['tag_key_1']
+    remaining_tags = {'tag_key_2': 'tag_value_2'}
+    response = conn.untag_log_group(logGroupName=log_group_name, tags=tags_to_remove)
+    response = conn.list_tags_log_group(logGroupName=log_group_name)
+    assert response['tags'] == remaining_tags
+
+    response = conn.delete_log_group(logGroupName=log_group_name)