We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
mark-for-op and tag actions missing for resource=kinesis
The mark-for-op and tag actions are available for ec2, rds, and other resource types is not available for kinesis, but would desire them to be

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `tools/dev/ratchet.py`
Content:
```
1 #!/usr/bin/env python3.6
2 # -*- coding: utf-8 -*-
3 """Ratchet up successes under Python 3.6.
4 """
5 from __future__ import (
6     absolute_import, division, print_function, unicode_literals)
7 
8 import sys
9 import xml.etree.ElementTree as Etree
10 
11 
12 INVALID_TYPE = 'E   botocore.exceptions.ParamValidationError: ' \
13     'Parameter validation failed:     Invalid type for parameter '
14 
15 
16 class TestResults(object):
17 
18     def __init__(self, results_path):
19         self.results_path = results_path
20         self.failed = []
21         self.failed_aggregates = {}
22         self.stderr_output = []
23         self.passed = []
24         self._tree = None
25 
26     def parse(self):
27         if self._tree:
28             raise AssertionError("Already Parsed")
29         self._tree = Etree.parse(self.results_path)
30         for testcase in self._tree.findall('testcase'):
31             self.process_testcase(testcase)
32         return self
33 
34     def process_testcase(self, case):
35         key = self.case_key(case)
36 
37         # look at children but throw away stderr output
38         nonsuccess = [c for c in case if not c.tag == 'system-err']
39         n = len(nonsuccess)
40         if n > 1:
41             raise AssertionError("multiple results for %s: %s" %
42                 (key, nonsuccess))
43         elif n == 1:
44             result = nonsuccess.pop()
45             self.failed.append(key)
46             message = result.get('message')
47             if message.startswith(INVALID_TYPE):
48                 message = INVALID_TYPE
49             self.failed_aggregates.setdefault(message, []).append(key)
50         else:
51             self.passed.append(key)
52 
53     @staticmethod
54     def case_key(case):
55         return "%s.%s" % (case.get('classname'), case.get('name'))
56 
57     def report(self, details=False):
58         for k, v in sorted(
59                 self.failed_aggregates.items(),
60                 key = lambda i: len(i[1]),
61                 reverse=True):
62             print("# %s" % k)
63             for t in v:
64                 print(" - %s" % t)
65 
66 
67 def load_expected_failures(txt):
68     expected_failures = open(txt).read()
69     parsed = set()
70     for line in expected_failures.splitlines():
71         if not line or line.startswith('#'):
72             continue
73         parsed.add(line)
74     return parsed
75 
76 
77 def list_tests(tests):
78     for test in sorted(tests):
79         print(' ', test)
80 
81 
82 def update_expectation(txt_path, tests):
83     new = set(t + '\n' for t in tests)
84     open(txt_path, 'w+').writelines(sorted(new))
85 
86 
87 def main(xml_path, txt_path):
88     """Takes two paths, one to XML output from pytest, the other to a text file
89     listing expected successes. Walks the former looking for the latter.
90     """
91     results = TestResults(xml_path).parse()
92 
93     if txt_path == '-':
94         results.report()
95         return
96 
97     previous = load_expected_failures(txt_path)
98     current = set(results.failed)
99 
100     expected = previous - current
101     if expected:
102         print("Some tests expected to fail under Python 3.6 didn't:")
103         list_tests(expected)
104         update_expectation(txt_path, current)
105         print("Conveniently, they have been removed from {} for you. Perhaps "
106             "commit that?".format(txt_path))
107 
108     unexpected = current - previous
109     if unexpected:
110         print("Some tests not expected to fail under Python 3.6 did:")
111         list_tests(unexpected)
112 
113     if expected or unexpected:
114         print("Previously %d tests failed under Python 3.6, now %d did." %
115             (len(previous), len(current)))
116         return 1
117 
118     print('All and only tests expected to fail under Python 3.6 did.')
119     return 0
120 
121 
122 if __name__ == '__main__':
123     try:
124         xml_path, txt_path = sys.argv[1:3]
125     except ValueError:
126         script = sys.argv[0]
127         print('usage: {} <junitxml filepath> <expected successes filepath>'
128               .format(script), file=sys.stderr)
129         result = 1
130     else:
131         result = main(xml_path, txt_path)
132     sys.exit(result)
133
```

Path: `c7n/resources/glacier.py`
Content:
```
1 # Copyright 2016 Capital One Services, LLC
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 # http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 from __future__ import absolute_import, division, print_function, unicode_literals
15 
16 from botocore.exceptions import ClientError
17 
18 from c7n.filters import CrossAccountAccessFilter
19 from c7n.query import QueryResourceManager
20 from c7n.manager import resources
21 from c7n.utils import get_retry, local_session
22 
23 
24 @resources.register('glacier')
25 class Glacier(QueryResourceManager):
26 
27     permissions = ('glacier:ListTagsForVault',)
28     retry = staticmethod(get_retry(('Throttled',)))
29 
30     class resource_type(object):
31         service = 'glacier'
32         enum_spec = ('list_vaults', 'VaultList', None)
33         name = "VaultName"
34         id = "VaultARN"
35         dimension = None
36         universal_taggable = True
37 
38     def augment(self, resources):
39         def process_tags(resource):
40             client = local_session(self.session_factory).client('glacier')
41             tag_dict = self.retry(
42                 client.list_tags_for_vault,
43                 vaultName=resource[self.get_model().name])['Tags']
44             tag_list = []
45             for k, v in tag_dict.items():
46                 tag_list.append({'Key': k, 'Value': v})
47             resource['Tags'] = tag_list
48             return resource
49 
50         with self.executor_factory(max_workers=2) as w:
51             return list(w.map(process_tags, resources))
52 
53 
54 @Glacier.filter_registry.register('cross-account')
55 class GlacierCrossAccountAccessFilter(CrossAccountAccessFilter):
56     """Filter to return all glacier vaults with cross account access permissions
57 
58     The whitelist parameter will omit the accounts that match from the return
59 
60     :example:
61 
62         .. code-block:
63 
64             policies:
65               - name: glacier-cross-account
66                 resource: glacier
67                 filters:
68                   - type: cross-account
69                     whitelist:
70                       - permitted-account-01
71                       - permitted-account-02
72     """
73     permissions = ('glacier:GetVaultAccessPolicy',)
74 
75     def process(self, resources, event=None):
76         def _augment(r):
77             client = local_session(
78                 self.manager.session_factory).client('glacier')
79             try:
80                 r['Policy'] = client.get_vault_access_policy(
81                     vaultName=r['VaultName'])['policy']['Policy']
82                 return r
83             except ClientError as e:
84                 if e.response['Error']['Code'] == 'AccessDeniedException':
85                     self.log.warning(
86                         "Access denied getting policy glacier:%s",
87                         r['FunctionName'])
88 
89         self.log.debug("fetching policy for %d glacier" % len(resources))
90         with self.executor_factory(max_workers=3) as w:
91             resources = list(filter(None, w.map(_augment, resources)))
92 
93         return super(GlacierCrossAccountAccessFilter, self).process(
94             resources, event)
95
```

Path: `c7n/resources/kinesis.py`
Content:
```
1 # Copyright 2016 Capital One Services, LLC
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 # http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 from __future__ import absolute_import, division, print_function, unicode_literals
15 
16 from c7n.actions import Action
17 from c7n.manager import resources
18 from c7n.query import QueryResourceManager
19 from c7n.utils import local_session, type_schema
20 
21 
22 @resources.register('kinesis')
23 class KinesisStream(QueryResourceManager):
24 
25     class resource_type(object):
26         service = 'kinesis'
27         type = 'stream'
28         enum_spec = ('list_streams', 'StreamNames', None)
29         detail_spec = (
30             'describe_stream', 'StreamName', None, 'StreamDescription')
31         name = id = 'StreamName'
32         filter_name = None
33         filter_type = None
34         date = None
35         dimension = 'StreamName'
36 
37 
38 @KinesisStream.action_registry.register('delete')
39 class Delete(Action):
40 
41     schema = type_schema('delete')
42     permissions = ("kinesis:DeleteStream",)
43 
44     def process(self, resources):
45         client = local_session(self.manager.session_factory).client('kinesis')
46         not_active = [r['StreamName'] for r in resources
47                       if r['StreamStatus'] != 'ACTIVE']
48         self.log.warning(
49             "The following streams cannot be deleted (wrong state): %s" % (
50                 ", ".join(not_active)))
51         for r in resources:
52             if not r['StreamStatus'] == 'ACTIVE':
53                 continue
54             client.delete_stream(
55                 StreamName=r['StreamName'])
56 
57 
58 @resources.register('firehose')
59 class DeliveryStream(QueryResourceManager):
60 
61     class resource_type(object):
62         service = 'firehose'
63         type = 'deliverystream'
64         enum_spec = ('list_delivery_streams', 'DeliveryStreamNames', None)
65         detail_spec = (
66             'describe_delivery_stream', 'DeliveryStreamName', None,
67             'DeliveryStreamDescription')
68         name = id = 'DeliveryStreamName'
69         filter_name = None
70         filter_type = None
71         date = 'CreateTimestamp'
72         dimension = 'DeliveryStreamName'
73 
74 
75 @DeliveryStream.action_registry.register('delete')
76 class FirehoseDelete(Action):
77 
78     schema = type_schema('delete')
79     permissions = ("firehose:DeleteDeliveryStream",)
80 
81     def process(self, resources):
82         client = local_session(self.manager.session_factory).client('firehose')
83         creating = [r['DeliveryStreamName'] for r in resources
84                     if r['DeliveryStreamStatus'] == 'CREATING']
85         if creating:
86             self.log.warning(
87                 "These delivery streams can't be deleted (wrong state): %s" % (
88                     ", ".join(creating)))
89         for r in resources:
90             if not r['DeliveryStreamStatus'] == 'ACTIVE':
91                 continue
92             client.delete_delivery_stream(
93                 DeliveryStreamName=r['DeliveryStreamName'])
94 
95 
96 @resources.register('kinesis-analytics')
97 class AnalyticsApp(QueryResourceManager):
98 
99     class resource_type(object):
100         service = "kinesisanalytics"
101         enum_spec = ('list_applications', 'ApplicationSummaries', None)
102         detail_spec = ('describe_application', 'ApplicationName',
103                        'ApplicationName', 'ApplicationDetail')
104         name = "ApplicationName"
105         id = "ApplicationARN"
106         dimension = None
107         filter_name = None
108         filter_type = None
109 
110 
111 @AnalyticsApp.action_registry.register('delete')
112 class AppDelete(Action):
113 
114     schema = type_schema('delete')
115     permissions = ("kinesisanalytics:DeleteApplication",)
116 
117     def process(self, resources):
118         client = local_session(
119             self.manager.session_factory).client('kinesisanalytics')
120         for r in resources:
121             client.delete_application(
122                 ApplicationName=r['ApplicationName'],
123                 CreateTimestamp=r['CreateTimestamp'])
124
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/c7n/resources/glacier.py b/c7n/resources/glacier.py
--- a/c7n/resources/glacier.py
+++ b/c7n/resources/glacier.py
@@ -32,6 +32,7 @@
         enum_spec = ('list_vaults', 'VaultList', None)
         name = "VaultName"
         id = "VaultARN"
+        filter_name = None
         dimension = None
         universal_taggable = True
 
diff --git a/c7n/resources/kinesis.py b/c7n/resources/kinesis.py
--- a/c7n/resources/kinesis.py
+++ b/c7n/resources/kinesis.py
@@ -16,6 +16,7 @@
 from c7n.actions import Action
 from c7n.manager import resources
 from c7n.query import QueryResourceManager
+from c7n.tags import universal_augment
 from c7n.utils import local_session, type_schema
 
 
@@ -33,6 +34,11 @@
         filter_type = None
         date = None
         dimension = 'StreamName'
+        universal_taggable = True
+
+    def augment(self, resources):
+        return universal_augment(
+            self, super(KinesisStream, self).augment(resources))
 
 
 @KinesisStream.action_registry.register('delete')
diff --git a/tools/dev/ratchet.py b/tools/dev/ratchet.py
--- a/tools/dev/ratchet.py
+++ b/tools/dev/ratchet.py
@@ -44,6 +44,8 @@
             result = nonsuccess.pop()
             self.failed.append(key)
             message = result.get('message')
+            if message is None:
+                return
             if message.startswith(INVALID_TYPE):
                 message = INVALID_TYPE
             self.failed_aggregates.setdefault(message, []).append(key)

{"golden_diff": "diff --git a/c7n/resources/glacier.py b/c7n/resources/glacier.py\n--- a/c7n/resources/glacier.py\n+++ b/c7n/resources/glacier.py\n@@ -32,6 +32,7 @@\n         enum_spec = ('list_vaults', 'VaultList', None)\n         name = \"VaultName\"\n         id = \"VaultARN\"\n+        filter_name = None\n         dimension = None\n         universal_taggable = True\n \ndiff --git a/c7n/resources/kinesis.py b/c7n/resources/kinesis.py\n--- a/c7n/resources/kinesis.py\n+++ b/c7n/resources/kinesis.py\n@@ -16,6 +16,7 @@\n from c7n.actions import Action\n from c7n.manager import resources\n from c7n.query import QueryResourceManager\n+from c7n.tags import universal_augment\n from c7n.utils import local_session, type_schema\n \n \n@@ -33,6 +34,11 @@\n         filter_type = None\n         date = None\n         dimension = 'StreamName'\n+        universal_taggable = True\n+\n+    def augment(self, resources):\n+        return universal_augment(\n+            self, super(KinesisStream, self).augment(resources))\n \n \n @KinesisStream.action_registry.register('delete')\ndiff --git a/tools/dev/ratchet.py b/tools/dev/ratchet.py\n--- a/tools/dev/ratchet.py\n+++ b/tools/dev/ratchet.py\n@@ -44,6 +44,8 @@\n             result = nonsuccess.pop()\n             self.failed.append(key)\n             message = result.get('message')\n+            if message is None:\n+                return\n             if message.startswith(INVALID_TYPE):\n                 message = INVALID_TYPE\n             self.failed_aggregates.setdefault(message, []).append(key)\n", "issue": "mark-for-op and tag actions missing for resource=kinesis\nThe mark-for-op and tag actions are available for ec2, rds, and other resource types is not available for kinesis, but would desire them to be\n", "before_files": [{"content": "#!/usr/bin/env python3.6\n# -*- coding: utf-8 -*-\n\"\"\"Ratchet up successes under Python 3.6.\n\"\"\"\nfrom __future__ import (\n    absolute_import, division, print_function, unicode_literals)\n\nimport sys\nimport xml.etree.ElementTree as Etree\n\n\nINVALID_TYPE = 'E   botocore.exceptions.ParamValidationError: ' \\\n    'Parameter validation failed:     Invalid type for parameter '\n\n\nclass TestResults(object):\n\n    def __init__(self, results_path):\n        self.results_path = results_path\n        self.failed = []\n        self.failed_aggregates = {}\n        self.stderr_output = []\n        self.passed = []\n        self._tree = None\n\n    def parse(self):\n        if self._tree:\n            raise AssertionError(\"Already Parsed\")\n        self._tree = Etree.parse(self.results_path)\n        for testcase in self._tree.findall('testcase'):\n            self.process_testcase(testcase)\n        return self\n\n    def process_testcase(self, case):\n        key = self.case_key(case)\n\n        # look at children but throw away stderr output\n        nonsuccess = [c for c in case if not c.tag == 'system-err']\n        n = len(nonsuccess)\n        if n > 1:\n            raise AssertionError(\"multiple results for %s: %s\" %\n                (key, nonsuccess))\n        elif n == 1:\n            result = nonsuccess.pop()\n            self.failed.append(key)\n            message = result.get('message')\n            if message.startswith(INVALID_TYPE):\n                message = INVALID_TYPE\n            self.failed_aggregates.setdefault(message, []).append(key)\n        else:\n            self.passed.append(key)\n\n    @staticmethod\n    def case_key(case):\n        return \"%s.%s\" % (case.get('classname'), case.get('name'))\n\n    def report(self, details=False):\n        for k, v in sorted(\n                self.failed_aggregates.items(),\n                key = lambda i: len(i[1]),\n                reverse=True):\n            print(\"# %s\" % k)\n            for t in v:\n                print(\" - %s\" % t)\n\n\ndef load_expected_failures(txt):\n    expected_failures = open(txt).read()\n    parsed = set()\n    for line in expected_failures.splitlines():\n        if not line or line.startswith('#'):\n            continue\n        parsed.add(line)\n    return parsed\n\n\ndef list_tests(tests):\n    for test in sorted(tests):\n        print(' ', test)\n\n\ndef update_expectation(txt_path, tests):\n    new = set(t + '\\n' for t in tests)\n    open(txt_path, 'w+').writelines(sorted(new))\n\n\ndef main(xml_path, txt_path):\n    \"\"\"Takes two paths, one to XML output from pytest, the other to a text file\n    listing expected successes. Walks the former looking for the latter.\n    \"\"\"\n    results = TestResults(xml_path).parse()\n\n    if txt_path == '-':\n        results.report()\n        return\n\n    previous = load_expected_failures(txt_path)\n    current = set(results.failed)\n\n    expected = previous - current\n    if expected:\n        print(\"Some tests expected to fail under Python 3.6 didn't:\")\n        list_tests(expected)\n        update_expectation(txt_path, current)\n        print(\"Conveniently, they have been removed from {} for you. Perhaps \"\n            \"commit that?\".format(txt_path))\n\n    unexpected = current - previous\n    if unexpected:\n        print(\"Some tests not expected to fail under Python 3.6 did:\")\n        list_tests(unexpected)\n\n    if expected or unexpected:\n        print(\"Previously %d tests failed under Python 3.6, now %d did.\" %\n            (len(previous), len(current)))\n        return 1\n\n    print('All and only tests expected to fail under Python 3.6 did.')\n    return 0\n\n\nif __name__ == '__main__':\n    try:\n        xml_path, txt_path = sys.argv[1:3]\n    except ValueError:\n        script = sys.argv[0]\n        print('usage: {} <junitxml filepath> <expected successes filepath>'\n              .format(script), file=sys.stderr)\n        result = 1\n    else:\n        result = main(xml_path, txt_path)\n    sys.exit(result)\n", "path": "tools/dev/ratchet.py"}, {"content": "# Copyright 2016 Capital One Services, LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nfrom __future__ import absolute_import, division, print_function, unicode_literals\n\nfrom botocore.exceptions import ClientError\n\nfrom c7n.filters import CrossAccountAccessFilter\nfrom c7n.query import QueryResourceManager\nfrom c7n.manager import resources\nfrom c7n.utils import get_retry, local_session\n\n\n@resources.register('glacier')\nclass Glacier(QueryResourceManager):\n\n    permissions = ('glacier:ListTagsForVault',)\n    retry = staticmethod(get_retry(('Throttled',)))\n\n    class resource_type(object):\n        service = 'glacier'\n        enum_spec = ('list_vaults', 'VaultList', None)\n        name = \"VaultName\"\n        id = \"VaultARN\"\n        dimension = None\n        universal_taggable = True\n\n    def augment(self, resources):\n        def process_tags(resource):\n            client = local_session(self.session_factory).client('glacier')\n            tag_dict = self.retry(\n                client.list_tags_for_vault,\n                vaultName=resource[self.get_model().name])['Tags']\n            tag_list = []\n            for k, v in tag_dict.items():\n                tag_list.append({'Key': k, 'Value': v})\n            resource['Tags'] = tag_list\n            return resource\n\n        with self.executor_factory(max_workers=2) as w:\n            return list(w.map(process_tags, resources))\n\n\n@Glacier.filter_registry.register('cross-account')\nclass GlacierCrossAccountAccessFilter(CrossAccountAccessFilter):\n    \"\"\"Filter to return all glacier vaults with cross account access permissions\n\n    The whitelist parameter will omit the accounts that match from the return\n\n    :example:\n\n        .. code-block:\n\n            policies:\n              - name: glacier-cross-account\n                resource: glacier\n                filters:\n                  - type: cross-account\n                    whitelist:\n                      - permitted-account-01\n                      - permitted-account-02\n    \"\"\"\n    permissions = ('glacier:GetVaultAccessPolicy',)\n\n    def process(self, resources, event=None):\n        def _augment(r):\n            client = local_session(\n                self.manager.session_factory).client('glacier')\n            try:\n                r['Policy'] = client.get_vault_access_policy(\n                    vaultName=r['VaultName'])['policy']['Policy']\n                return r\n            except ClientError as e:\n                if e.response['Error']['Code'] == 'AccessDeniedException':\n                    self.log.warning(\n                        \"Access denied getting policy glacier:%s\",\n                        r['FunctionName'])\n\n        self.log.debug(\"fetching policy for %d glacier\" % len(resources))\n        with self.executor_factory(max_workers=3) as w:\n            resources = list(filter(None, w.map(_augment, resources)))\n\n        return super(GlacierCrossAccountAccessFilter, self).process(\n            resources, event)\n", "path": "c7n/resources/glacier.py"}, {"content": "# Copyright 2016 Capital One Services, LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nfrom __future__ import absolute_import, division, print_function, unicode_literals\n\nfrom c7n.actions import Action\nfrom c7n.manager import resources\nfrom c7n.query import QueryResourceManager\nfrom c7n.utils import local_session, type_schema\n\n\n@resources.register('kinesis')\nclass KinesisStream(QueryResourceManager):\n\n    class resource_type(object):\n        service = 'kinesis'\n        type = 'stream'\n        enum_spec = ('list_streams', 'StreamNames', None)\n        detail_spec = (\n            'describe_stream', 'StreamName', None, 'StreamDescription')\n        name = id = 'StreamName'\n        filter_name = None\n        filter_type = None\n        date = None\n        dimension = 'StreamName'\n\n\n@KinesisStream.action_registry.register('delete')\nclass Delete(Action):\n\n    schema = type_schema('delete')\n    permissions = (\"kinesis:DeleteStream\",)\n\n    def process(self, resources):\n        client = local_session(self.manager.session_factory).client('kinesis')\n        not_active = [r['StreamName'] for r in resources\n                      if r['StreamStatus'] != 'ACTIVE']\n        self.log.warning(\n            \"The following streams cannot be deleted (wrong state): %s\" % (\n                \", \".join(not_active)))\n        for r in resources:\n            if not r['StreamStatus'] == 'ACTIVE':\n                continue\n            client.delete_stream(\n                StreamName=r['StreamName'])\n\n\n@resources.register('firehose')\nclass DeliveryStream(QueryResourceManager):\n\n    class resource_type(object):\n        service = 'firehose'\n        type = 'deliverystream'\n        enum_spec = ('list_delivery_streams', 'DeliveryStreamNames', None)\n        detail_spec = (\n            'describe_delivery_stream', 'DeliveryStreamName', None,\n            'DeliveryStreamDescription')\n        name = id = 'DeliveryStreamName'\n        filter_name = None\n        filter_type = None\n        date = 'CreateTimestamp'\n        dimension = 'DeliveryStreamName'\n\n\n@DeliveryStream.action_registry.register('delete')\nclass FirehoseDelete(Action):\n\n    schema = type_schema('delete')\n    permissions = (\"firehose:DeleteDeliveryStream\",)\n\n    def process(self, resources):\n        client = local_session(self.manager.session_factory).client('firehose')\n        creating = [r['DeliveryStreamName'] for r in resources\n                    if r['DeliveryStreamStatus'] == 'CREATING']\n        if creating:\n            self.log.warning(\n                \"These delivery streams can't be deleted (wrong state): %s\" % (\n                    \", \".join(creating)))\n        for r in resources:\n            if not r['DeliveryStreamStatus'] == 'ACTIVE':\n                continue\n            client.delete_delivery_stream(\n                DeliveryStreamName=r['DeliveryStreamName'])\n\n\n@resources.register('kinesis-analytics')\nclass AnalyticsApp(QueryResourceManager):\n\n    class resource_type(object):\n        service = \"kinesisanalytics\"\n        enum_spec = ('list_applications', 'ApplicationSummaries', None)\n        detail_spec = ('describe_application', 'ApplicationName',\n                       'ApplicationName', 'ApplicationDetail')\n        name = \"ApplicationName\"\n        id = \"ApplicationARN\"\n        dimension = None\n        filter_name = None\n        filter_type = None\n\n\n@AnalyticsApp.action_registry.register('delete')\nclass AppDelete(Action):\n\n    schema = type_schema('delete')\n    permissions = (\"kinesisanalytics:DeleteApplication\",)\n\n    def process(self, resources):\n        client = local_session(\n            self.manager.session_factory).client('kinesisanalytics')\n        for r in resources:\n            client.delete_application(\n                ApplicationName=r['ApplicationName'],\n                CreateTimestamp=r['CreateTimestamp'])\n", "path": "c7n/resources/kinesis.py"}], "after_files": [{"content": "#!/usr/bin/env python3.6\n# -*- coding: utf-8 -*-\n\"\"\"Ratchet up successes under Python 3.6.\n\"\"\"\nfrom __future__ import (\n    absolute_import, division, print_function, unicode_literals)\n\nimport sys\nimport xml.etree.ElementTree as Etree\n\n\nINVALID_TYPE = 'E   botocore.exceptions.ParamValidationError: ' \\\n    'Parameter validation failed:     Invalid type for parameter '\n\n\nclass TestResults(object):\n\n    def __init__(self, results_path):\n        self.results_path = results_path\n        self.failed = []\n        self.failed_aggregates = {}\n        self.stderr_output = []\n        self.passed = []\n        self._tree = None\n\n    def parse(self):\n        if self._tree:\n            raise AssertionError(\"Already Parsed\")\n        self._tree = Etree.parse(self.results_path)\n        for testcase in self._tree.findall('testcase'):\n            self.process_testcase(testcase)\n        return self\n\n    def process_testcase(self, case):\n        key = self.case_key(case)\n\n        # look at children but throw away stderr output\n        nonsuccess = [c for c in case if not c.tag == 'system-err']\n        n = len(nonsuccess)\n        if n > 1:\n            raise AssertionError(\"multiple results for %s: %s\" %\n                (key, nonsuccess))\n        elif n == 1:\n            result = nonsuccess.pop()\n            self.failed.append(key)\n            message = result.get('message')\n            if message is None:\n                return\n            if message.startswith(INVALID_TYPE):\n                message = INVALID_TYPE\n            self.failed_aggregates.setdefault(message, []).append(key)\n        else:\n            self.passed.append(key)\n\n    @staticmethod\n    def case_key(case):\n        return \"%s.%s\" % (case.get('classname'), case.get('name'))\n\n    def report(self, details=False):\n        for k, v in sorted(\n                self.failed_aggregates.items(),\n                key = lambda i: len(i[1]),\n                reverse=True):\n            print(\"# %s\" % k)\n            for t in v:\n                print(\" - %s\" % t)\n\n\ndef load_expected_failures(txt):\n    expected_failures = open(txt).read()\n    parsed = set()\n    for line in expected_failures.splitlines():\n        if not line or line.startswith('#'):\n            continue\n        parsed.add(line)\n    return parsed\n\n\ndef list_tests(tests):\n    for test in sorted(tests):\n        print(' ', test)\n\n\ndef update_expectation(txt_path, tests):\n    new = set(t + '\\n' for t in tests)\n    open(txt_path, 'w+').writelines(sorted(new))\n\n\ndef main(xml_path, txt_path):\n    \"\"\"Takes two paths, one to XML output from pytest, the other to a text file\n    listing expected successes. Walks the former looking for the latter.\n    \"\"\"\n    results = TestResults(xml_path).parse()\n\n    if txt_path == '-':\n        results.report()\n        return\n\n    previous = load_expected_failures(txt_path)\n    current = set(results.failed)\n\n    expected = previous - current\n    if expected:\n        print(\"Some tests expected to fail under Python 3.6 didn't:\")\n        list_tests(expected)\n        update_expectation(txt_path, current)\n        print(\"Conveniently, they have been removed from {} for you. Perhaps \"\n            \"commit that?\".format(txt_path))\n\n    unexpected = current - previous\n    if unexpected:\n        print(\"Some tests not expected to fail under Python 3.6 did:\")\n        list_tests(unexpected)\n\n    if expected or unexpected:\n        print(\"Previously %d tests failed under Python 3.6, now %d did.\" %\n            (len(previous), len(current)))\n        return 1\n\n    print('All and only tests expected to fail under Python 3.6 did.')\n    return 0\n\n\nif __name__ == '__main__':\n    try:\n        xml_path, txt_path = sys.argv[1:3]\n    except ValueError:\n        script = sys.argv[0]\n        print('usage: {} <junitxml filepath> <expected successes filepath>'\n              .format(script), file=sys.stderr)\n        result = 1\n    else:\n        result = main(xml_path, txt_path)\n    sys.exit(result)\n", "path": "tools/dev/ratchet.py"}, {"content": "# Copyright 2016 Capital One Services, LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nfrom __future__ import absolute_import, division, print_function, unicode_literals\n\nfrom botocore.exceptions import ClientError\n\nfrom c7n.filters import CrossAccountAccessFilter\nfrom c7n.query import QueryResourceManager\nfrom c7n.manager import resources\nfrom c7n.utils import get_retry, local_session\n\n\n@resources.register('glacier')\nclass Glacier(QueryResourceManager):\n\n    permissions = ('glacier:ListTagsForVault',)\n    retry = staticmethod(get_retry(('Throttled',)))\n\n    class resource_type(object):\n        service = 'glacier'\n        enum_spec = ('list_vaults', 'VaultList', None)\n        name = \"VaultName\"\n        id = \"VaultARN\"\n        filter_name = None\n        dimension = None\n        universal_taggable = True\n\n    def augment(self, resources):\n        def process_tags(resource):\n            client = local_session(self.session_factory).client('glacier')\n            tag_dict = self.retry(\n                client.list_tags_for_vault,\n                vaultName=resource[self.get_model().name])['Tags']\n            tag_list = []\n            for k, v in tag_dict.items():\n                tag_list.append({'Key': k, 'Value': v})\n            resource['Tags'] = tag_list\n            return resource\n\n        with self.executor_factory(max_workers=2) as w:\n            return list(w.map(process_tags, resources))\n\n\n@Glacier.filter_registry.register('cross-account')\nclass GlacierCrossAccountAccessFilter(CrossAccountAccessFilter):\n    \"\"\"Filter to return all glacier vaults with cross account access permissions\n\n    The whitelist parameter will omit the accounts that match from the return\n\n    :example:\n\n        .. code-block:\n\n            policies:\n              - name: glacier-cross-account\n                resource: glacier\n                filters:\n                  - type: cross-account\n                    whitelist:\n                      - permitted-account-01\n                      - permitted-account-02\n    \"\"\"\n    permissions = ('glacier:GetVaultAccessPolicy',)\n\n    def process(self, resources, event=None):\n        def _augment(r):\n            client = local_session(\n                self.manager.session_factory).client('glacier')\n            try:\n                r['Policy'] = client.get_vault_access_policy(\n                    vaultName=r['VaultName'])['policy']['Policy']\n                return r\n            except ClientError as e:\n                if e.response['Error']['Code'] == 'AccessDeniedException':\n                    self.log.warning(\n                        \"Access denied getting policy glacier:%s\",\n                        r['FunctionName'])\n\n        self.log.debug(\"fetching policy for %d glacier\" % len(resources))\n        with self.executor_factory(max_workers=3) as w:\n            resources = list(filter(None, w.map(_augment, resources)))\n\n        return super(GlacierCrossAccountAccessFilter, self).process(\n            resources, event)\n", "path": "c7n/resources/glacier.py"}, {"content": "# Copyright 2016 Capital One Services, LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nfrom __future__ import absolute_import, division, print_function, unicode_literals\n\nfrom c7n.actions import Action\nfrom c7n.manager import resources\nfrom c7n.query import QueryResourceManager\nfrom c7n.tags import universal_augment\nfrom c7n.utils import local_session, type_schema\n\n\n@resources.register('kinesis')\nclass KinesisStream(QueryResourceManager):\n\n    class resource_type(object):\n        service = 'kinesis'\n        type = 'stream'\n        enum_spec = ('list_streams', 'StreamNames', None)\n        detail_spec = (\n            'describe_stream', 'StreamName', None, 'StreamDescription')\n        name = id = 'StreamName'\n        filter_name = None\n        filter_type = None\n        date = None\n        dimension = 'StreamName'\n        universal_taggable = True\n\n    def augment(self, resources):\n        return universal_augment(\n            self, super(KinesisStream, self).augment(resources))\n\n\n@KinesisStream.action_registry.register('delete')\nclass Delete(Action):\n\n    schema = type_schema('delete')\n    permissions = (\"kinesis:DeleteStream\",)\n\n    def process(self, resources):\n        client = local_session(self.manager.session_factory).client('kinesis')\n        not_active = [r['StreamName'] for r in resources\n                      if r['StreamStatus'] != 'ACTIVE']\n        self.log.warning(\n            \"The following streams cannot be deleted (wrong state): %s\" % (\n                \", \".join(not_active)))\n        for r in resources:\n            if not r['StreamStatus'] == 'ACTIVE':\n                continue\n            client.delete_stream(\n                StreamName=r['StreamName'])\n\n\n@resources.register('firehose')\nclass DeliveryStream(QueryResourceManager):\n\n    class resource_type(object):\n        service = 'firehose'\n        type = 'deliverystream'\n        enum_spec = ('list_delivery_streams', 'DeliveryStreamNames', None)\n        detail_spec = (\n            'describe_delivery_stream', 'DeliveryStreamName', None,\n            'DeliveryStreamDescription')\n        name = id = 'DeliveryStreamName'\n        filter_name = None\n        filter_type = None\n        date = 'CreateTimestamp'\n        dimension = 'DeliveryStreamName'\n\n\n@DeliveryStream.action_registry.register('delete')\nclass FirehoseDelete(Action):\n\n    schema = type_schema('delete')\n    permissions = (\"firehose:DeleteDeliveryStream\",)\n\n    def process(self, resources):\n        client = local_session(self.manager.session_factory).client('firehose')\n        creating = [r['DeliveryStreamName'] for r in resources\n                    if r['DeliveryStreamStatus'] == 'CREATING']\n        if creating:\n            self.log.warning(\n                \"These delivery streams can't be deleted (wrong state): %s\" % (\n                    \", \".join(creating)))\n        for r in resources:\n            if not r['DeliveryStreamStatus'] == 'ACTIVE':\n                continue\n            client.delete_delivery_stream(\n                DeliveryStreamName=r['DeliveryStreamName'])\n\n\n@resources.register('kinesis-analytics')\nclass AnalyticsApp(QueryResourceManager):\n\n    class resource_type(object):\n        service = \"kinesisanalytics\"\n        enum_spec = ('list_applications', 'ApplicationSummaries', None)\n        detail_spec = ('describe_application', 'ApplicationName',\n                       'ApplicationName', 'ApplicationDetail')\n        name = \"ApplicationName\"\n        id = \"ApplicationARN\"\n        dimension = None\n        filter_name = None\n        filter_type = None\n\n\n@AnalyticsApp.action_registry.register('delete')\nclass AppDelete(Action):\n\n    schema = type_schema('delete')\n    permissions = (\"kinesisanalytics:DeleteApplication\",)\n\n    def process(self, resources):\n        client = local_session(\n            self.manager.session_factory).client('kinesisanalytics')\n        for r in resources:\n            client.delete_application(\n                ApplicationName=r['ApplicationName'],\n                CreateTimestamp=r['CreateTimestamp'])\n", "path": "c7n/resources/kinesis.py"}]}