[#311] Add AWS CLI command to report from Boto3 request

Signed-off-by: Kirill Sosnovskikh <k.sosnovskikh@yadro.com>
This commit is contained in:
k.sosnovskikh 2024-10-23 14:08:54 +03:00 committed by Andrey Berezin
parent 3d6a356e20
commit 26139767f4
2 changed files with 672 additions and 400 deletions

File diff suppressed because it is too large Load diff

View file

@ -9,13 +9,12 @@ import csv
import json import json
import logging import logging
import re import re
import subprocess
import sys import sys
from contextlib import suppress from contextlib import suppress
from datetime import datetime from datetime import datetime
from io import StringIO from io import StringIO
from textwrap import shorten from textwrap import shorten
from typing import Dict, List, Optional, TypedDict, Union from typing import Any, Optional, Union
import pexpect 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") 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}") logger.info(f"{cmd}: {output}")
with suppress(Exception): if not params:
json_output = json.dumps(output, indent=4, sort_keys=True) params = {}
output = json_output
output_params = params
try: 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: except TypeError as err:
logger.warning(f"Failed to serialize '{cmd}' request parameters:\n{params}\nException: {err}") logger.warning(f"Failed to serialize '{cmd}' request parameters:\n{params}\nException: {err}")
else: 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" output = json.dumps(output, indent=4, sort_keys=True, default=str)
reporter.attach(command_attachment, "Command execution")
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]: def parse_netmap_output(output: str) -> list[NodeNetmapInfo]: