blacked

api/tacticalrmm/core/utils.py

@@ -228,15 +228,19 @@ def find_and_replace_db_values_str(*, text: str, instance):
  return text
 
 
-def find_and_replace_db_values_dict(*, dict_value: Dict[str, Any], instance)-> Dict[str, Any]:
+def find_and_replace_db_values_dict(
+ *, dict_value: Dict[str, Any], instance
+) -> Dict[str, Any]:
  new_dict = {}
 
  for key, value in dict_value.items():
  new_key = find_and_replace_db_values_str(text=key, instance=instance)
 
  # Check if the value is a dictionary and recursively call the function if it is
  if isinstance(value, dict):
- new_value = find_and_replace_db_values_dict(dict_value=value, instance=instance)
+ new_value = find_and_replace_db_values_dict(
+ dict_value=value, instance=instance
+ )
  else:
  new_value = find_and_replace_db_values_str(text=value, instance=instance)
 
@@ -248,7 +252,9 @@ def find_and_replace_db_values_dict(*, dict_value: Dict[str, Any], instance)-> D
 def _run_url_rest_action(*, url: str, method, body: str, headers: str, instance=None):
  # replace url
  new_url = find_and_replace_db_values_str(text=url, instance=instance)
- new_body = find_and_replace_db_values_dict(dict_value=json.loads(body), instance=instance)
+ new_body = find_and_replace_db_values_dict(
+ dict_value=json.loads(body), instance=instance
+ )
  new_headers = find_and_replace_db_values_dict(
  dict_value=json.loads(headers), instance=instance
  )
@@ -258,6 +264,7 @@ def _run_url_rest_action(*, url: str, method, body: str, headers: str, instance=
  else:
  return getattr(requests, method)(new_url, data=new_body, headers=new_headers)
 
+
 def run_url_rest_action(*, action_id: int, instance=None) -> Tuple[str, int]:
  import core.models
 
@@ -267,11 +274,22 @@ def run_url_rest_action(*, action_id: int, instance=None) -> Tuple[str, int]:
  body = action.rest_body
  headers = action.rest_headers
 
- response = _run_url_rest_action(url=url, method=method, body=body, headers=headers, instance=instance)
+ response = _run_url_rest_action(
+ url=url, method=method, body=body, headers=headers, instance=instance
+ )
 
  return (response.text, response.status_code)
 
-def run_test_url_rest_action(*, url: str, method, body: str, headers: str, instance_type: Optional[str], instance_id: Optional[int]) -> Tuple[str, int]:
+
+def run_test_url_rest_action(
+ *,
+ url: str,
+ method,
+ body: str,
+ headers: str,
+ instance_type: Optional[str],
+ instance_id: Optional[int],
+) -> Tuple[str, int]:
  lookup_instance = None
  if instance_type and instance_id:
  if instance_type == "client":
@@ -282,12 +300,15 @@ def run_test_url_rest_action(*, url: str, method, body: str, headers: str, inst
  lookup_instance = Site.objects.get(pk=instance_id)
  elif instance_type == "agent":
  Agent = apps.get_model("agents.Agent")
- lookup_instance = Agent.objects.get(pk=instance_id) 
+ lookup_instance = Agent.objects.get(pk=instance_id)
 
- response = _run_url_rest_action(url=url, method=method, body=body, headers=headers, instance=lookup_instance)
+ response = _run_url_rest_action(
+ url=url, method=method, body=body, headers=headers, instance=lookup_instance
+ )
 
  return (response.text, response.status_code)
 
+
 def run_server_task(*, server_task_id: int):
  from autotasks.models import AutomatedTask, TaskResult
  from tacticalrmm.constants import TaskStatus

api/tacticalrmm/core/views.py

@@ -29,7 +29,7 @@
  token_is_valid,
  run_server_script,
  run_server_task,
- run_test_url_rest_action
+ run_test_url_rest_action,
 )
 from logs.models import AuditLog
 from tacticalrmm.constants import AuditActionType, PAStatus
@@ -424,15 +424,20 @@ def patch(self, request):
 
  return Response(requote_uri(url_pattern))
 
+
 class RunTestURLAction(APIView):
  permission_classes = [IsAuthenticated, URLActionPerms]
 
  class InputSerializer(serializers.Serializer):
  pattern = serializers.CharField(required=True)
  rest_body = serializers.CharField()
  rest_headers = serializers.CharField()
- rest_method = serializers.ChoiceField(required=True, choices=["get", "post", "put", "delete", "patch"])
- run_instance_type = serializers.ChoiceField(choices=["agent", "client", "site", "none"])
+ rest_method = serializers.ChoiceField(
+ required=True, choices=["get", "post", "put", "delete", "patch"]
+ )
+ run_instance_type = serializers.ChoiceField(
+ choices=["agent", "client", "site", "none"]
+ )
  run_instance_id = serializers.IntegerField(allow_null=True)
 
  def post(self, request):
@@ -446,10 +451,18 @@ def post(self, request):
  method = serializer.validated_data.get("rest_method")
  instance_type = serializer.validated_data.get("run_instance_type", None)
  instance_id = serializer.validated_data.get("instance_id", None)
- result, _ = run_test_url_rest_action(url=url, body=body, headers=headers, method=method, instance_type=instance_type, instance_id=instance_id)
+ result, _ = run_test_url_rest_action(
+ url=url,
+ body=body,
+ headers=headers,
+ method=method,
+ instance_type=instance_type,
+ instance_id=instance_id,
+ )
 
  return Response({"url": url, "result": result})
 
+
 class ServerTaskSerializer(serializers.ModelSerializer):
 
  task_result = serializers.SerializerMethodField()