xk6-frostfs/scenarios/preset/helpers/frostfs_cli.py

import re
from helpers.cmd import execute_cmd, log

def create_container(endpoint, policy, wallet_file, wallet_config):
    if wallet_file:
        wallet_file = "--wallet " + wallet_file
    if wallet_config:
        wallet_config = "--config " + wallet_config
    cmd_line = f"frostfs-cli --rpc-endpoint {endpoint} container create {wallet_file} {wallet_config} " \
               f" --policy '{policy}' --basic-acl public-read-write --await"

    output, success = execute_cmd(cmd_line)

    if not success:
        log(f"{cmd_line}\n"
            f"Container has not been created\n"
            f"{output}", endpoint)
        return False
    
    try:
        fst_str = output.split('\n')[0]
    except Exception:
        log(f"{cmd_line}\n"
            f"Incorrect output\n"
            f"Output: {output or '<empty>'}", endpoint)
        return False
    splitted = fst_str.split(": ")
    if len(splitted) != 2:
        raise ValueError(f"no CID was parsed from command output:\t{fst_str}")

    log(f"Created container {splitted[1]}", endpoint)

    return splitted[1]


def upload_object(container, payload_filepath, endpoint, wallet_file, wallet_config):
    object_name = ""
    if wallet_file:
        wallet_file = "--wallet " + wallet_file
    if wallet_config:
        wallet_config = "--config " + wallet_config
    cmd_line = f"frostfs-cli --rpc-endpoint {endpoint} object put --file {payload_filepath} {wallet_file} {wallet_config} " \
               f"--cid {container} --no-progress"
    output, success = execute_cmd(cmd_line)

    if not success:
        log(f"{cmd_line}\n"
            f"Object {object_name} has not been uploaded\n"
            f"Error: {output}", endpoint)
        return False

    try:
        # taking second string from command output
        snd_str = output.split('\n')[1]
    except Exception:
        log(f"{cmd_line}\n"
            f"Incorrect output\n"
            f"Output: {output or '<empty>'}", endpoint)
        return False
    splitted = snd_str.split(": ")
    if len(splitted) != 2:
        raise Exception(f"no OID was parsed from command output: \t{snd_str}")
    return container, endpoint, splitted[1]


def get_object(cid, oid, endpoint, out_filepath, wallet_file, wallet_config):
    if wallet_file:
        wallet_file = "--wallet " + wallet_file
    if wallet_config:
        wallet_config = "--config " + wallet_config
    cmd_line = f"frostfs-cli object get -r {endpoint} --cid {cid} --oid {oid} {wallet_file} {wallet_config} " \
               f"--file {out_filepath}"

    output, success = execute_cmd(cmd_line)

    if not success:
        log(f"{cmd_line}\n"
            f"Failed to get object {oid} from container {cid}\n"
            f"Error: {output}", endpoint)
        return False

    return True


def search_object_by_id(cid, oid, endpoint, wallet_file, wallet_config, ttl=2):
    if wallet_file:
        wallet_file = "--wallet " + wallet_file
    if wallet_config:
        wallet_config = "--config " + wallet_config
    cmd_line = f"frostfs-cli object search --ttl {ttl} -r {endpoint} --cid {cid} --oid {oid} {wallet_file} {wallet_config} "

    output, success = execute_cmd(cmd_line)

    if not success:
        log(f"{cmd_line}\n"
            f"Failed to search object {oid} for container {cid}\n"
            f"Error: {output}", endpoint)
        return False

    re_rst = re.search(r'Found (\d+) objects', output)

    if not re_rst:
        raise Exception("Failed to parse search results")

    return re_rst.group(1)