[#311] Add AWS CLI command to report from Boto3 request
Some checks failed
DCO action / DCO (pull_request) Has been cancelled
Some checks failed
DCO action / DCO (pull_request) Has been cancelled
Signed-off-by: Kirill Sosnovskikh <k.sosnovskikh@yadro.com>
This commit is contained in:
parent
b2bf6677f1
commit
c4173bf804
2 changed files with 672 additions and 400 deletions
File diff suppressed because it is too large
Load diff
|
@ -9,13 +9,12 @@ import csv
|
|||
import json
|
||||
import logging
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from contextlib import suppress
|
||||
from datetime import datetime
|
||||
from io import StringIO
|
||||
from textwrap import shorten
|
||||
from typing import Dict, List, Optional, TypedDict, Union
|
||||
from typing import Any, Optional, Union
|
||||
|
||||
import pexpect
|
||||
|
||||
|
@ -75,22 +74,75 @@ def _attach_allure_log(cmd: str, output: str, return_code: int, start_time: date
|
|||
reporter.attach(command_attachment, "Command execution")
|
||||
|
||||
|
||||
def log_command_execution(url: str, cmd: str, output: Union[str, dict], params: Optional[dict] = None) -> None:
|
||||
def log_command_execution(cmd: str, output: Union[str, dict], params: Optional[dict] = None, **kwargs) -> None:
|
||||
logger.info(f"{cmd}: {output}")
|
||||
|
||||
with suppress(Exception):
|
||||
json_output = json.dumps(output, indent=4, sort_keys=True)
|
||||
output = json_output
|
||||
if not params:
|
||||
params = {}
|
||||
|
||||
output_params = params
|
||||
|
||||
try:
|
||||
json_params = json.dumps(params, indent=4, sort_keys=True)
|
||||
json_params = json.dumps(params, indent=4, sort_keys=True, default=str)
|
||||
except TypeError as err:
|
||||
logger.warning(f"Failed to serialize '{cmd}' request parameters:\n{params}\nException: {err}")
|
||||
else:
|
||||
params = json_params
|
||||
output_params = json_params
|
||||
|
||||
command_attachment = f"COMMAND: '{cmd}'\n" f"URL: {url}\n" f"PARAMS:\n{params}\n" f"OUTPUT:\n{output}\n"
|
||||
reporter.attach(command_attachment, "Command execution")
|
||||
output = json.dumps(output, indent=4, sort_keys=True, default=str)
|
||||
|
||||
command_execution = f"COMMAND: '{cmd}'\n" f"URL: {kwargs['endpoint']}\n" f"PARAMS:\n{output_params}\n" f"OUTPUT:\n{output}\n"
|
||||
aws_command = _convert_request_to_aws_cli_command(cmd, params, **kwargs)
|
||||
|
||||
reporter.attach(command_execution, "Command execution")
|
||||
reporter.attach(aws_command, "AWS CLI Command")
|
||||
|
||||
|
||||
def _convert_request_to_aws_cli_command(command: str, params: dict, **kwargs) -> str:
|
||||
overriden_names = [_convert_json_name_to_aws_cli(name) for name in kwargs.keys()]
|
||||
command = command.replace("_", "-")
|
||||
options = []
|
||||
|
||||
for name, value in params.items():
|
||||
name = _convert_json_name_to_aws_cli(name)
|
||||
|
||||
# To override parameters for AWS CLI
|
||||
if name in overriden_names:
|
||||
continue
|
||||
|
||||
if option := _create_option(name, value):
|
||||
options.append(option)
|
||||
|
||||
for name, value in kwargs.items():
|
||||
name = _convert_json_name_to_aws_cli(name)
|
||||
if option := _create_option(name, value):
|
||||
options.append(option)
|
||||
|
||||
options = " ".join(options)
|
||||
api = "s3api" if "s3" in kwargs["endpoint"] else "iam"
|
||||
return f"aws --no-verify-ssl --no-paginate {api} {command} {options}"
|
||||
|
||||
|
||||
def _convert_json_name_to_aws_cli(name: str) -> str:
|
||||
specific_names = {"CORSConfiguration": "cors-configuration"}
|
||||
|
||||
if aws_cli_name := specific_names.get(name):
|
||||
return aws_cli_name
|
||||
return re.sub(r"([a-z])([A-Z])", r"\1 \2", name).lower().replace(" ", "-").replace("_", "-")
|
||||
|
||||
|
||||
def _create_option(name: str, value: Any) -> str | None:
|
||||
if isinstance(value, bool) and value:
|
||||
return f"--{name}"
|
||||
|
||||
if isinstance(value, dict):
|
||||
value = json.dumps(value, indent=4, sort_keys=True, default=str)
|
||||
return f"--{name} '{value}'"
|
||||
|
||||
if value:
|
||||
return f"--{name} {value}"
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def parse_netmap_output(output: str) -> list[NodeNetmapInfo]:
|
||||
|
|
Loading…
Reference in a new issue