Merge pull request #2694 from brady-gsa/events-tagging

adds tagging support for cloudwatch events service

tests/test_events/test_events.py

@@ -1,12 +1,15 @@
-import random
-import boto3
 import json
-import sure  # noqa
+import random
+import unittest
 
-from moto.events import mock_events
+import boto3
 from botocore.exceptions import ClientError
 from nose.tools import assert_raises
+
 from moto.core import ACCOUNT_ID
+from moto.core.exceptions import JsonRESTError
+from moto.events import mock_events
+from moto.events.models import EventsBackend
 
 RULES = [
     {"Name": "test1", "ScheduleExpression": "rate(5 minutes)"},
@@ -136,14 +139,6 @@ def test_list_rule_names_by_target():
         assert rule in test_2_target["Rules"]
 
 
-@mock_events
-def test_list_rules():
-    client = generate_environment()
-
-    rules = client.list_rules()
-    assert len(rules["Rules"]) == len(RULES)
-
-
 @mock_events
 def test_delete_rule():
     client = generate_environment()
@@ -461,3 +456,50 @@ def test_delete_event_bus_errors():
     client.delete_event_bus.when.called_with(Name="default").should.throw(
         ClientError, "Cannot delete event bus default."
     )
+
+
+@mock_events
+def test_rule_tagging_happy():
+    client = generate_environment()
+    rule_name = get_random_rule()["Name"]
+    rule_arn = client.describe_rule(Name=rule_name).get("Arn")
+
+    tags = [{"Key": "key1", "Value": "value1"}, {"Key": "key2", "Value": "value2"}]
+    client.tag_resource(ResourceARN=rule_arn, Tags=tags)
+
+    actual = client.list_tags_for_resource(ResourceARN=rule_arn).get("Tags")
+    tc = unittest.TestCase("__init__")
+    expected = [{"Value": "value1", "Key": "key1"}, {"Value": "value2", "Key": "key2"}]
+    tc.assertTrue(
+        (expected[0] == actual[0] and expected[1] == actual[1])
+        or (expected[1] == actual[0] and expected[0] == actual[1])
+    )
+
+    client.untag_resource(ResourceARN=rule_arn, TagKeys=["key1"])
+
+    actual = client.list_tags_for_resource(ResourceARN=rule_arn).get("Tags")
+    expected = [{"Key": "key2", "Value": "value2"}]
+    assert expected == actual
+
+
+@mock_events
+def test_rule_tagging_sad():
+    back_end = EventsBackend("us-west-2")
+
+    try:
+        back_end.tag_resource("unknown", [])
+        raise "tag_resource should fail if ResourceARN is not known"
+    except JsonRESTError:
+        pass
+
+    try:
+        back_end.untag_resource("unknown", [])
+        raise "untag_resource should fail if ResourceARN is not known"
+    except JsonRESTError:
+        pass
+
+    try:
+        back_end.list_tags_for_resource("unknown")
+        raise "list_tags_for_resource should fail if ResourceARN is not known"
+    except JsonRESTError:
+        pass

tests/test_utilities/test_tagging_service.py

@@ -0,0 +1,79 @@
+import sure
+
+from moto.utilities.tagging_service import TaggingService
+
+
+def test_list_empty():
+    svc = TaggingService()
+    result = svc.list_tags_for_resource("test")
+
+    {"Tags": []}.should.be.equal(result)
+
+
+def test_create_tag():
+    svc = TaggingService("TheTags", "TagKey", "TagValue")
+    tags = [{"TagKey": "key_key", "TagValue": "value_value"}]
+    svc.tag_resource("arn", tags)
+    actual = svc.list_tags_for_resource("arn")
+    expected = {"TheTags": [{"TagKey": "key_key", "TagValue": "value_value"}]}
+
+    expected.should.be.equal(actual)
+
+
+def test_create_tag_without_value():
+    svc = TaggingService()
+    tags = [{"Key": "key_key"}]
+    svc.tag_resource("arn", tags)
+    actual = svc.list_tags_for_resource("arn")
+    expected = {"Tags": [{"Key": "key_key", "Value": None}]}
+
+    expected.should.be.equal(actual)
+
+
+def test_delete_tag_using_names():
+    svc = TaggingService()
+    tags = [{"Key": "key_key", "Value": "value_value"}]
+    svc.tag_resource("arn", tags)
+    svc.untag_resource_using_names("arn", ["key_key"])
+    result = svc.list_tags_for_resource("arn")
+
+    {"Tags": []}.should.be.equal(result)
+
+
+def test_delete_all_tags_for_resource():
+    svc = TaggingService()
+    tags = [{"Key": "key_key", "Value": "value_value"}]
+    tags2 = [{"Key": "key_key2", "Value": "value_value2"}]
+    svc.tag_resource("arn", tags)
+    svc.tag_resource("arn", tags2)
+    svc.delete_all_tags_for_resource("arn")
+    result = svc.list_tags_for_resource("arn")
+
+    {"Tags": []}.should.be.equal(result)
+
+
+def test_list_empty_delete():
+    svc = TaggingService()
+    svc.untag_resource_using_names("arn", ["key_key"])
+    result = svc.list_tags_for_resource("arn")
+
+    {"Tags": []}.should.be.equal(result)
+
+
+def test_delete_tag_using_tags():
+    svc = TaggingService()
+    tags = [{"Key": "key_key", "Value": "value_value"}]
+    svc.tag_resource("arn", tags)
+    svc.untag_resource_using_tags("arn", tags)
+    result = svc.list_tags_for_resource("arn")
+
+    {"Tags": []}.should.be.equal(result)
+
+
+def test_extract_tag_names():
+    svc = TaggingService()
+    tags = [{"Key": "key1", "Value": "value1"}, {"Key": "key2", "Value": "value2"}]
+    actual = svc.extract_tag_names(tags)
+    expected = ["key1", "key2"]
+
+    expected.should.be.equal(actual)