diff --git a/plugins/modules/activation.py b/plugins/modules/activation.py index b48ade4a..bfce5b55 100644 --- a/plugins/modules/activation.py +++ b/plugins/modules/activation.py @@ -184,9 +184,9 @@ """ -import yaml -from typing import Any, Dict, List, Union +from typing import Any, Dict, List +import yaml from ansible.module_utils.basic import AnsibleModule from ..module_utils.arguments import AUTH_ARGSPEC @@ -196,7 +196,9 @@ from ..module_utils.errors import EDAError -def find_matching_source(event: Dict[str, Any], sources: List[Dict[str, Any]], module: AnsibleModule) -> Dict[str, Any]: +def find_matching_source( + event: Dict[str, Any], sources: List[Dict[str, Any]], module: AnsibleModule +) -> Dict[str, Any]: """ Finds a matching source based on the source_name in the event. Raises an error if no match is found. @@ -210,9 +212,7 @@ def find_matching_source(event: Dict[str, Any], sources: List[Dict[str, Any]], m return source # Return the matching source if found # If no match is found, raise an error - module.fail_json( - msg=f"The specified source_name {source_name} does not exist." - ) + module.fail_json(msg=f"The specified source_name {source_name} does not exist.") return {} # Explicit return to satisfy mypy @@ -261,8 +261,12 @@ def process_event_streams( # Handle source_index if event.get("source_index") is not None: try: - source_mapping["source_name"] = sources[event["source_index"]].get("name") - source_mapping["rulebook_hash"] = sources[event["source_index"]].get("rulebook_hash") + source_mapping["source_name"] = sources[event["source_index"]].get( + "name" + ) + source_mapping["rulebook_hash"] = sources[event["source_index"]].get( + "rulebook_hash" + ) except IndexError as e: module.fail_json( msg=f"The specified source_index {event['source_index']} is out of range: {e}" @@ -275,9 +279,7 @@ def process_event_streams( source_mapping["rulebook_hash"] = matching_source.get("rulebook_hash") if event.get("event_stream") is None: - module.fail_json( - msg="You must specify an event stream name." - ) + module.fail_json(msg="You must specify an event stream name.") # Lookup event_stream_id event_stream_id = lookup_resource_id( @@ -389,11 +391,13 @@ def create_params( if not is_aap_24 and module.params.get("event_streams"): # Process event streams and source mappings - activation_params["source_mappings"] = yaml.dump(process_event_streams( - rulebook_id=rulebook_id, - controller=controller, - module=module, - )) + activation_params["source_mappings"] = yaml.dump( + process_event_streams( + rulebook_id=rulebook_id, + controller=controller, + module=module, + ) + ) if not is_aap_24 and module.params.get("log_level"): activation_params["log_level"] = module.params["log_level"]