prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  aws - new generic resource action "invoke-stepfunction" enhancement
  Enhancement which behaves just like action invoke-lambda but invokes (aka starts execution of) step function state machines for use cases which need sequencing and conditional logic which is more easily implemented in step functions.  
  
  Doc hint: See https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/stepfunctions.html#SFN.Client.start_execution
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `c7n/resources/sfn.py`
  Content:
  ```
  1 # Copyright 2015-2017 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.manager import resources
  17 from c7n.query import QueryResourceManager
  18 from c7n.tags import Tag, RemoveTag
  19 
  20 
  21 @resources.register('step-machine')
  22 class StepFunction(QueryResourceManager):
  23     """AWS Step Functions State Machine"""
  24 
  25     class resource_type(object):
  26         service = 'stepfunctions'
  27         enum_spec = ('list_state_machines', 'stateMachines', None)
  28         arn = id = 'stateMachineArn'
  29         name = 'name'
  30         date = 'creationDate'
  31         dimension = None
  32         detail_spec = (
  33             "describe_state_machine", "stateMachineArn",
  34             'stateMachineArn', None)
  35         filter_name = None
  36 
  37 
  38 @StepFunction.action_registry.register('tag')
  39 class TagStepFunction(Tag):
  40     """Action to create tag(s) on a step function
  41 
  42     :example:
  43 
  44     .. code-block:: yaml
  45 
  46             policies:
  47               - name: tag-step-function
  48                 resource: step-machine
  49                 actions:
  50                   - type: tag
  51                     key: target-tag
  52                     value: target-tag-value
  53     """
  54 
  55     permissions = ('stepfunctions:TagResource',)
  56 
  57     def process_resource_set(self, client, resources, tags):
  58 
  59         tags_lower = []
  60 
  61         for tag in tags:
  62             tags_lower.append({k.lower(): v for k, v in tag.items()})
  63 
  64         for r in resources:
  65             client.tag_resource(resourceArn=r['stateMachineArn'], tags=tags_lower)
  66 
  67 
  68 @StepFunction.action_registry.register('remove-tag')
  69 class UnTagStepFunction(RemoveTag):
  70     """Action to create tag(s) on a step function
  71 
  72     :example:
  73 
  74     .. code-block:: yaml
  75 
  76             policies:
  77               - name: step-function-remove-tag
  78                 resource: step-machine
  79                 actions:
  80                   - type: remove-tag
  81                     tags: ["test"]
  82     """
  83 
  84     permissions = ('stepfunctions:UntagResource',)
  85 
  86     def process_resource_set(self, client, resources, tag_keys):
  87 
  88         for r in resources:
  89             client.untag_resource(resourceArn=r['stateMachineArn'], tagKeys=tag_keys)
  90
  ```
  --- 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()
  
  ```
  
golden_diff: diff --git a/c7n/resources/sfn.py b/c7n/resources/sfn.py
  --- a/c7n/resources/sfn.py
  +++ b/c7n/resources/sfn.py
  @@ -13,9 +13,12 @@
   # limitations under the License.
   from __future__ import absolute_import, division, print_function, unicode_literals
   
  +
  +from c7n.actions import Action
   from c7n.manager import resources
   from c7n.query import QueryResourceManager
   from c7n.tags import Tag, RemoveTag
  +from c7n.utils import type_schema, local_session, dumps, chunks
   
   
   @resources.register('step-machine')
  @@ -35,6 +38,95 @@
           filter_name = None
   
   
  +class InvokeStepFunction(Action):
  +    """Invoke step function on resources.
  +
  +    By default this will invoke a step function for each resource
  +    providing both the `policy` and `resource` as input.
  +
  +    That behavior can be configured setting policy and bulk
  +    boolean flags on the action.
  +
  +    If bulk action parameter is set to true, then the step
  +    function will be invoked in bulk, with a set of resource arns
  +    under the `resources` key.
  +
  +    The size of the batch can be configured via the batch-size
  +    parameter. Note step function state (input, execution, etc)must
  +    fit within 32k, we default to batch size 250.
  +
  +    :example:
  +
  +    .. code-block:: yaml
  +
  +       policies:
  +         - name: invoke-step-function
  +           resource: s3
  +           filters:
  +             - is-log-target
  +             - "tag:IngestSetup": absent
  +           actions:
  +             - type: invoke-sfn
  +               # This will cause the workflow to be invoked
  +               # with many resources arns in a single execution.
  +               # Note this is *not* the default.
  +               bulk: true
  +               batch-size: 10
  +               state-machine: LogIngestSetup
  +    """
  +
  +    schema = type_schema(
  +        'invoke-sfn',
  +        required=['state-machine'],
  +        **{'state-machine': {'type': 'string'},
  +           'batch-size': {'type': 'integer'},
  +           'bulk': {'type': 'boolean'},
  +           'policy': {'type': 'boolean'}})
  +    schema_alias = True
  +    permissions = ('stepfunctions:StartExecution',)
  +
  +    def process(self, resources):
  +        client = local_session(
  +            self.manager.session_factory).client('stepfunctions')
  +        arn = self.data['state-machine']
  +        if not arn.startswith('arn'):
  +            arn = 'arn:aws:states:{}:{}:stateMachine:{}'.format(
  +                self.manager.config.region, self.manager.config.account_id, arn)
  +
  +        params = {'stateMachineArn': arn}
  +        pinput = {}
  +
  +        if self.data.get('policy', True):
  +            pinput['policy'] = dict(self.manager.data)
  +
  +        resource_set = list(zip(self.manager.get_arns(resources), resources))
  +        if self.data.get('bulk', False) is True:
  +            return self.invoke_batch(client, params, pinput, resource_set)
  +
  +        for arn, r in resource_set:
  +            pinput['resource'] = r
  +            params['input'] = dumps(pinput)
  +            r['c7n:execution-arn'] = self.manager.retry(
  +                client.start_execution, **params).get('executionArn')
  +
  +    def invoke_batch(self, client, params, pinput, resource_set):
  +        for batch_rset in chunks(resource_set, self.data.get('batch-size', 250)):
  +            pinput['resources'] = [rarn for rarn, _ in batch_rset]
  +            params['input'] = dumps(pinput)
  +            exec_arn = self.manager.retry(
  +                client.start_execution, **params).get('executionArn')
  +            for _, r in resource_set:
  +                r['c7n:execution-arn'] = exec_arn
  +
  +    @classmethod
  +    def register(cls, registry, key):
  +        for _, r in registry.items():
  +            r.action_registry.register('invoke-sfn', cls)
  +
  +
  +resources.subscribe(resources.EVENT_FINAL, InvokeStepFunction.register)
  +
  +
   @StepFunction.action_registry.register('tag')
   class TagStepFunction(Tag):
       """Action to create tag(s) on a step function
  
verification_info: {"golden_diff": "diff --git a/c7n/resources/sfn.py b/c7n/resources/sfn.py\n--- a/c7n/resources/sfn.py\n+++ b/c7n/resources/sfn.py\n@@ -13,9 +13,12 @@\n # limitations under the License.\n from __future__ import absolute_import, division, print_function, unicode_literals\n \n+\n+from c7n.actions import Action\n from c7n.manager import resources\n from c7n.query import QueryResourceManager\n from c7n.tags import Tag, RemoveTag\n+from c7n.utils import type_schema, local_session, dumps, chunks\n \n \n @resources.register('step-machine')\n@@ -35,6 +38,95 @@\n         filter_name = None\n \n \n+class InvokeStepFunction(Action):\n+    \"\"\"Invoke step function on resources.\n+\n+    By default this will invoke a step function for each resource\n+    providing both the `policy` and `resource` as input.\n+\n+    That behavior can be configured setting policy and bulk\n+    boolean flags on the action.\n+\n+    If bulk action parameter is set to true, then the step\n+    function will be invoked in bulk, with a set of resource arns\n+    under the `resources` key.\n+\n+    The size of the batch can be configured via the batch-size\n+    parameter. Note step function state (input, execution, etc)must\n+    fit within 32k, we default to batch size 250.\n+\n+    :example:\n+\n+    .. code-block:: yaml\n+\n+       policies:\n+         - name: invoke-step-function\n+           resource: s3\n+           filters:\n+             - is-log-target\n+             - \"tag:IngestSetup\": absent\n+           actions:\n+             - type: invoke-sfn\n+               # This will cause the workflow to be invoked\n+               # with many resources arns in a single execution.\n+               # Note this is *not* the default.\n+               bulk: true\n+               batch-size: 10\n+               state-machine: LogIngestSetup\n+    \"\"\"\n+\n+    schema = type_schema(\n+        'invoke-sfn',\n+        required=['state-machine'],\n+        **{'state-machine': {'type': 'string'},\n+           'batch-size': {'type': 'integer'},\n+           'bulk': {'type': 'boolean'},\n+           'policy': {'type': 'boolean'}})\n+    schema_alias = True\n+    permissions = ('stepfunctions:StartExecution',)\n+\n+    def process(self, resources):\n+        client = local_session(\n+            self.manager.session_factory).client('stepfunctions')\n+        arn = self.data['state-machine']\n+        if not arn.startswith('arn'):\n+            arn = 'arn:aws:states:{}:{}:stateMachine:{}'.format(\n+                self.manager.config.region, self.manager.config.account_id, arn)\n+\n+        params = {'stateMachineArn': arn}\n+        pinput = {}\n+\n+        if self.data.get('policy', True):\n+            pinput['policy'] = dict(self.manager.data)\n+\n+        resource_set = list(zip(self.manager.get_arns(resources), resources))\n+        if self.data.get('bulk', False) is True:\n+            return self.invoke_batch(client, params, pinput, resource_set)\n+\n+        for arn, r in resource_set:\n+            pinput['resource'] = r\n+            params['input'] = dumps(pinput)\n+            r['c7n:execution-arn'] = self.manager.retry(\n+                client.start_execution, **params).get('executionArn')\n+\n+    def invoke_batch(self, client, params, pinput, resource_set):\n+        for batch_rset in chunks(resource_set, self.data.get('batch-size', 250)):\n+            pinput['resources'] = [rarn for rarn, _ in batch_rset]\n+            params['input'] = dumps(pinput)\n+            exec_arn = self.manager.retry(\n+                client.start_execution, **params).get('executionArn')\n+            for _, r in resource_set:\n+                r['c7n:execution-arn'] = exec_arn\n+\n+    @classmethod\n+    def register(cls, registry, key):\n+        for _, r in registry.items():\n+            r.action_registry.register('invoke-sfn', cls)\n+\n+\n+resources.subscribe(resources.EVENT_FINAL, InvokeStepFunction.register)\n+\n+\n @StepFunction.action_registry.register('tag')\n class TagStepFunction(Tag):\n     \"\"\"Action to create tag(s) on a step function\n", "issue": "aws - new generic resource action \"invoke-stepfunction\" enhancement\nEnhancement which behaves just like action invoke-lambda but invokes (aka starts execution of) step function state machines for use cases which need sequencing and conditional logic which is more easily implemented in step functions.  \r\n\r\nDoc hint: See https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/stepfunctions.html#SFN.Client.start_execution\n", "before_files": [{"content": "# Copyright 2015-2017 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.manager import resources\nfrom c7n.query import QueryResourceManager\nfrom c7n.tags import Tag, RemoveTag\n\n\n@resources.register('step-machine')\nclass StepFunction(QueryResourceManager):\n    \"\"\"AWS Step Functions State Machine\"\"\"\n\n    class resource_type(object):\n        service = 'stepfunctions'\n        enum_spec = ('list_state_machines', 'stateMachines', None)\n        arn = id = 'stateMachineArn'\n        name = 'name'\n        date = 'creationDate'\n        dimension = None\n        detail_spec = (\n            \"describe_state_machine\", \"stateMachineArn\",\n            'stateMachineArn', None)\n        filter_name = None\n\n\n@StepFunction.action_registry.register('tag')\nclass TagStepFunction(Tag):\n    \"\"\"Action to create tag(s) on a step function\n\n    :example:\n\n    .. code-block:: yaml\n\n            policies:\n              - name: tag-step-function\n                resource: step-machine\n                actions:\n                  - type: tag\n                    key: target-tag\n                    value: target-tag-value\n    \"\"\"\n\n    permissions = ('stepfunctions:TagResource',)\n\n    def process_resource_set(self, client, resources, tags):\n\n        tags_lower = []\n\n        for tag in tags:\n            tags_lower.append({k.lower(): v for k, v in tag.items()})\n\n        for r in resources:\n            client.tag_resource(resourceArn=r['stateMachineArn'], tags=tags_lower)\n\n\n@StepFunction.action_registry.register('remove-tag')\nclass UnTagStepFunction(RemoveTag):\n    \"\"\"Action to create tag(s) on a step function\n\n    :example:\n\n    .. code-block:: yaml\n\n            policies:\n              - name: step-function-remove-tag\n                resource: step-machine\n                actions:\n                  - type: remove-tag\n                    tags: [\"test\"]\n    \"\"\"\n\n    permissions = ('stepfunctions:UntagResource',)\n\n    def process_resource_set(self, client, resources, tag_keys):\n\n        for r in resources:\n            client.untag_resource(resourceArn=r['stateMachineArn'], tagKeys=tag_keys)\n", "path": "c7n/resources/sfn.py"}], "after_files": [{"content": "# Copyright 2015-2017 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\n\nfrom c7n.actions import Action\nfrom c7n.manager import resources\nfrom c7n.query import QueryResourceManager\nfrom c7n.tags import Tag, RemoveTag\nfrom c7n.utils import type_schema, local_session, dumps, chunks\n\n\n@resources.register('step-machine')\nclass StepFunction(QueryResourceManager):\n    \"\"\"AWS Step Functions State Machine\"\"\"\n\n    class resource_type(object):\n        service = 'stepfunctions'\n        enum_spec = ('list_state_machines', 'stateMachines', None)\n        arn = id = 'stateMachineArn'\n        name = 'name'\n        date = 'creationDate'\n        dimension = None\n        detail_spec = (\n            \"describe_state_machine\", \"stateMachineArn\",\n            'stateMachineArn', None)\n        filter_name = None\n\n\nclass InvokeStepFunction(Action):\n    \"\"\"Invoke step function on resources.\n\n    By default this will invoke a step function for each resource\n    providing both the `policy` and `resource` as input.\n\n    That behavior can be configured setting policy and bulk\n    boolean flags on the action.\n\n    If bulk action parameter is set to true, then the step\n    function will be invoked in bulk, with a set of resource arns\n    under the `resources` key.\n\n    The size of the batch can be configured via the batch-size\n    parameter. Note step function state (input, execution, etc)must\n    fit within 32k, we default to batch size 250.\n\n    :example:\n\n    .. code-block:: yaml\n\n       policies:\n         - name: invoke-step-function\n           resource: s3\n           filters:\n             - is-log-target\n             - \"tag:IngestSetup\": absent\n           actions:\n             - type: invoke-sfn\n               # This will cause the workflow to be invoked\n               # with many resources arns in a single execution.\n               # Note this is *not* the default.\n               bulk: true\n               batch-size: 10\n               state-machine: LogIngestSetup\n    \"\"\"\n\n    schema = type_schema(\n        'invoke-sfn',\n        required=['state-machine'],\n        **{'state-machine': {'type': 'string'},\n           'batch-size': {'type': 'integer'},\n           'bulk': {'type': 'boolean'},\n           'policy': {'type': 'boolean'}})\n    schema_alias = True\n    permissions = ('stepfunctions:StartExecution',)\n\n    def process(self, resources):\n        client = local_session(\n            self.manager.session_factory).client('stepfunctions')\n        arn = self.data['state-machine']\n        if not arn.startswith('arn'):\n            arn = 'arn:aws:states:{}:{}:stateMachine:{}'.format(\n                self.manager.config.region, self.manager.config.account_id, arn)\n\n        params = {'stateMachineArn': arn}\n        pinput = {}\n\n        if self.data.get('policy', True):\n            pinput['policy'] = dict(self.manager.data)\n\n        resource_set = list(zip(self.manager.get_arns(resources), resources))\n        if self.data.get('bulk', False) is True:\n            return self.invoke_batch(client, params, pinput, resource_set)\n\n        for arn, r in resource_set:\n            pinput['resource'] = r\n            params['input'] = dumps(pinput)\n            r['c7n:execution-arn'] = self.manager.retry(\n                client.start_execution, **params).get('executionArn')\n\n    def invoke_batch(self, client, params, pinput, resource_set):\n        for batch_rset in chunks(resource_set, self.data.get('batch-size', 250)):\n            pinput['resources'] = [rarn for rarn, _ in batch_rset]\n            params['input'] = dumps(pinput)\n            exec_arn = self.manager.retry(\n                client.start_execution, **params).get('executionArn')\n            for _, r in resource_set:\n                r['c7n:execution-arn'] = exec_arn\n\n    @classmethod\n    def register(cls, registry, key):\n        for _, r in registry.items():\n            r.action_registry.register('invoke-sfn', cls)\n\n\nresources.subscribe(resources.EVENT_FINAL, InvokeStepFunction.register)\n\n\n@StepFunction.action_registry.register('tag')\nclass TagStepFunction(Tag):\n    \"\"\"Action to create tag(s) on a step function\n\n    :example:\n\n    .. code-block:: yaml\n\n            policies:\n              - name: tag-step-function\n                resource: step-machine\n                actions:\n                  - type: tag\n                    key: target-tag\n                    value: target-tag-value\n    \"\"\"\n\n    permissions = ('stepfunctions:TagResource',)\n\n    def process_resource_set(self, client, resources, tags):\n\n        tags_lower = []\n\n        for tag in tags:\n            tags_lower.append({k.lower(): v for k, v in tag.items()})\n\n        for r in resources:\n            client.tag_resource(resourceArn=r['stateMachineArn'], tags=tags_lower)\n\n\n@StepFunction.action_registry.register('remove-tag')\nclass UnTagStepFunction(RemoveTag):\n    \"\"\"Action to create tag(s) on a step function\n\n    :example:\n\n    .. code-block:: yaml\n\n            policies:\n              - name: step-function-remove-tag\n                resource: step-machine\n                actions:\n                  - type: remove-tag\n                    tags: [\"test\"]\n    \"\"\"\n\n    permissions = ('stepfunctions:UntagResource',)\n\n    def process_resource_set(self, client, resources, tag_keys):\n\n        for r in resources:\n            client.untag_resource(resourceArn=r['stateMachineArn'], tagKeys=tag_keys)\n", "path": "c7n/resources/sfn.py"}]}