lf_test_generic.py

#!/usr/bin/env python3
"""
NAME: lf_test_generic.py

PURPOSE:
lf_test_generic.py will create stations and endpoints to generate traffic based on a command-line specified command type.

This script will create a variable number of stations to test generic endpoints. Multiple command types can be tested
including ping, speedtest, lfcurl, iperf, generic types. The test will check the last-result attribute for different things
depending on what test is being run. Ping will test for successful pings, speedtest will test for download
speed, upload speed, and ping time, generic will test for successful generic commands.

This script also *does not* use any other file except lanforge_api.py.

SETUP:
Make sure the generic tab is enabled in the GUI by going to the Port Manager, clicking the '+' tab, checking the 'generic' tab.

EXAMPLES:
    LFPING :
        ./lf_test_generic.py --mgr 192.168.102.211 --test_type ping --lf_user lanforge --lf_passwd lanforge --num_stations 3 --log_level debug
                    --ssid eero-mesh-lanforge --passwd lanforge --security wpa2 --radio wiphy1 --target www.google.com --test_duration
                    "4s" --create_report --report_file_path "/home/diptidhond/test_generic_1"
    LFCURL :
        ./lf_test_generic.py --mgr 192.168.102.211 --test_type lfcurl --lf_user lanforge --lf_passwd lanforge --num_stations 3 --log_level debug
                    --ssid eero-mesh-lanforge --passwd lanforge --security wpa2 --radio wiphy1  --test_duration 4s --create_report
                    --report_file_path "/home/lanforge/test_generic_1"
    SPEEDTEST :
        ./lf_test_generic.py --mgr 192.168.102.211 --test_type speedtest --lf_user lanforge --lf_passwd lanforge --num_stations 3 --log_level debug
                    --ssid mesh-lanforge --passwd lanforge --security wpa2 --radio wiphy1  --test_duration 2m --create_report --no_upload --single_connection
                    --report_file_path "/home/lanforge/test_generic_1"

    IPERF3 :
        iperf: create 1 client and 1 server. client is already existing & server is on lanforge.

            ./lf_test_generic.py --mgr 192.168.102.211 --port 8080 --use_existing_eid "1.1.sta00015,1.1.eth3" --test_type iperf3
                --server_port 5201 --client_port 5201 --target "1.1.eth3" --create_report --port_mgr_cols "alias,rx bytes,tx bytes"


        iperf -- client only: creates 3 client : 2 clients on sta, 1 on eth port (with existing eid):

            ./lf_test_generic.py --mgr 192.168.102.211 --test_type iperf3-client --lf_user lanforge --lf_passwd lanforge --num_stations 2 --log_level debug --test_duration 20s
                --ssid mesh-lanforge --passwd lanforge --security wpa2 --radio wiphy1 --target 192.168.3.3 --use_existing_eid 1.1.eth2 --client_port 9191 --server_port 9191
                --create_report --report_file_path "/home/lanforge/iperf3_reports" --port_mgr_cols "alias,rx bytes,tx bytes"

        iperf -- server only: creates 3 servers : 2 servers on sta, 1 on eth port (with existing eid):

            ./lf_test_generic.py --mgr 192.168.102.211 --test_type iperf3-server --lf_user lanforge --lf_passwd lanforge --num_stations 2 --log_level debug --test_duration 20s
                --ssid mesh-lanforge --passwd lanforge --security wpa2 --radio wiphy1 --target --use_existing_eid 1.1.eth2 --client_port 9191 --server_port 9191 --create_report
                --report_file_path "/home/lanforge/iperf3_reports" --port_mgr_cols "alias,rx bytes,tx bytes"


Use './test_generic.py --help' to see command line usage and options.

Copyright (C) 2020-2026 Candela Technologies Inc
License: Free to distribute and modify. LANforge systems must be licensed.
"""
import sys
import os
import importlib
import argparse
import time
import datetime
import logging
import re
import csv
import pandas as pd

# from pandas import json_normalize
# from lf_json_util import standardize_json_results


sys.path.append(os.path.join(os.path.abspath(__file__ + "../../../")))
lanforge_api = importlib.import_module("lanforge_client.lanforge_api")
from lanforge_client.lanforge_api import LFSession  # noqa: E402
from lanforge_client.lanforge_api import LFJsonCommand  # noqa: E402
from lanforge_client.lanforge_api import LFJsonQuery  # noqa: E402
from lanforge_client.logg import Logg  # noqa: E402

# stand-alone (not dependent on realm)
lf_logger_config = importlib.import_module("py-scripts.lf_logger_config")
lf_report = importlib.import_module("py-scripts.lf_report")
lf_graph = importlib.import_module("py-scripts.lf_graph")
lf_kpi_csv = importlib.import_module("py-scripts.lf_kpi_csv")

# from lf_graph import lf_bar_graph_horizontal
# from lf_graph import lf_bar_graph  # noqa: E402
from lf_report import lf_report  # noqa: E402

lf_bar_graph = lf_graph.lf_bar_graph
lf_scatter_graph = lf_graph.lf_scatter_graph
lf_stacked_graph = lf_graph.lf_stacked_graph
lf_horizontal_stacked_graph = lf_graph.lf_horizontal_stacked_graph

logger = logging.getLogger(__name__)


if sys.version_info[0] != 3:
    logger.critical("This script requires Python 3")
    exit(1)


class GenTest():
    def __init__(self, lf_user, lf_passwd, ssid, security, passwd,
                 udp, receive,
                 name_prefix, num_stations, port_mgr_cols, gen_tab_cols, report_file_path, output_format="csv", client_port=None, server_port=None,
                 host="localhost", port=8080, create_report=False, use_existing_eid=None, monitor_interval="2s",
                 test_duration="20s", test_type="ping", target=None, cmd=None, spdtest_no_download=False, spdtest_no_upload=False, spdtest_single_connection=False,
                 spdtest_enable_debug=False, spdtest_enable_report=False, spdtest_ookla=False, interval=0.1, destination_url_lfcurl=None,
                 radio=None, file_output_lfcurl=None, lf_logger_json=None, log_level="debug", loop_count=None, port_wait_time=60,
                 _debug_on=False, _exit_on_error=False, die_on_error=False, _exit_on_fail=False, endp_names=None):
        self.port_wait_time = port_wait_time
        self.host = host
        self.port = port
        self.lf_user = lf_user
        self.lf_passwd = lf_passwd
        self.ssid = ssid
        self.radio = radio
        self.num_stations = num_stations
        self.security = security
        self.passwd = passwd
        self.name_prefix = name_prefix
        self.endp_names = endp_names

        self.udp = udp
        self.receive = receive

        # Generic test args
        self.cmd = cmd
        self.test_type = test_type
        self.file_output_lfcurl = file_output_lfcurl
        self.destination_url_lfcurl = destination_url_lfcurl

        # speedtest specific
        self.spdtest_no_download = spdtest_no_download
        self.spdtest_no_upload = spdtest_no_upload
        self.spdtest_single_connection = spdtest_single_connection
        self.spdtest_enable_debug = spdtest_enable_debug
        self.spdtest_enable_report = spdtest_enable_report
        self.spdtest_ookla = spdtest_ookla

        self.loop_count = loop_count
        self.target = target
        self.interval = interval
        self.client_port = client_port
        self.server_port = server_port

        # Test duration, sleep interval for monitoring/data collection, logging/debugging
        self.monitor_interval = monitor_interval
        self.lfclient_url = "http://%s:%s" % (self.host, self.port)
        self.test_duration = test_duration
        self.debug = _debug_on
        self.report_timer = 1500
        self.log_level = log_level
        self.lf_logger_json = lf_logger_json

        # TODO currently un-used args. Add to test below.
        self.exit_on_error = _exit_on_error
        self.exit_on_fail = _exit_on_fail
        self.die_on_error = die_on_error

        # reporting
        self.output_format = output_format
        self.report_file_path = report_file_path
        self.create_report = create_report

        # create a session
        self.session = LFSession(lfclient_url="http://%s:%s" % (self.host, self.port),
                                 debug=_debug_on,
                                 connection_timeout_sec=4.0,
                                 stream_errors=True,
                                 stream_warnings=True,
                                 require_session=True,
                                 exit_on_error=self.exit_on_error)
        if _debug_on:
            Logg.register_method_name("json_post")
            Logg.register_method_name("json_get")
            Logg.register_method_name("get_as_json")

        # type hinting
        self.command: LFJsonCommand
        self.command = self.session.get_command()
        self.query: LFJsonQuery
        self.query = self.session.get_query()

        self.created_endp = []
        self.created_cx = []

        # CSV data collection, add files and writers
        self.port_csv_files = {}
        self.port_csv_writers = {}

        self.gen_csv_files = {}
        self.gen_csv_writers = {}

        if self.create_report:
            # TODO add to results.csv as a 'combined' results.
            # split self.port_mgr_cols into list
            self.port_mgr_cols = str.split(port_mgr_cols, ",")
            # split self.gen_tab_cols into list
            self.gen_tab_cols = str.split(gen_tab_cols, ",")
            # create csv results writer and open results file to be written to
            results = self.report_file_path + "/results." + self.output_format
            self.csv_results_file = open(results, "w")
            self.csv_results_writer = csv.writer(self.csv_results_file, delimiter=",")

        number_template = "000"
        self.sta_list = []
        if self.radio and self.num_stations and int(self.num_stations) > 0:
            self.sta_list = self.port_name_series(prefix="sta",
                                                  start_id=int(number_template),
                                                  end_id=int(self.num_stations) + int(number_template) - 1,
                                                  padding_number=10000,
                                                  radio=self.radio)
        if (use_existing_eid):
            # split list from user to create functional list used in script.
            self.use_existing_eid = use_existing_eid.split(",")
        else:
            self.use_existing_eid = use_existing_eid

        # evaluate if iperf3-server on lanforge
        self.iperf3_target_lanforge = False
        if self.test_type in ["iperf3", "iperf3-client", "iperf3-server", "iperf"]:
            if self.target:
                if re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$', self.target) is None:
                    # not an ip address. obvious URLs are remote targets.
                    if any(self.target.startswith(p) for p in ("http://", "https://", "www.")):
                        self.iperf3_target_lanforge = False
                    else:
                        # confirm the target actually exists as a LANforge port before treating it as one,
                        # instead of guessing off the hostname string (avoids long hangs on remote hostnames)
                        try:
                            t_shelf, t_resource, t_port_name, *nil = self.name_to_eid(self.target)
                            # a 404 here just means the target isn't a lanforge port; silence the
                            # api client's error-level diagnostics for this expected outcome
                            prev_log_level = lanforge_api.LOGGER.level
                            lanforge_api.LOGGER.setLevel(logging.CRITICAL)
                            try:
                                port_resp = self.query.json_get(url="%s/ports/%s/%s/%s?fields=device" % (self.lfclient_url, t_shelf, t_resource, t_port_name), debug=self.debug)
                            finally:
                                lanforge_api.LOGGER.setLevel(prev_log_level)
                            self.iperf3_target_lanforge = port_resp is not None and ('interface' in port_resp or 'interfaces' in port_resp)
                        except Exception as err:
                            logger.debug("Exception while checking iperf3 target port '%s': %s", self.target, err)
                            self.iperf3_target_lanforge = False

    def check_tab_exists(self):
        json_response = self.command.json_get(self.lfclient_url + "/generic/")
        if json_response is None:
            return False
        return True

    # Write initial headers to port csv file (all ports used, created & existing ports)
    def csv_add_port_column_headers(self, port_eid):
        # create file name
        file_name = self.report_file_path + "/port-" + port_eid + "." + self.output_format

        # open file in write mode, returns TextIOWrapper
        txt_strm = open(file_name, "w")

        # create writer
        port_csv_writer = csv.writer(txt_strm, delimiter=",")

        # save writer and TextIOWrapper
        self.port_csv_files[port_eid] = txt_strm
        self.port_csv_writers[port_eid] = port_csv_writer

        # write headers
        port_csv_writer.writerow(self.port_mgr_cols)

        # flush stream
        txt_strm.flush()

    # write initial headers to gen cx files
    def csv_add_gen_column_headers(self, gen_endp):
        # create file name
        file_name = self.report_file_path + "/gen-endp-" + gen_endp + "." + self.output_format

        # open file in write mode, returns TextIOWrapper
        txt_strm = open(file_name, "w")

        # create writer
        gen_csv_writer = csv.writer(txt_strm, delimiter=",")

        # save writer and TextIOWrapper
        self.gen_csv_files[gen_endp] = txt_strm
        self.gen_csv_writers[gen_endp] = gen_csv_writer

        # write headers
        gen_csv_writer.writerow(self.gen_tab_cols)

        # flush stream
        txt_strm.flush()

    def write_port_csv(self, eid):
        port_shelf, port_resource, port_name, *nil = self.name_to_eid(eid)
        # get writer
        port_csv_writer = self.port_csv_writers[eid]
        port_csv_file = self.port_csv_files[eid]

        # write fields to string, to add to json_get command
        fields_str = "fields="
        for cols in self.port_mgr_cols[:-1]:
            fields_str = fields_str + "" + cols + ","
        fields_str = fields_str + self.port_mgr_cols[-1]  # add last field

        # fetch data w/json
        json_url = "%s/ports/%s/%s/%s?%s" % (self.lfclient_url, port_shelf, port_resource, port_name, fields_str)
        json_response = self.query.json_get(url=json_url,
                                            debug=self.debug)

        # append fetched data to row, to add to csv file
        row = []
        if (json_response is not None):
            json_re_intf = json_response['interface']
            for field in self.port_mgr_cols:
                row.append(json_re_intf[field])

        # post data to csv file on disk
        port_csv_writer.writerow(row)
        port_csv_file.flush()

    def write_gen_csv(self, endp):
        # get writer
        gen_csv_writer = self.gen_csv_writers[endp]
        gen_csv_file = self.gen_csv_files[endp]

        # write fields to string, to add to json_get command
        fields_str = "fields="
        for cols in self.gen_tab_cols[:-1]:
            fields_str = fields_str + "" + cols + ","
        fields_str = fields_str + self.gen_tab_cols[-1]  # add last field
        # fetch data w/json
        json_url = "%s/generic/%s?%s" % (self.lfclient_url, endp, fields_str)
        json_response = self.query.json_get(url=json_url,
                                            debug=self.debug)

        # append fetched data to row, to add to csv file
        row = []
        if (json_response is not None):
            json_re_intf = json_response['endpoint']
            for field in self.gen_tab_cols:
                row.append(json_re_intf[field])

        # post data to csv file on disk
        gen_csv_writer.writerow(row)
        gen_csv_file.flush()

    # TODO: This is an example and is only configured for ping currently.
    # This can be edited and added to if the user wants reporting and the test they are running is not ping.
    def generate_report(self, result_dir='Generic_Test_Report', report_path=''):
        print('Generating Report for lf_test_generic...')
        report = lf_report(_output_pdf='lf_test_generic.pdf',
                           _output_html='lf_test_generic.html',
                           _results_dir_name='lf_test_generic_report',
                           _path=self.report_file_path)
        report_path = report.get_path()
        report_path_date_time = report.get_path_date_time()
        print('path: {}'.format(report_path))
        print('path_date_time: {}'.format(report_path_date_time))

        # setting report title
        report.set_title('Generic Test Report: Ping')
        report.build_banner()

        total_devices = 0
        if (self.sta_list):
            total_devices += len(self.sta_list)
        if (self.use_existing_eid):
            total_devices += len(self.use_existing_eid)

        # test setup info
        test_setup_info = {
            'SSID': self.ssid,
            'Security': self.security,
            'Website / IP': self.target,
            'Number of clients (virtual and real)': total_devices,
            'Duration': self.test_duration
        }

        report.test_setup_table(test_setup_data=test_setup_info, value='Test Setup Information')

        # objective and description
        report.set_obj_html(_obj_title='Objective',
                            _obj='''The objective of the Generic Ping Test is to evaluate network connectivity and measure the round-trip time taken for
                            data packets to travel from the source to the destination and back. This test helps assess the reliability and latency of the network,
                            packet loss, packet delays, or variations in response times to ensure that devices can communicate
                            effectively over the network and pinpoint potential issues affecting connectivity.
                            ''')
        report.build_objective()

        # TODO make graph creation customizable via command line
        # get data from saved csv.
        gen_endp_names = []
        dataset = []
        for endp_name in self.created_endp:
            gen_endp_names.append(endp_name)
            io_file = self.gen_csv_files[endp_name]
            # get csv and convert to pandas, take last row of pandas.
            # csv_df = pd.read_csv("/home/diptidhond/test_generic_1/gen-endp-ping-3.csv")
            csv_df = pd.read_csv(io_file.name)
            # edit df with columns & rows we want
            # take last row
            last_row_df = csv_df.tail(1)
            # take out all other un-necessary columns if needed
            # last_row = last_row_df[["tx pkts", "rx pkts", "dropped"]]

            # convert dataframe to 1 dimensional array, to reg list, take idx 0, append to dataset array
            array = last_row_df.to_numpy().tolist()[0]
            array_but_first = array[1:]
            dataset.append(array_but_first)

        graph = lf_bar_graph(_data_set=dataset,
                             _xaxis_name="Generic Cross-Connects",
                             _yaxis_name="Pkt Count",
                             _xaxis_categories=gen_endp_names,
                             _graph_image_name="Rx vs Tx Vs Dropped",
                             _label=["rx pkts", "tx pkts", "dropped pkts"],
                             _color=['darkorange', 'forestgreen', 'blueviolet'],
                             _color_edge='red',
                             _grp_title="Rx Pkts vs Tx Pkts vs Dropped Pkts",
                             _xaxis_step=1,
                             _show_bar_value=True,
                             _text_font=7,
                             _text_rotation=45,
                             _xticks_font=7,
                             _legend_loc="best",
                             _legend_box=(1, 1),
                             _legend_ncol=1,
                             _legend_fontsize=None,
                             _enable_csv=True)

        graph_png = graph.build_bar_graph()

        print("graph name {}".format(graph_png))

        report.set_graph_image(graph_png)
        report.move_graph_image()
        report.build_graph()

        # close report, add final touches
        report.build_custom()
        report.build_footer()
        report.write_html()
        report.write_pdf()

    def monitor_test(self):
        logger.info("Starting monitoring")
        # TODO: add checking if stations have disconnected, try to reconnect
        # TODO: add checking if cross-connects have stopped. figure out why (if due to test being done, or if just randomly stopped.)
        # TODO: add reporting-- save data at same intervals as sleeping?
        monitor_interval = self.duration_time_to_seconds(self.monitor_interval)
        end_time = datetime.datetime.now().timestamp() + self.duration_time_to_seconds(self.test_duration)

        while (datetime.datetime.now().timestamp() < end_time):
            # write to all csv files
            if (self.create_report):
                if (self.sta_list and self.port_mgr_cols):
                    for sta_alias in self.sta_list:
                        self.write_port_csv(sta_alias)
                if (self.use_existing_eid and self.port_mgr_cols):
                    for eid in self.use_existing_eid:
                        self.write_port_csv(eid)
                if (self.created_endp and self.gen_tab_cols):
                    for endp in self.created_endp:
                        self.write_gen_csv(endp)
            time.sleep(monitor_interval)

    def start(self):
        # admin up all created stations & existing stations
        interest_flags_list = ['ifdown']
        set_port_interest_rslt = self.command.set_flags(LFJsonCommand.SetPortInterest, starting_value=0, flag_names=interest_flags_list)
        if self.sta_list:
            for sta_alias in self.sta_list:
                port_shelf, port_resource, port_name, *nil = self.name_to_eid(sta_alias)
                # write headers to csv
                if (self.create_report):
                    self.csv_add_port_column_headers(sta_alias)
                self.command.post_set_port(shelf=port_shelf,
                                           resource=port_resource,
                                           port=port_name,
                                           netmask="255.255.255.0",  # sometimes the cli complains about the netmask being NA, so set it to a random netmask (netmask is overriden anyways with dhcp)
                                           current_flags=0,
                                           interest=set_port_interest_rslt,
                                           report_timer=self.report_timer)

        if self.use_existing_eid:
            for eid in self.use_existing_eid:
                port_shelf, port_resource, port_name, *nil = self.name_to_eid(eid)
                # write headers to csv
                if (self.create_report):
                    self.csv_add_port_column_headers(eid)
                self.command.post_set_port(shelf=port_shelf,
                                           resource=port_resource,
                                           port=port_name,
                                           netmask="255.255.255.0",  # sometimes the cli complains about the netmask being NA, so set it to a random netmask(netmask is overriden anyways with dhcp)
                                           current_flags=0,
                                           interest=set_port_interest_rslt,
                                           report_timer=self.report_timer)

        # Check if created stations admin-up and have ips
        if (self.sta_list):
            if self.wait_for_action("port", self.sta_list, "up", self.port_wait_time):
                print("All created stations went admin up.")
            else:
                print("All creation stations  did not admin up.")

            if self.wait_for_action("port", self.sta_list, "ip", self.port_wait_time):
                logger.info("All creation stations got IPs")
            else:
                self.print("Stations failed to get IPs")

        # Check if existing ports admin-up and got ips
        if (self.use_existing_eid):
            if self.wait_for_action("port", self.use_existing_eid, "up", self.port_wait_time):
                print("All exiting ports went admin up.")
            else:
                print("All existing ports did not admin up.")

            if self.wait_for_action("port", self.use_existing_eid, "ip", self.port_wait_time):
                print("All existing ports got IPs")
            else:
                self.print("All existing ports failed to get IPs")

        # at this point, all endpoints have been created, start all endpoints
        if self.created_cx:
            for cx in self.created_cx:
                self.command.post_set_cx_state(cx_name=cx,
                                               cx_state="RUNNING",
                                               test_mgr="default_tm",
                                               debug=self.debug)

    def stop(self):
        logger.info("Stopping Test...")
        # set_cx_state default_tm CX_ping-hi STOPPED
        if self.created_cx:
            for cx in self.created_cx:
                self.command.post_set_cx_state(cx_name=cx,
                                               cx_state="STOPPED",
                                               test_mgr="default_tm",
                                               debug=self.debug)
        # admin stations down
        if self.sta_list:
            for sta_alias in self.sta_list:
                port_shelf, port_resource, port_name, *nil = self.name_to_eid(sta_alias)
                self.command.post_set_port(shelf=port_shelf,
                                           resource=port_resource,
                                           port=port_name,
                                           current_flags=1,  # vs 0x0 = interface up
                                           interest=8388610,  # = current_flags + ifdown
                                           report_timer=self.report_timer)
        # admin stations down
        if self.use_existing_eid:
            for eid in self.use_existing_eid:
                port_shelf, port_resource, port_name, *nil = self.name_to_eid(eid)
                self.command.post_set_port(shelf=port_shelf,
                                           resource=port_resource,
                                           port=port_name,
                                           current_flags=1,  # vs 0x0 = interface up
                                           interest=8388610,  # = current_flags + ifdown
                                           report_timer=self.report_timer)

    def build(self):
        # create stations
        if self.sta_list:
            logger.info("Creating stations")
            types = {"wep": "wep_enable", "wpa": "wpa_enable", "wpa2": "wpa2_enable", "wpa3": "use-wpa3", "open": "[BLANK]"}
            if self.security in types.keys():
                add_sta_flags = []
                set_port_interest = ['rpt_timer', 'current_flags', 'dhcp']
                set_port_current = ['use_dhcp']
                # add appropriate flags for security
                if self.security != "open":
                    add_sta_flags.extend([types[self.security], "create_admin_down"])

                # go through each test-created station and create station in lanforge
                for sta_alias in self.sta_list:
                    port_shelf, port_resource, port_name, *nil = self.name_to_eid(sta_alias)
                    sta_flags_rslt = self.command.set_flags(LFJsonCommand.AddStaFlags, starting_value=0, flag_names=add_sta_flags)
                    set_port_interest_rslt = self.command.set_flags(LFJsonCommand.SetPortInterest, starting_value=0, flag_names=set_port_interest)
                    set_port_current_rslt = self.command.set_flags(LFJsonCommand.SetPortCurrentFlags, starting_value=0, flag_names=set_port_current)

                    # Check if station already exists on LANforge manager (ignore stale/phantom ports)
                    sta_exists = False
                    try:
                        prev_log_level = lanforge_api.LOGGER.level
                        lanforge_api.LOGGER.setLevel(logging.CRITICAL)
                        try:
                            port_resp = self.query.json_get(url="%s/ports/%s/%s/%s?fields=device,phantom" % (self.lfclient_url, port_shelf, port_resource, port_name), debug=self.debug)
                        finally:
                            lanforge_api.LOGGER.setLevel(prev_log_level)
                        if port_resp is not None:
                            iface = port_resp.get('interface', port_resp.get('interfaces'))
                            if iface is not None and str(iface.get('phantom')).lower() not in ("true", "1"):
                                sta_exists = True
                    except Exception as err:
                        logger.debug("Exception while checking if station '%s' exists: %s", sta_alias, err)
                        sta_exists = False

                    if sta_exists:
                        logger.info("Station '%s' already exists on LANforge manager. Overwriting with new configuration", sta_alias)
                        sta_mac = "NA"
                    else:
                        sta_mac = "xx:xx:xx:*:*:xx"

                    if self.security == "wpa3":
                        self.command.post_add_sta(flags=sta_flags_rslt,
                                                  flags_mask=sta_flags_rslt,
                                                  radio=self.radio,
                                                  resource=port_resource,
                                                  shelf=port_shelf,
                                                  sta_name=port_name,
                                                  ieee80211w=2,
                                                  mode=0,
                                                  mac=sta_mac,
                                                  ssid=self.ssid,
                                                  key=self.passwd,
                                                  debug=self.debug)

                    else:
                        self.command.post_add_sta(flags=sta_flags_rslt,
                                                  flags_mask=sta_flags_rslt,
                                                  radio=self.radio,
                                                  resource=port_resource,
                                                  shelf=port_shelf,
                                                  mac=sta_mac,
                                                  key=self.passwd,
                                                  mode=0,
                                                  ssid=self.ssid,
                                                  sta_name=port_name,
                                                  debug=self.debug)
                    # tell lanforge to show ports
                    self.command.post_nc_show_ports(port=port_name,
                                                    resource=port_resource,
                                                    shelf=port_shelf)

                    # wait until port appears
                    wfa_list = [sta_alias]
                    self.wait_for_action("port", wfa_list, "appear", self.port_wait_time)

                    # set use_dhcp and rpt_timer
                    self.command.post_set_port(alias=port_name,
                                               port=port_name,
                                               shelf=port_shelf,
                                               interest=set_port_interest_rslt,
                                               current_flags=set_port_current_rslt,
                                               resource=port_resource,
                                               report_timer=self.report_timer)
                    # set port admin up
                    set_port_interest_rslt = self.command.set_flags(LFJsonCommand.SetPortInterest, starting_value=0, flag_names=['ifdown'])
                    self.command.post_set_port(shelf=port_shelf,
                                               resource=port_resource,
                                               port=port_name,
                                               current_flags=0,  # vs 0x0 = interface up
                                               interest=set_port_interest_rslt,
                                               report_timer=self.report_timer)
                    # wait for station to associate with AP
                    self.wait_for_action("port", wfa_list, "up", self.port_wait_time)
                    self.wait_for_action("port", wfa_list, "ip", self.port_wait_time)

            else:
                raise ValueError("security type given: %s : is invalid. Please set security type as wep, wpa, wpa2, wpa3, or open." % self.security)
        if (self.test_type == 'iperf3'):
            # admin up server port, we need IP for client generic endp creation.
            # This code is only executed if we are NOT given a target ip address. (e.g. given 1.1.eth2 instead of 192.168.101.3)
            if (self.iperf3_target_lanforge):
                server_shelf, server_resource, server_port_name, *nil = self.name_to_eid(self.target)
                if server_port_name:
                    logger.info("Setting server port up: %s.%s.%s", server_shelf, server_resource, server_port_name)
                    set_port_interest_rslt = self.command.set_flags(LFJsonCommand.SetPortInterest, starting_value=0, flag_names=['ifdown'])
                    self.command.post_set_port(shelf=server_shelf,
                                               resource=server_resource,
                                               port=server_port_name,
                                               current_flags=0,
                                               interest=set_port_interest_rslt,
                                               report_timer=self.report_timer)
                    self.wait_for_action("port", [self.target], "up", self.port_wait_time)
                    self.wait_for_action("port", [self.target], "ip", self.port_wait_time)

        # create endpoints
        # create 1 endp for each eid.
        unique_alias = 0
        if self.sta_list:
            unique_alias += len(self.sta_list)
        if self.use_existing_eid:
            unique_alias += len(self.use_existing_eid)

        endp_idx = 0
        # these cannot be interop clients
        if self.sta_list:
            for sta_alias in self.sta_list:
                sta_eid = self.name_to_eid(sta_alias)
                custom_name = self.endp_names[endp_idx] if (self.endp_names and endp_idx < len(self.endp_names)) else None
                self.create_generic_endp(sta_eid, self.test_type, unique_alias, interop_device=False, custom_alias=custom_name)
                unique_alias -= 1
                endp_idx += 1

        # check if these are interop devices..
        if self.use_existing_eid:  # use existing eid will have iperf3-server eid, if we are using lanforge iperf3-server.
            for eid_alias in self.use_existing_eid:
                eid_list = self.name_to_eid(eid_alias)
                is_interop = self.is_device_interop(eid_list)
                custom_name = self.endp_names[endp_idx] if (self.endp_names and endp_idx < len(self.endp_names)) else None
                self.create_generic_endp(eid_list, self.test_type, unique_alias, interop_device=is_interop, custom_alias=custom_name)
                unique_alias -= 1
                endp_idx += 1

        # show all endps
        self.command.post_nc_show_endpoints(endpoint='all', extra='history')

        # create cross-connects
        if self.created_endp is not None:
            for endp in self.created_endp:
                endp_cx_name = "CX_" + endp
                self.command.post_add_cx(alias=endp_cx_name,
                                         test_mgr="default_tm",
                                         rx_endp="D_" + endp,
                                         tx_endp=endp,
                                         debug=self.debug)
                self.created_cx.append(endp_cx_name)

        # wait for cross-connects to appear
        cx_wait_time = max(10, self.port_wait_time) if self.port_wait_time > 0 else 10
        if self.wait_for_action("cx", self.created_endp, "appear", cx_wait_time):
            logger.info("Generic cx creation completed")
        else:
            logger.error("Generic cx creation was not completed")

    # This function takes an eid list and returns an ip address.
    def eid_to_ip(self, eid_list):
        device_shelf, device_resource, device_port_name, *nil = eid_list
        json_url = "%s/ports/%s/%s/%s?fields=device,ip" % (self.lfclient_url, device_shelf, device_resource, device_port_name)
        json_response = self.query.json_get(url=json_url,
                                            debug=self.debug)
        if json_response is not None and (json_response['interface']['ip'] != "0.0.0.0"):
            return json_response['interface']['ip']
        else:
            return None

    # This function takes eid_list and returns interop device type or False.
    def is_device_interop(self, eid_list):
        return False
        # device_shelf, device_resource, device_port_name, *nil = eid_list
        # json_url = "%s/ports/%s/%s/%s?fields=hw version" % (self.lfclient_url, device_shelf, device_resource, device_port_name)
        # json_response = self.query.json_get(url=json_url,
        # debug=self.debug)
        # if json_response is not None:
        # return json_response['hw version']
        # else:
        # return False

    def create_generic_endp(self, eid_list, type, unique_num, interop_device, custom_alias=None):
        """
        :param eid: list format of eid. example: ['1', '1', 'sta0010']
        :param type: takes type of generic test: 'ping', 'iperf3-client', 'iperf3-server', 'lfcurl', 'speedtest', 'iperf'
        :param unique_num: takes in unique ending prefix. example: 3
        :param interop device: is this station interop?
        :param custom_alias: optional custom endpoint name
        :return: no return
        """
        if custom_alias:
            unique_alias = str(custom_alias).strip()
        else:
            unique_alias = type + "-" + str(unique_num)
        # construct generic endp command

        cmd = ""
        eid_shelf = eid_list[0]
        eid_resource = eid_list[1]
        eid_name = eid_list[2]
        if (self.cmd):
            cmd = self.cmd
        elif (type == 'iperf3'):
            # lanforge hosts the iperf3 server only when the target is a lanforge port and this endpoint is that port;
            # every other endpoint (and every endpoint when the target is remote) is an iperf3 client.
            is_server = self.iperf3_target_lanforge and (eid_list == self.name_to_eid(self.target))
            if not custom_alias:
                unique_alias = ("server-" if is_server else "client-") + unique_alias
            cmd = self.do_iperf('server' if is_server else 'client', unique_alias, eid_list)

        elif (type == 'ping'):
            standard_ping = True
            if (interop_device):
                standard_ping = False
                os_type = interop_device
                # construct ping command based on interop device type
                if 'Win/x86' in os_type:
                    cmd = "py lfping.py -S %s -n -1 -dest_ip %s" % (self.eid_to_ip(eid_name), self.target)
                elif 'Apple/x86' in os_type:
                    cmd = "lfping -b %s -i %s %s" % (eid_name, self.interval, self.target)
                elif 'Linux/x86' in os_type:
                    standard_ping = True
                else:
                    # Android
                    cmd = "ping -i %s %s" % (self.interval, self.target)

            if (standard_ping):
                # lfping  -s 128 -i 0.1 -c 10000 -I sta0000 www.google.com
                cmd = "lfping"
                if (self.interval):
                    cmd = cmd + " -i %s" % self.interval
                if (self.loop_count):
                    cmd = cmd + " -c %d" % self.loop_count
                cmd = cmd + " -I %s " % eid_name
                if (self.target):
                    cmd = cmd + str(self.target)

        elif (type == 'iperf3-client'):
            #  iperf3 --forceflush --format k --precision 4 -c 192.168.10.1 -t 60 --tos 0 -b 1K --bind_dev sta0000
            # -i 5 --pidfile /tmp/lf_helper_iperf3_testing.pid -p 101
            cmd = self.do_iperf('client', unique_alias, eid_list)

        elif (self.test_type == 'iperf3-server'):
            # iperf3 --forceflush --format k --precision 4 -s --bind_dev sta0000
            # -i 5 --pidfile /tmp/lf_helper_iperf3_testing.pid -p 101
            cmd = self.do_iperf('server', unique_alias, eid_list)

        elif (self.test_type == 'lfcurl'):
            # ./scripts/lf_curl.sh  -p sta0000 -i 192.168.50.167 -o /dev/null -n 1 -d 8.8.8.8
            cmd = cmd + str("./scripts/lf_curl.sh -p %s" % eid_name)
            # TODO: get ip address of -i (sta0000) if i is a station, but not if eth port.
            if (self.file_output_lfcurl):
                cmd = cmd + " -o %s" % self.file_output_lfcurl
            if (self.loop_count):
                cmd = cmd + " -n %i" % self.loop_count
            if (self.destination_url_lfcurl):
                cmd = cmd + " -d %s" % self.destination_url_lfcurl
        elif (self.test_type == 'speedtest'):
            if (self.spdtest_ookla):
                cmd = "speedtest --interface=%s --format=csv" % eid_name
            else:
                # do lanforge speedtest
                # vrf_exec.bash eth3 speedtest-cli --csv --share --single --debug
                # vrf_exec.bash eth3 speedtest-cli --csv --no-upload
                # vrf_exec.bash eth3 speedtest-cli --csv
                cmd = cmd + "vrf_exec.bash %s speedtest-cli --csv" % eid_name  # bas command
                if (self.spdtest_enable_report):
                    cmd = cmd + " --share"
                if (self.spdtest_no_download):
                    cmd = cmd + " --no_download"
                if (self.spdtest_no_upload):
                    cmd = cmd + " --no_upload"
                if (self.spdtest_single_connection):
                    cmd = cmd + " --single"
                if (self.spdtest_enable_debug):
                    cmd = cmd + " --debug"
        else:
            raise ValueError("Was not able to identify type given in arguments.")

        # create initial generic endp
        #  add_gen_endp testing 1 1 sta0000 gen_generic
        self.command.post_add_gen_endp(alias=unique_alias,
                                       shelf=eid_shelf,
                                       resource=eid_resource,
                                       port=eid_name,
                                       p_type="gen_generic")

        self.created_endp.append(unique_alias)

        # did generic endp appear in port manager?
        endp_wait_time = max(10, self.port_wait_time) if self.port_wait_time > 0 else 10
        if self.wait_for_action("endp", self.created_endp, "appear", endp_wait_time):
            logger.info("Generic endpoint creation completed")
        else:
            logger.error("Generic endpoints were not created")

        logger.info("This is the generic cmd we are sending to server...:   " + cmd)
        self.command.post_set_gen_cmd(name=unique_alias,
                                      command=cmd)

        # add headers to endp file if user asks to create report
        if (self.create_report):
            self.csv_add_gen_column_headers(unique_alias)

    def do_iperf(self, type, alias, eid_list):
        """
        :param type: takes in options 'client' or 'server'
        :param alias: takes in alias. example: 'sta0000'
        :param eid: takes in eid list. ['1','1','sta0000']
        :return: returns constructed iperf command
        """
        # TODO: allow for multiple ports to be passed in for multiple servers on 1 eth port or 1 virt sta (for example).
        # TODO: allow for multiple targets to be passed in for multiple servers.
        cmd = "iperf3 --forceflush --format k --precision 4"
        # TODO check if dest, client_port and server_port are not empty
        # eid_shelf = eid_list[0]
        # eid_resource = eid_list[1]
        eid_name = eid_list[2]
        if (type == 'client'):
            if (self.iperf3_target_lanforge):
                server_ip_addr = self.eid_to_ip(self.name_to_eid(self.target))
            else:
                server_ip_addr = self.target
            cmd = cmd + str(" -c %s" % server_ip_addr)
            if self.udp:
                cmd = cmd + " -u "
            if self.receive:
                cmd = cmd + " -R "
            cmd = cmd + " -t 60 --tos 0 -b 1K " + str("--bind_dev %s" % eid_name)
            cmd = cmd + " -i 3 --pidfile /tmp/lf_helper_iperf3_%s.pid" % alias
            if (self.client_port):
                # add port that should match server_port
                cmd = cmd + " -p %s" % self.client_port
        # server
        else:
            # iperf3 --forceflush --format k --precision 4 -s --bind_dev eth2 -i 5 --pidfile /tmp/lf_helper_iperf3_server_iperf_1.pid eth2
            cmd = cmd + " -s" + str(" --bind_dev %s" % eid_name) + " -i 3 --pidfile /tmp/lf_helper_iperf3_%s.pid" % alias
            if (self.server_port):
                # add port that should match client port
                cmd = cmd + " -p %s" % self.server_port
        return cmd

    def cleanup(self):
        # delete all created endps, cross-connects and created stations
        logger.info("Cleaning up all cxs, endpoints, and created stations")
        if self.created_cx:
            for cx_name in self.created_cx:
                self.command.post_rm_cx(cx_name=cx_name, test_mgr="default_tm", debug=self.debug)
        if self.created_endp:
            for endp_name in self.created_endp:
                self.command.post_rm_endp(endp_name=endp_name, debug=self.debug)
        if self.sta_list:
            for sta_name in self.sta_list:
                if self.port_exists(sta_name, self.debug):
                    port_shelf, port_resource, port_name, *nil = self.name_to_eid(sta_name)
                    self.command.post_rm_vlan(port=port_name,
                                              resource=port_resource,
                                              shelf=port_shelf,
                                              debug=self.debug)

        if self.wait_for_action("port", self.sta_list, "disappear", self.port_wait_time):
            print("Ports successfully cleaned up.")
        else:
            print("Ports were not successfully cleaned up.")

    def duration_time_to_seconds(self, time_string):
        # this function is used to convert self.test_duration to seconds
        if isinstance(time_string, str):
            pattern = re.compile("^(\\d+)([dhms]$)")
            td = pattern.match(time_string)
            if td:
                dur_time = int(td.group(1))
                dur_measure = str(td.group(2))
                if dur_measure == "d":
                    duration_sec = dur_time * 24 * 60 * 60
                elif dur_measure == "h":
                    duration_sec = dur_time * 60 * 60
                elif dur_measure == "m":
                    duration_sec = dur_time * 60
                else:
                    duration_sec = dur_time * 1
            else:
                raise ValueError("Unknown value for time_string: %s" % time_string)
        else:
            raise ValueError("time_string must be of type str. Type %s provided" % type(time_string))
        return duration_sec

    def port_name_series(self, prefix="sta", start_id=0, end_id=1, padding_number=10000, radio=None):
        """
        This produces a named series similar to "sta000, sta001, sta002...sta0(end_id)"
        the padding_number is added to the start and end numbers and the resulting sum
        has the first digit trimmed, so f(0, 1, 10000) => {"0000", "0001"}
        :param radio:
        :param prefix: defaults to 'sta'
        :param start_id: beginning id
        :param end_id: ending_id
        :param padding_number: used for width of resulting station number
        :return: list of stations
        """
        name_list = []
        if radio is not None:
            radio_shelf, radio_resource, sta_name, *nil = self.name_to_eid(radio)
            for i in range((padding_number + start_id), (padding_number + end_id + 1)):
                sta_name = "%s%s" % (prefix, str(i)[1:])
                name_list.append("%i.%i.%s" % (radio_shelf, radio_resource, sta_name))
        return name_list

    # This function takes in eid '1.1.sta0000' and returns a boolean (true or false)
    def port_exists(self, port_eid, debug=None):
        if port_eid:
            current_stations = self.query.get_port(port_eid, debug=debug)
        if current_stations:
            return True
        return False

    def wait_for_action(self, lf_type, type_list, action, secs_to_wait):
        """
        :param lf_type: type of object given in object list.
        :param type_list: object list that needs to be iterated through to check for attributes.
                          E.g. : (station list, port list, cx list, etc.)
        :param action: what function is waiting for, example: IP, appear, disappear.
        :param secs_to_wait: secs to wait until "wait" fails (if <= 0, skips waiting)
        :return: no returns

        """
        if int(secs_to_wait) <= 0:
            return
        if type(type_list) is not list:
            raise Exception("wait_for_action: type_list is not a list")
        else:
            compared_pass = len(type_list)
            for attempt in range(0, int(secs_to_wait / 2)):
                passed = set()

                # Port Manager Actions
                if lf_type == "port":
                    for sta_alias in type_list:
                        port_shelf, port_resource, port_name, *nil = self.name_to_eid(sta_alias)
                        if action == "appear":
                            # http://192.168.102.211:8080/ports/1/1/sta0000?fields=device,down,phantom
                            json_url = "%s/ports/%s/%s/%s?fields=device,down,phantom" % (self.lfclient_url, port_shelf, port_resource, port_name)
                            json_response = self.query.json_get(url=json_url,
                                                                debug=self.debug)
                            # if sta is found in port manager and is not phantom
                            if json_response is not None:
                                iface = json_response.get('interface', json_response.get('interfaces'))
                                if iface is not None and str(iface.get('phantom')).lower() not in ("true", "1"):
                                    passed.add("%s.%s.%s" % (port_shelf, port_resource, port_name))

                        elif action == "up":
                            json_url = "%s/ports/%s/%s/%s?fields=device,down" % (self.lfclient_url, port_shelf, port_resource, port_name)
                            json_response = self.query.json_get(url=json_url,
                                                                debug=self.debug)
                            # if sta is up
                            if json_response is not None and 'interface' in json_response:
                                if json_response['interface'].get('down') is False:
                                    passed.add("%s.%s.%s" % (port_shelf, port_resource, port_name))

                        elif action == "disappear":
                            json_url = "%s/ports/%s/%s/%s?fields=device" % (self.lfclient_url, port_shelf, port_resource, port_name)
                            json_response = self.query.json_get(url=json_url,
                                                                debug=self.debug)
                            # if device is not found
                            if json_response is None or ('interface' not in json_response and 'interfaces' not in json_response):
                                passed.add("%s.%s.%s" % (port_shelf, port_resource, port_name))
                        elif action == "ip":
                            json_url = "%s/ports/%s/%s/%s?fields=device,ip" % (self.lfclient_url, port_shelf, port_resource, port_name)
                            json_response = self.query.json_get(url=json_url,
                                                                debug=self.debug)
                            # if device has valid IP assigned
                            if json_response is not None and 'interface' in json_response:
                                ip_val = str(json_response['interface'].get('ip', '0.0.0.0')).strip()
                                if ip_val and ip_val != "0.0.0.0":
                                    passed.add("%s.%s.%s" % (port_shelf, port_resource, port_name))

                # Generic Tab Actions
                else:
                    compared_pass = len(self.created_endp)
                    if action == "appear":
                        for endp_name in type_list:
                            json_url = "%s/generic/%s" % (self.lfclient_url, endp_name)
                            json_response = self.query.json_get(url=json_url,
                                                                debug=self.debug)
                            if (lf_type == "endp"):
                                if json_response is not None:
                                    passed.add(endp_name)

                            # check if cross-connect appears
                            else:
                                if json_response is not None and (json_response['endpoint']['status'] != "NO-CX"):
                                    passed.add(endp_name)

                if len(passed) < compared_pass:
                    time.sleep(2)
                    logger.info('Found %s out of %s %ss in %s out of %s tries in wait_until_%s_%s' % (len(passed), compared_pass, lf_type, attempt, int(secs_to_wait / 2), lf_type, action))
                else:
                    logger.info('All %s %ss appeared' % (len(passed), lf_type))
                    return True
        return False

    def check_args(self):
        print("Checking args passed into test")
        # TODO validate all args, depending on which test is used.
        # TODO: in args, check if file_out_lfcurl and destination_url_lfcurl is None. then state that defaults are being used and apply defaults
        # Station creation specific
        if (self.sta_list) and self.security != "open":
            if (self.passwd is None) or (self.passwd == ""):
                raise ValueError("use_security: %s requires passphrase or [BLANK]" % self.security)
        # Generic endp specific
        if ((self.test_type == "iperf3" or self.test_type == "iperf3-client") and self.target is None):
            raise ValueError("To execute test type 'iperf3' or 'iperf3-client', a target must be specified as an IP address or port eid.")

    # This takes in a eid string (e.g. '1.1.sta000') and returns an eid list in list order of [shelf, resource, port name]
    def name_to_eid(self, eid_input, non_port=False):
        rv = [1, 1, "", ""]
        if (eid_input is None) or (eid_input == ""):
            logger.critical("name_to_eid wants eid like 1.1.sta0 but given[%s]" % eid_input)
            raise ValueError("name_to_eid wants eid like 1.1.sta0 but given[%s]" % eid_input)
        if type(eid_input) is not str:
            logger.critical(
                "name_to_eid wants string formatted like '1.2.name', not a tuple or list or [%s]" % type(eid_input))
            raise ValueError(
                "name_to_eid wants string formatted like '1.2.name', not a tuple or list or [%s]" % type(eid_input))

        info = eid_input.split('.')
        if len(info) == 1:
            rv[2] = info[0]  # just port name
            return rv

        if (len(info) == 2) and info[0].isnumeric() and not info[1].isnumeric():  # resource.port-name
            rv[1] = int(info[0])
            rv[2] = info[1]
            return rv

        elif (len(info) == 2) and not info[0].isnumeric():  # port-name.qvlan
            rv[2] = info[0] + "." + info[1]
            return rv

        if (len(info) == 3) and info[0].isnumeric() and info[1].isnumeric():  # shelf.resource.port-name
            rv[0] = int(info[0])
            rv[1] = int(info[1])
            rv[2] = info[2]
            return rv

        elif (len(info) == 3) and info[0].isnumeric() and not info[1].isnumeric():  # resource.port-name.qvlan
            rv[1] = int(info[0])
            rv[2] = info[1] + "." + info[2]
            return rv

        if non_port:
            # Maybe attenuator or similar shelf.card.atten.index
            rv[0] = int(info[0])
            rv[1] = int(info[1])
            rv[2] = int(info[2])
            if len(info) >= 4:
                rv[3] = int(info[3])
            return rv

        if len(info) == 4:  # shelf.resource.port-name.qvlan
            rv[0] = int(info[0])
            rv[1] = int(info[1])
            rv[2] = info[2] + "." + info[3]

        return rv


def handle_toolbox(args):
    """Execute standalone toolbox actions for LANforge generic test operations.

    Args:
        args: Parsed command line arguments instance.

    Returns:
        bool: True if all toolbox actions succeeded, otherwise False.
    """
    if not getattr(args, 'toolbox', False):
        return False

    logger.info("Toolbox mode enabled")
    host = getattr(args, 'mgr', None) or getattr(args, 'lfmgr', None) or "localhost"
    port = getattr(args, 'mgr_port', None) or getattr(args, 'lfmgr_port', None) or 8080
    lfclient_url = "http://%s:%s" % (host, port)

    def parse_list(raw_val):
        if not raw_val:
            return []
        if isinstance(raw_val, list):
            items = []
            for item in raw_val:
                cleaned = str(item).replace('[', '').replace(']', '').replace("'", '').replace('"', '').split(',')
                for sub in cleaned:
                    if sub.strip():
                        items.append(sub.strip())
            return items
        return [p.strip() for p in str(raw_val).replace('[', '').replace(']', '').replace("'", '').replace('"', '').split(',') if p.strip()]

    def resolve_cx_name(cx_name):
        """Resolve a user-supplied cx/endpoint name to the actual cx name on LANforge, checking
        both the 'CX_'-prefixed form build() always creates and the name as given. Returns None
        if neither form exists, so callers can skip and log cleanly instead of sending a doomed
        write request that LANforge answers with a raw 'Invalid argument' error."""
        candidates = [cx_name] if cx_name.startswith("CX_") else ["CX_" + cx_name, cx_name]
        for candidate in candidates:
            try:
                # a 404 here just means this candidate name doesn't exist; silence the api
                # client's error-level diagnostics for this expected outcome
                prev_log_level = lanforge_api.LOGGER.level
                lanforge_api.LOGGER.setLevel(logging.CRITICAL)
                try:
                    resp = query.get_cx(eid_list=[candidate])
                finally:
                    lanforge_api.LOGGER.setLevel(prev_log_level)
                if resp:
                    return candidate
            except Exception as err:
                logger.debug("Exception while probing CX candidate '%s': %s", candidate, err)
                continue
        return None

    def get_generic_endpoint_names():
        """Query LANforge for all generic-tab endpoint names currently on the manager.
        Scoped to the generic tab (/generic/all, not /cx/all), so 'all' actions touch all generic
        endpoints/CXs on the manager, never unrelated L3/other test-type cross-connections."""
        try:
            gen_data = query.json_get(url=f"{lfclient_url}/generic/all?fields=name")
        except Exception as q_err:
            logger.error("Toolbox: Failed to query existing generic endpoints: %s", q_err)
            return None
        if not gen_data:
            return []
        names = []
        endpoints = gen_data.get('endpoint', gen_data.get('endpoints', []))
        if isinstance(endpoints, dict):
            # single endpoint: dict of the endpoint's own fields, name is under 'name'
            ep_name = endpoints.get('name')
            if ep_name:
                names.append(ep_name)
        elif isinstance(endpoints, list):
            # multiple endpoints: list of {key: {..., 'name': ...}} dicts, real name is the value's 'name'
            for ep in endpoints:
                if isinstance(ep, dict):
                    for ep_values in ep.values():
                        ep_name = ep_values.get('name') if isinstance(ep_values, dict) else None
                        if ep_name:
                            names.append(ep_name)
                elif isinstance(ep, str):
                    names.append(ep)
        return names

    def get_generic_cx_names():
        """Derive the CX names associated with generic endpoints on the manager (build() names
        them 'CX_<tx endpoint name>'), instead of querying manager-wide /cx/all which returns
        every CX type (L3/WanLink/etc.) and would let 'all' actions touch unrelated tests."""
        endp_names = get_generic_endpoint_names()
        if endp_names is None:
            return None
        # 'D_<name>' endpoints are the auto-created RX side of a pair; skip them so each cx is only counted once.
        return ["CX_" + name for name in endp_names if not name.startswith("D_")]

    # Parse custom endpoint / cx names if provided on command line
    raw_endp_names = getattr(args, 'endp_names', None) or getattr(args, 'endp_name', None) or getattr(args, 'cx_names', None) or getattr(args, 'cx_name', None)
    custom_endp_names = parse_list(raw_endp_names) if raw_endp_names else None

    # Retrieve action arguments
    build_requested = getattr(args, 'build', False) or getattr(args, 'build_cxs', False) or getattr(args, 'build_cx', False) or getattr(args, 'build_endp', False)
    start_cx_arg = getattr(args, 'start_cx', None)
    stop_cx_arg = getattr(args, 'stop_cx', None)
    del_cx_arg = getattr(args, 'del_cx', None)

    has_explicit_action = build_requested or (start_cx_arg is not None) or (stop_cx_arg is not None) or (del_cx_arg is not None)

    if not has_explicit_action and (args.test_type or args.cmd):
        build_requested = True

    if not has_explicit_action and not build_requested:
        logger.error("Toolbox: No action specified. Supported actions: --build, --start_gen_cx, --stop_gen_cx, --del_gen_cx")
        return False

    session = LFSession(
        lfclient_url=lfclient_url,
        debug=getattr(args, 'debug', False),
        connection_timeout_sec=4.0,
        stream_errors=True,
        stream_warnings=True,
        require_session=False,
        exit_on_error=False
    )
    command = session.get_command()
    query = session.get_query()

    # Step 1: Build action
    if build_requested:
        if not args.test_type and not args.cmd:
            logger.error("Toolbox: Build action requires --test_type (e.g. ping, iperf3, iperf3-client, iperf3-server, lfcurl, speedtest) or --cmd")
            return False

        logger.info("Toolbox: Executing build action (stations, generic endpoints, cross-connections)...")
        try:
            lf_generic_test = GenTest(
                host=host, port=port,
                lf_user=args.lf_user, lf_passwd=args.lf_passwd,
                radio=args.radio,
                num_stations=args.num_stations,
                use_existing_eid=args.use_existing_eid,
                name_prefix="GT",
                test_type=args.test_type,
                target=args.target,
                cmd=args.cmd,
                interval=args.interval,
                ssid=args.ssid,
                passwd=args.passwd,
                create_report=False,
                port_mgr_cols=args.port_mgr_cols,
                gen_tab_cols=args.gen_tab_cols,
                security=args.security,
                udp=args.udp,
                receive=args.receive,
                test_duration=args.test_duration,
                monitor_interval=args.monitor_interval,
                file_output_lfcurl=args.file_output_lfcurl,
                destination_url_lfcurl=args.destination_url_lfcurl,
                spdtest_enable_debug=args.spdtest_enable_debug,
                spdtest_enable_report=args.spdtest_enable_report,
                spdtest_no_download=args.spdtest_no_download,
                spdtest_no_upload=args.spdtest_no_upload,
                spdtest_single_connection=args.spdtest_single_connection,
                spdtest_ookla=args.spdtest_ookla,
                report_file_path="",
                output_format="csv",
                loop_count=args.loop_count,
                client_port=args.client_port,
                server_port=args.server_port,
                port_wait_time=getattr(args, 'port_wait_time', 60),
                endp_names=custom_endp_names,
                _debug_on=args.debug,
                log_level=args.log_level,
                lf_logger_json=args.lf_logger_json
            )

            if not lf_generic_test.check_tab_exists():
                logger.error("Toolbox: Generic tab is not accessible on LANforge manager at %s:%s. Ensure generic tab is enabled in GUI", host, port)
                return False

            lf_generic_test.check_args()
            lf_generic_test.build()

            # If the build step took the endp name on cmd line, no need to print it out as caller already knows it
            if custom_endp_names:
                logger.info("Toolbox: Generic build completed successfully with specified endpoint name(s): %s", custom_endp_names)
            else:
                # Print enough info to know the endpoint names
                logger.info("================================================================================")
                logger.info("Toolbox: Generic build completed successfully")
                logger.info("Created Generic Endpoints: %s", lf_generic_test.created_endp)
                logger.info("Created Cross-Connections: %s", lf_generic_test.created_cx)
                if lf_generic_test.sta_list:
                    logger.info("Created Stations:          %s", lf_generic_test.sta_list)
                logger.info("================================================================================")

        except Exception as build_err:
            logger.error("Toolbox: Build action failed: %s", build_err)
            return False

    # Step 2: Start cross-connections action (--start_gen_cx)
    if start_cx_arg is not None:
        target_cxs = parse_list(start_cx_arg)
        if not target_cxs or 'all' in [c.lower() for c in target_cxs]:
            logger.info("Toolbox: Starting all generic cross-connections...")
            all_cx_names = get_generic_cx_names()
            if all_cx_names is None:
                logger.error("Toolbox: Could not retrieve generic endpoint list; start-all aborted")
                logger.error("Toolbox: Start cross-connections action failed. Halting remaining actions")
                return False
            elif not all_cx_names:
                logger.warning("Toolbox: No generic cross-connections found on LANforge manager to start")
            else:
                for cx_name in all_cx_names:
                    command.post_set_cx_state(cx_name=cx_name, cx_state="RUNNING", test_mgr="default_tm")
                logger.info("Toolbox: Started %d generic cross-connection(s): %s", len(all_cx_names), all_cx_names)
        else:
            logger.info("Toolbox: Starting specified cross-connection(s): %s", target_cxs)
            any_not_found = False
            for cx in target_cxs:
                cx_name = resolve_cx_name(cx.strip())
                if cx_name is None:
                    logger.error("Toolbox: Cross-connection '%s' not found on LANforge manager, skipping", cx.strip())
                    any_not_found = True
                    continue
                command.post_set_cx_state(cx_name=cx_name, cx_state="RUNNING", test_mgr="default_tm")
                logger.info("Toolbox: Started cross-connection '%s'", cx_name)
            if any_not_found:
                logger.error("Toolbox: Start cross-connections action failed. Halting remaining actions")
                return False

    # Step 3: Stop cross-connections action (--stop_gen_cx)
    if stop_cx_arg is not None:
        target_cxs = parse_list(stop_cx_arg)
        if not target_cxs or 'all' in [c.lower() for c in target_cxs]:
            logger.info("Toolbox: Stopping all generic cross-connections...")
            all_cx_names = get_generic_cx_names()
            if all_cx_names is None:
                logger.error("Toolbox: Could not retrieve generic endpoint list; stop-all aborted")
                logger.error("Toolbox: Stop cross-connections action failed. Halting remaining actions")
                return False
            elif not all_cx_names:
                logger.warning("Toolbox: No generic cross-connections found on LANforge manager to stop")
            else:
                for cx_name in all_cx_names:
                    command.post_set_cx_state(cx_name=cx_name, cx_state="STOPPED", test_mgr="default_tm")
                logger.info("Toolbox: Stopped %d generic cross-connection(s): %s", len(all_cx_names), all_cx_names)
        else:
            logger.info("Toolbox: Stopping specified cross-connection(s): %s", target_cxs)
            any_not_found = False
            for cx in target_cxs:
                cx_name = resolve_cx_name(cx.strip())
                if cx_name is None:
                    logger.error("Toolbox: Cross-connection '%s' not found on LANforge manager, skipping", cx.strip())
                    any_not_found = True
                    continue
                command.post_set_cx_state(cx_name=cx_name, cx_state="STOPPED", test_mgr="default_tm")
                logger.info("Toolbox: Stopped cross-connection '%s'", cx_name)
            if any_not_found:
                logger.error("Toolbox: Stop cross-connections action failed. Halting remaining actions")
                return False

    # Step 4: Delete cross-connections action (--del_gen_cx)
    if del_cx_arg is not None:
        target_cxs = parse_list(del_cx_arg)
        if not target_cxs or 'all' in [c.lower() for c in target_cxs]:
            logger.info("Toolbox: Deleting all generic cross-connections and endpoints...")
            all_cx_names = get_generic_cx_names()
            if all_cx_names is None:
                logger.error("Toolbox: Could not retrieve generic endpoint list; delete-all aborted")
                logger.error("Toolbox: Delete cross-connections action failed. Halting remaining actions")
                return False
            elif not all_cx_names:
                logger.warning("Toolbox: No generic cross-connections found on LANforge manager to delete")
            else:
                for cx_name in all_cx_names:
                    command.post_set_cx_state(cx_name=cx_name, cx_state="STOPPED", test_mgr="default_tm")
                    command.post_rm_cx(cx_name=cx_name, test_mgr="default_tm")
                logger.info("Toolbox: Deleted %d generic cross-connection(s): %s", len(all_cx_names), all_cx_names)

            endp_names = get_generic_endpoint_names()
            if endp_names is None:
                logger.error("Toolbox: Could not retrieve generic endpoint list for endpoint cleanup")
                logger.error("Toolbox: Delete cross-connections action failed. Halting remaining actions")
                return False
            elif endp_names:
                for ep_name in endp_names:
                    command.post_rm_endp(endp_name=ep_name)
                logger.info("Toolbox: Deleted %d generic endpoint(s): %s", len(endp_names), endp_names)
            logger.info("Toolbox: Deleted all generic cross-connections and endpoints")
        else:
            logger.info("Toolbox: Deleting specified cross-connection(s) / endpoint(s): %s", target_cxs)
            any_not_found = False
            for cx in target_cxs:
                cx_name = resolve_cx_name(cx.strip())
                if cx_name is None:
                    logger.error("Toolbox: Cross-connection '%s' not found on LANforge manager, skipping", cx.strip())
                    any_not_found = True
                    continue
                command.post_set_cx_state(cx_name=cx_name, cx_state="STOPPED", test_mgr="default_tm")
                command.post_rm_cx(cx_name=cx_name, test_mgr="default_tm")
                endp_name = cx_name[3:] if cx_name.startswith("CX_") else cx_name
                command.post_rm_endp(endp_name=endp_name)
                logger.info("Toolbox: Deleted cross-connection and endpoint '%s'", cx_name)
            if any_not_found:
                logger.error("Toolbox: Delete cross-connections action failed. Halting remaining actions")
                return False

    return True


def main():

    # definition of create_basic_argparse  in lanforge-scripts/py-json/LANforge/lfcli_base.py around line 700
    parser = argparse.ArgumentParser(
        prog=__file__,
        formatter_class=argparse.RawTextHelpFormatter,
        epilog='''Create generic endpoints and test for their ability to execute chosen commands\n''',
        description='''
        lf_test_generic.py
        --------------------

            EXAMPLES:
                LFPING :
                    ./lf_test_generic.py --mgr 192.168.102.211 --test_type ping --lf_user lanforge --lf_passwd lanforge --num_stations 3 --log_level debug
                                --ssid eero-mesh-lanforge --passwd lanforge --security wpa2 --radio wiphy1 --target www.google.com --test_duration
                                "4s" --create_report --report_file_path "/home/diptidhond/test_generic_1"
                LFCURL :
                    ./lf_test_generic.py --mgr 192.168.102.211 --test_type lfcurl --lf_user lanforge --lf_passwd lanforge --num_stations 3 --log_level debug
                                --ssid eero-mesh-lanforge --passwd lanforge --security wpa2 --radio wiphy1  --test_duration 4s --create_report
                                --report_file_path "/home/lanforge/test_generic_1"
                SPEEDTEST :
                    ./lf_test_generic.py --mgr 192.168.102.211 --test_type speedtest --lf_user lanforge --lf_passwd lanforge --num_stations 3 --log_level debug
                                --ssid mesh-lanforge --passwd lanforge --security wpa2 --radio wiphy1  --test_duration 2m --create_report --no_upload --single_connection
                                --report_file_path "/home/lanforge/test_generic_1"

                IPERF3 :
                    iperf: create 1 client and 1 server. client is already existing & server is on lanforge.

                        ./lf_test_generic.py --mgr 192.168.102.211 --port 8080 --use_existing_eid "1.1.sta00015,1.1.eth3" --test_type iperf3
                            --server_port 5201 --client_port 5201 --target "1.1.eth3" --create_report --port_mgr_cols "alias,rx bytes,tx bytes"


                    iperf -- client only: creates 3 client : 2 clients on sta, 1 on eth port (with existing eid):

                        ./lf_test_generic.py --mgr 192.168.102.211 --test_type iperf3-client --lf_user lanforge --lf_passwd lanforge --num_stations 2 --log_level debug --test_duration 20s
                            --ssid mesh-lanforge --passwd lanforge --security wpa2 --radio wiphy1 --target 192.168.3.3 --use_existing_eid 1.1.eth2 --client_port 9191 --server_port 9191
                            --create_report --report_file_path "/home/lanforge/iperf3_reports" --port_mgr_cols "alias,rx bytes,tx bytes"

                    iperf -- server only: creates 3 servers : 2 servers on sta, 1 on eth port (with existing eid):

                        ./lf_test_generic.py --mgr 192.168.102.211 --test_type iperf3-server --lf_user lanforge --lf_passwd lanforge --num_stations 2 --log_level debug --test_duration 20s
                            --ssid mesh-lanforge --passwd lanforge --security wpa2 --radio wiphy1 --target --use_existing_eid 1.1.eth2 --client_port 9191 --server_port 9191 --create_report
                            --report_file_path "/home/lanforge/iperf3_reports" --port_mgr_cols "alias,rx bytes,tx bytes"


            Port Mgr Cols available to be reported:
                4way time (us)'
                'activity'
                'alias'
                'anqp time (us)'
                'ap'
                'beacon'
                'bps rx'
                'bps rx ll'
                'bps tx'
                'bps tx ll'
                'bytes rx ll'
                'bytes tx ll'
                'channel'
                'collisions'
                'connections'
                'crypt'
                'cx ago'
                'cx time (us)'
                'device'
                'dhcp (ms)'
                'down'
                'entity id'
                'gateway ip'
                'ip'
                'ipv6 address'
                'ipv6 gateway'
                'key/phrase'
                'login-fail'
                'login-ok'
                'logout-fail'
                'logout-ok'
                'mac'
                'mask'
                'misc'
                'mode'
                'mtu'
                'no cx (us)'
                'noise'
                'parent dev'
                'phantom'
                'port'
                'port type'
                'pps rx'
                'pps tx'
                'qlen'
                'reset'
                'retry failed'
                'rx bytes'
                'rx crc'
                'rx drop'
                'rx errors'
                'rx fifo'
                'rx frame'
                'rx length'
                'rx miss'
                'rx over'
                'rx pkts'
                'rx-rate'
                'sec'
                'signal'
                'ssid'
                'status'
                'time-stamp'
                'tx abort'
                'tx bytes'
                'tx crr'
                'tx errors'
                'tx fifo'
                'tx hb'
                'tx pkts'
                'tx wind'
                'tx-failed %'
                'tx-rate'
                'wifi retries'

            Generic Tab Cols available to be reported:
                'bps rx'
                'bps tx'
                'command'
                'dropped'
                'eid'
                'elapsed'
                'entity id'
                'last results'
                'name'
                'pdu/s rx'
                'pdu/s tx'
                'rpt timer'
                'rpt#'
                'rx bytes'
                'rx pkts'
                'status'
                'tx bytes'
                'tx pkts'
                'type'

            TOOLBOX EXAMPLES:
                1. Build Generic Endpoints & Cross-Connections (and exit without running traffic):
                    ./lf_test_generic.py --mgr 192.168.102.211 --toolbox --build --test_type ping --target www.google.com --use_existing_eid 1.1.sta0000,1.1.sta0001
                    ./lf_test_generic.py --mgr 192.168.102.211 --toolbox --build --test_type ping --target 8.8.8.8
                    --radio wiphy1 --num_stations 2 --ssid test --passwd test --security wpa2 --interval 0.01 --endp_names roam_ping_sta0
                2. Start Generic Cross-Connections:
                    ./lf_test_generic.py --mgr 192.168.102.211 --toolbox --start_gen_cx CX_roam_ping_sta0
                    ./lf_test_generic.py --mgr 192.168.102.211 --toolbox --start_gen_cx all
                3. Stop Generic Cross-Connections:
                    ./lf_test_generic.py --mgr 192.168.102.211 --toolbox --stop_gen_cx CX_roam_ping_sta0
                    ./lf_test_generic.py --mgr 192.168.102.211 --toolbox --stop_gen_cx all
                4. Delete Cross-Connections and Endpoints:
                    ./lf_test_generic.py --mgr 192.168.102.211 --toolbox --del_gen_cx CX_roam_ping_sta0
                    ./lf_test_generic.py --mgr 192.168.102.211 --toolbox --del_gen_cx all
                NOTE: "all" acts on all generic-tab endpoints/CXs on the manager, never touching other test types (e.g. L3).
        ''',
    )
    required = parser.add_argument_group('Arguments that must be defined by user:')
    optional = parser.add_argument_group('Arguments that do not need to be defined by user:')

    required.add_argument("--lf_user", type=str, help="user: lanforge", default=None)
    required.add_argument("--lf_passwd", type=str, help="passwd: lanforge", default=None)
    required.add_argument('--test_type', type=str, help='type of command to run. Options: ping, iperf3-client, iperf3-server, iperf3, lfcurl', default=None)

    optional.add_argument('--mgr', '--lfmgr', dest='mgr', help='ip address of lanforge script should be run on. example: 192.168.102.211', default='localhost')
    optional.add_argument('--mgr_port', '--lfmgr_port', dest='mgr_port', help='port which lanforge is running on, on lanforge machine script should be run on. example: 8080', default=8080)
    optional.add_argument('--cmd', help='specifies command to be run by generic type endp', default=None)

    # TOOLBOX ARGS
    optional.add_argument('--toolbox', '--tool_box',
                          dest='toolbox',
                          action='store_true',
                          help='Enable toolbox mode to execute standalone building block actions and exit immediately.')

    optional.add_argument('--build', '--build_cxs', '--build_cx', '--build_endp',
                          dest='build',
                          action='store_true',
                          help='Toolbox action: Execute the build() phase (stations, generic endpoints, cross-connections) and exit.')

    optional.add_argument('--endp_names', '--endp_name', '--cx_names', '--cx_name',
                          dest='endp_names',
                          nargs='+',
                          default=None,
                          help='Specify custom endpoint / cross-connection name(s) (comma or space separated).')

    optional.add_argument('--start_gen_cx',
                          dest='start_cx',
                          nargs='+',
                          default=None,
                          help='Toolbox action: Start specified generic cross-connection(s) by name, or "all" to start all generic CXs on the manager. '
                               'Scoped to generic-tab cross-connections, never other test types.')

    optional.add_argument('--stop_gen_cx',
                          dest='stop_cx',
                          nargs='+',
                          default=None,
                          help='Toolbox action: Stop specified generic cross-connection(s) by name, or "all" to stop all generic CXs on the manager. '
                               'Scoped to generic-tab cross-connections, never other test types.')

    optional.add_argument('--del_gen_cx',
                          dest='del_cx',
                          nargs='+',
                          default=None,
                          help='Toolbox action: Delete specified generic cross-connection(s) and endpoints by name, or "all" to delete all generic endpoints and CXs on the manager. '
                               'Scoped to generic-tab endpoints/CXs, never other test types.')

    # generic endpoint configurations
    optional.add_argument('--spdtest_enable_debug', action="store_true", help='check enable debug box for speedtest cross connect(s)')
    optional.add_argument('--spdtest_enable_report', action="store_true", help='check enable report box for speedtest cross connect(s)')
    optional.add_argument('--spdtest_no_download', action="store_true", help='do not run download for speedtest cross connect')
    optional.add_argument('--spdtest_no_upload', action="store_true", help='do not run upload for speedtest cross connect')
    optional.add_argument('--spdtest_single_connection', action="store_true", help='run speedtest single connection')
    optional.add_argument('--spdtest_ookla', action="store_true", help='run ookla speedtest. ookla license must be on machine in order for this test to run.')
    optional.add_argument('--file_output_lfcurl', help='location to output results of lf_curl, absolute path preferred', default=None)
    optional.add_argument('--destination_url_lfcurl', help='destination url for lfcurl', default=None)
    optional.add_argument('--loop_count', help='determines the number of loops to use in lf_curl and lfping', default=None)
    optional.add_argument('--interval', help='ping interval configuration', default=0.2)
    optional.add_argument('--target',
                          help="""Target for lfping (ex: www.google.com). ALSO arg to specify IP address
                          (if server is OFF lanforge, ex: 192.168.101.1) or LANforge eid
                          (if server port is ON lanfoge, ex: 1.1.eth3) of iperf3 server used for iperf3-client target.""", default=None)
    optional.add_argument('--client_port', help="the port number of the iperf client endpoint. example: -p 5011", default=None)
    optional.add_argument('--server_port', help="the port number of the iperf server endpoint. example: -p 5011", default=None)

    # args for creating stations or using existing eid
    optional.add_argument('--use_existing_eid', help="EID of ports we want to use. Example: '1.1.sta000, 1.1.eth1, 1.1.eth2' ", default=None)
    optional.add_argument('--radio', help="radio that stations should be created on", default=None)
    optional.add_argument('--num_stations', help="number of stations that are to be made, defaults to 1", default=1)
    optional.add_argument('--ssid', help="ssid for stations to connect to", default=None)
    optional.add_argument('--passwd', '-p', help="password to ssid for stations to connect to", default=None)
    optional.add_argument('--mode', help='Used to force mode of stations', default=None)
    optional.add_argument('--ap', help='Used to force a connection to a particular AP, bssid of specific AP', default=None)
    optional.add_argument('--security', help='security for station ssids. options: {open | wep | wpa | wpa2 | wpa3}')

    # additional Client Configuration
    optional.add_argument('--udp', '--UDP', dest='udp', action="store_true", help='Set client mode to UDP, default is TCP')
    optional.add_argument('--receive', '--rcv',  dest='receive', action="store_true", help='Set client mode to receive, default is transmit')

    # dut info
    optional.add_argument("--dut_hw_version", help="dut hw version for kpi.csv, hardware version of the device under test", default="")
    optional.add_argument("--dut_sw_version", help="dut sw version for kpi.csv, software version of the device under test", default="")
    optional.add_argument("--dut_model_num", help="dut model for kpi.csv,  model number / name of the device under test", default="")
    optional.add_argument("--dut_serial_num", help="dut serial for kpi.csv, serial number of the device under test", default="")

    # test tag info
    optional.add_argument("--test_rig", help="test rig for kpi.csv, testbed that the tests are run on", default="")
    optional.add_argument("--test_tag", help="test tag for kpi.csv,  test specific information to differentiate the test", default="")

    # args for reporting
    optional.add_argument('--output_format', help='choose either csv or xlsx. currently xlsx is under construction.', default='csv')
    optional.add_argument('--report_file_path', help='directory to store results in. example: /home/lanforge/report-data/directory-to-store-results', default=None)
    optional.add_argument('--gen_tab_cols', help='Columns wished to be monitored from generic endpoint tab. please use list format. examples example: "name,tx pkts,rx pkts".', default="name,tx pkts,rx pkts,dropped")  # noqa: E501
    optional.add_argument('--port_mgr_cols', help='Columns wished to be monitored from port manager tab. example: "ap,ip,parent dev"', default="ap,ip,parent dev")
    optional.add_argument('--compared_report', help='report path and file which is wished to be compared with new report', default=None)
    optional.add_argument('--create_report', action="store_true", help='specify this flag if test should create report. This means that html, pdf, and csv data is saved and created.')

    # args for test duration & monitor interval
    optional.add_argument('--monitor_interval', help='frequency of monitors measurements;example: 250ms, 35s, 2h', default='2s')
    optional.add_argument('--test_duration', help='duration of the test eg: 30s, 2m, 4h', default="2m")
    optional.add_argument('--port_wait_time', type=int, default=60, help='Max timeout in seconds to wait for port actions (appear, up, ip). Default: 60s')

    # debug and logger
    optional.add_argument('--log_level', default=None, help='Set logging level: debug | info | warning | error | critical')
    optional.add_argument('--lf_logger_json', help="--lf_logger_config_json <json file> , json configuration of logger")
    optional.add_argument('--debug', '-d', default=False, action="store_true", help='Enable debugging')
    optional.add_argument('--help_summary', action="store_true", help='Show summary of what this script does')

    # check if the arguments are empty?
    if len(sys.argv) <= 1 or (len(sys.argv) == 2 and not sys.argv[1]):
        print("lf_test_generic.py script needs the minimum required args. See add the --help flag to check out all possible arguments.")
        sys.exit(1)

    help_summary = '''\
lf_test_generic.py will create stations and endpoints to generate traffic based on a command-line specified command type.

This script will create a variable number of stations to test generic endpoints. Multiple command types can be tested
including ping, speedtest, lfcurl, iperf, generic types. The test will check the last-result attribute for different things
depending on what test is being run. Ping will test for successful pings, speedtest will test for download
speed, upload speed, and ping time, generic will test for successful generic commands.

Toolbox Mode (--toolbox):
  Enables standalone building-block execution (build endpoints & CXs) and exits immediately without running full traffic tests or generating reports.

This script also *does not* use any other file except lanforge_api.py.
'''

    args = parser.parse_args()
    if args.help_summary:
        print(help_summary)
        exit(0)

    logger_config = lf_logger_config.lf_logger_config()
    # set the logger level to requested value
    logger_config.set_level(level=args.log_level)
    logger_config.set_json(json_file=args.lf_logger_json)

    if args.toolbox:
        toolbox_success = handle_toolbox(args)
        sys.exit(0 if toolbox_success else 1)

    if not args.test_type:
        logger.critical("--test_type is a required parameter Options: ping, iperf3-client, iperf3-server, iperf3, lfcurl")
        sys.exit(1)

    if args.create_report:
        if args.report_file_path is None:
            new_file_path = str(datetime.datetime.now().strftime("%Y-%m-%d-%H-h-%M-m-%S-s")).replace(':',
                                                                                                     '-') + '-lf_test_generic'  # create path name
            if os.path.exists('/home/lanforge/report-data/'):
                rpt_file_path = os.path.join('/home/lanforge/report-data/', new_file_path)
                os.mkdir(rpt_file_path)
            else:
                curr_dir_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
                rpt_file_path = os.path.join(curr_dir_path, new_file_path)
                os.mkdir(rpt_file_path)
                # create correct file path
        else:
            if not os.path.exists(args.report_file_path):
                os.mkdir(args.report_file_path)
            rpt_file_path = args.report_file_path

        if args.output_format in ['csv', 'json', 'html', 'hdf', 'stata', 'pickle', 'pdf', 'png', 'xlsx']:
            output = args.output_format
        else:
            logger.info('Not supporting report format: %s. Defaulting to csv data file output type, naming it data.csv.' % args.output_format)
            output = 'csv'
        print("Report file path for csv files and report generated (if expecting report) :" + rpt_file_path)
    else:
        # give blank values for parser
        rpt_file_path = ""
        output = "csv"

    # TODO edit name_prefix
    lf_generic_test = GenTest(host=args.mgr, port=args.mgr_port,
                              lf_user=args.lf_user, lf_passwd=args.lf_passwd,
                              radio=args.radio,
                              num_stations=args.num_stations,
                              use_existing_eid=args.use_existing_eid,
                              name_prefix="GT",
                              test_type=args.test_type,
                              target=args.target,
                              cmd=args.cmd,
                              interval=args.interval,
                              ssid=args.ssid,
                              passwd=args.passwd,
                              create_report=args.create_report,
                              port_mgr_cols=args.port_mgr_cols,
                              gen_tab_cols=args.gen_tab_cols,
                              security=args.security,
                              udp=args.udp,
                              receive=args.receive,
                              test_duration=args.test_duration,
                              monitor_interval=args.monitor_interval,
                              file_output_lfcurl=args.file_output_lfcurl,
                              destination_url_lfcurl=args.destination_url_lfcurl,
                              spdtest_enable_debug=args.spdtest_enable_debug,
                              spdtest_enable_report=args.spdtest_enable_report,
                              spdtest_no_download=args.spdtest_no_download,
                              spdtest_no_upload=args.spdtest_no_upload,
                              spdtest_single_connection=args.spdtest_single_connection,
                              spdtest_ookla=args.spdtest_ookla,
                              report_file_path=rpt_file_path,
                              output_format=output,
                              loop_count=args.loop_count,
                              client_port=args.client_port,
                              server_port=args.server_port,
                              port_wait_time=args.port_wait_time,
                              _debug_on=args.debug,
                              log_level=args.log_level,
                              lf_logger_json=args.lf_logger_json)

    if not lf_generic_test.check_tab_exists():
        raise ValueError("Error received from GUI when trying to request generic tab information, please ensure generic tab is enabled")

    lf_generic_test.check_args()

    lf_generic_test.build()

    lf_generic_test.start()

    logger.info("Starting connections with 5 second settle time")
    lf_generic_test.start()
    time.sleep(5)  # give traffic a chance to get started.

    lf_generic_test.monitor_test()
    print("Done with connection monitoring")

    lf_generic_test.stop()
    # lf_generic_test.generate_report()
    lf_generic_test.cleanup()


if __name__ == "__main__":
    main()