diff --git a/py-scripts/lf_interop_throughput.py b/py-scripts/lf_interop_throughput.py index 62886fe50..cc49b6b69 100755 --- a/py-scripts/lf_interop_throughput.py +++ b/py-scripts/lf_interop_throughput.py @@ -388,15 +388,21 @@ def __init__(self, self.bssids = bssids if bssids else [] # Keep monitoring output stable when a client disconnects or temporarily disappears # from LANforge. Reports still retain one row per configured client. + # Track CX availability and state changes across monitoring iterations. self.missing_cx_logged = set() + self.cx_missing_until_running = set() self.all_devices_stopped = False self.missing_signal_logged = set() - self.cx_not_running_logged = set() + self.last_cx_status = {} + self.cx_has_run = set() self.device_issue_log = [] self.actual_monitoring_duration_seconds = 0 + self.monitoring_started_with_available_cx = False + self.stopped_by_user = False self.pre_monitoring_missing_logged = False self.last_monitor_url = None - self.last_monitor_response = None + self.last_monitor_present_keys = [] + self.current_iteration_cxs = [] self.monitor_start_time = None # Variables related to Robo self.robo_ip = robo_ip @@ -414,39 +420,41 @@ def __init__(self, self.robot.coordinate_list = self.coordinate_list self.robot.total_cycles = self.total_cycles - def record_device_issue(self, device, issue): + def record_device_issue(self, device, issue, api_response=""): """Append a timestamped device/issue entry, later written out as clients_issue.csv.""" + if isinstance(api_response, (dict, list, tuple)): + api_response = json.dumps(api_response, sort_keys=True, default=str) self.device_issue_log.append({ "Time": datetime.now().strftime('%Y-%m-%d %H:%M:%S'), "Device": device, "Issue": issue, + "API Response": api_response, }) - def should_stop_for_missing_cx(self, timeout=40, poll_interval=5): - """Stop the current test run when every CX disappears and recovery never succeeds.""" + def should_stop_for_missing_cx(self, active_cxs=None, timeout=40, poll_interval=5): + """Return True when every CX active in this iteration fails to recover.""" if self.stop_test: logger.info("Stop already requested; skipping missing-CX recovery wait.") return True - if not self.cx_profile.created_cx: + active_cxs = set(active_cxs or self.current_iteration_cxs) + if not active_cxs: return False - if len(self.missing_cx_logged) != len(self.cx_profile.created_cx): + if not active_cxs.issubset(self.missing_cx_logged): return False - logger.warning("All CXs are missing from monitoring data; stopping the current test run.") - if not self.wait_for_any_cx_recovery(timeout=timeout, poll_interval=poll_interval): - logger.error("No devices recovered after the retry window; stopping the test run.") - self.stop_test = True - self.all_devices_stopped = True - if self.robo_ip: - self.robot.update_nav_data_for_all_cxs_stopped() - if self.actual_monitoring_duration_seconds == 0: - raise RuntimeError( - "All {} CX(s) are missing right from the start of monitoring; no data was ever collected.".format( - len(self.cx_profile.created_cx))) + logger.warning("All CXs active in the current iteration are missing from monitoring data.") + if not self.wait_for_any_cx_recovery(active_cxs, timeout, poll_interval): + logger.error("No active CX recovered after the retry window; skipping the current iteration.") + if self.do_bandsteering: + self.stop_test = True + self.all_devices_stopped = True + if self.robo_ip: + self.robot.update_nav_data_for_all_cxs_stopped() return True return self.stop_test - def wait_for_any_cx_recovery(self, timeout=40, poll_interval=5): + def wait_for_any_cx_recovery(self, active_cxs, timeout=40, poll_interval=5): """Wait briefly for a CX to recover, while honoring WebUI stop requests.""" + active_cxs = set(active_cxs) wait_start = datetime.now() while (datetime.now() - wait_start).total_seconds() < timeout: time.sleep(poll_interval) @@ -458,12 +466,13 @@ def wait_for_any_cx_recovery(self, timeout=40, poll_interval=5): if json.load(file).get("status") != "Running": logger.info("Test was stopped by the user during the CX recovery wait.") self.stop_test = True + self.stopped_by_user = True return True except (FileNotFoundError, json.JSONDecodeError) as error: logger.warning("Unable to read WebUI test status during CX recovery: %s", error) - self.get_layer3_endp_data() + self.get_layer3_endp_data(active_cxs) elapsed = (datetime.now() - wait_start).total_seconds() - if len(self.missing_cx_logged) < len(self.cx_profile.created_cx): + if not active_cxs.issubset(self.missing_cx_logged): logger.info("Device(s) responded again after {:.0f}s, resuming.".format(elapsed)) return True logger.warning("Still no devices responding after {:.0f}s, retrying...".format(elapsed)) @@ -475,6 +484,49 @@ def format_monitoring_duration(self): minutes, seconds = divmod(total_seconds, 60) return "{}m {}s".format(minutes, seconds) + def ensure_monitoring_data_collected(self): + """Require monitoring data unless the user explicitly stopped the test.""" + if not self.monitoring_started_with_available_cx and not self.stopped_by_user: + raise RuntimeError("All active CXs were missing in every iteration; no monitoring data was collected.") + + def precheck_all_created_cx_endpoints(self, timeout=40, poll_interval=5): + """Require at least one created CX endpoint before starting iterations.""" + # Validate global endpoint availability independently of CX run state. + all_created_cxs = set(self.cx_profile.created_cx.keys()) + if not all_created_cxs: + raise RuntimeError("No CXs were created; monitoring cannot start.") + self.get_layer3_endp_data(all_created_cxs) + if not all_created_cxs.issubset(self.missing_cx_logged): + return True + logger.warning("All created CX endpoints are missing before monitoring; retrying before starting iterations.") + if self.wait_for_any_cx_recovery(all_created_cxs, timeout, poll_interval): + return not self.stop_test + raise RuntimeError( + "All {} created CX(s) are missing before monitoring; no endpoint was available after retries.".format( + len(all_created_cxs))) + + def append_stopped_monitor_row(self, dataframe, iteration, incremental_capacity_list, overall_start_time): + """Append a final Stopped row when WebUI stops during endpoint retries.""" + if (not dataframe.empty and 'status' in dataframe.columns and + dataframe.iloc[-1].get('status') == 'Stopped'): + return dataframe + timestamp = datetime.now().strftime("%d/%m %I:%M:%S %p") + row = {column: 0 for column in dataframe.columns} + row.update({'Iteration': iteration + 1, 'TIMESTAMP': timestamp, + 'Start_time': overall_start_time.strftime("%d/%m %I:%M:%S %p"), + 'End_time': timestamp, 'Remaining_Time': 0, + 'Incremental_list': ', '.join(str(n) for n in incremental_capacity_list), + 'status': 'Stopped'}) + if 'Angle' in dataframe.columns: + row['Angle'] = self.current_angle if self.current_angle is not None else 0 + dataframe.loc[len(dataframe)] = [row[column] for column in dataframe.columns] + if self.dowebgui: + runtime_csv = 'overall_throughput.csv' if self.group_name else 'throughput_data.csv' + if self.robo_ip and self.current_coordinate is not None: + runtime_csv = '{}_{}'.format(self.current_coordinate, runtime_csv) + dataframe.to_csv(os.path.join(self.result_dir, runtime_csv), index=False) + return dataframe + def perform_robo(self, args, clients_to_run): """ Execute robot-assisted throughput testing across multiple coordinates and angles. @@ -502,6 +554,10 @@ def perform_robo(self, args, clients_to_run): iterations_before_test_stopped_by_user = [] test_stopped_by_user = False + if not self.precheck_all_created_cx_endpoints(): + self.stop() + return + # if band steering is enabled if self.do_bandsteering: # checking the battery status of robot before moving to a point @@ -590,11 +646,14 @@ def perform_robo(self, args, clients_to_run): if not matched: continue - # To add last entry in the csv - all_dataframes = pd.concat( - [df for df in all_dataframes if isinstance(df, pd.DataFrame)], - ignore_index=True - ) + # Generate a band-steering report only when monitoring collected rows. + collected_dataframes = [df for df in all_dataframes if isinstance(df, pd.DataFrame)] + if not collected_dataframes or all(df.empty for df in collected_dataframes): + self.stop() + if args.postcleanup: + self.cleanup() + raise RuntimeError("All active CXs were missing; no band-steering monitoring data was collected.") + all_dataframes = pd.concat(collected_dataframes, ignore_index=True) last_idx = all_dataframes.index[-1] all_dataframes.loc[last_idx, "status"] = "Stopped" @@ -605,6 +664,7 @@ def perform_robo(self, args, clients_to_run): self.stop() if args.postcleanup: self.cleanup() + self.ensure_monitoring_data_collected() iterations_before_test_stopped_by_user.append(0) self.generate_report(list(set(iterations_before_test_stopped_by_user)), incremental_capacity_list, data=all_dataframes, data1=to_run_cxs_len, report_path=self.result_dir) if self.dowebgui: @@ -678,6 +738,9 @@ def perform_robo(self, args, clients_to_run): if args.do_interopability and i != 0: self.stop_specific(to_run_cxs[i - 1]) time.sleep(5) + elif not args.do_interopability and i != 0: + # Reset the previous cumulative CX set before the next capacity. + self.stop_specific(created_cx_lists_keys[:incremental_capacity_list[i - 1]]) if args.interopability_config: if args.do_interopability and i == 0: # To disconnect all the selected devices at the starting selected @@ -687,10 +750,17 @@ def perform_robo(self, args, clients_to_run): # To configure device which is under test is_device_configured = self.configure_specific([device_to_run_resource]) if is_device_configured: - self.start_specific(to_run_cxs[i]) + # Start cumulative CXs for capacity tests or isolated CXs for interoperability. + if args.do_interopability: + self.start_specific(to_run_cxs[i]) + else: + self.start_specific(created_cx_lists_keys[:incremental_capacity_list[i]]) # Determine device names based on the current iteration - device_names = created_cx_lists_keys[:to_run_cxs_len[i][-1]] + if args.do_interopability and args.load_type != "wc_intended_load": + device_names = list(to_run_cxs[i]) + else: + device_names = created_cx_lists_keys[:to_run_cxs_len[i][-1]] # Monitor throughput and capture all dataframes and test stop status all_dataframes, test_stopped_by_user = self.monitor_for_robo(i, individual_df, device_names, incremental_capacity_list, overall_start_time, overall_end_time, is_device_configured) @@ -718,6 +788,8 @@ def perform_robo(self, args, clients_to_run): if args.postcleanup: self.cleanup() + self.ensure_monitoring_data_collected() + # Mark nav_data.json as completed for the Web UI. if args.dowebgui: with open(nav_data, 'r') as x: @@ -1321,20 +1393,46 @@ def stop_specific(self, cx_list): "cx_name": cx_name, "cx_state": "STOPPED" }, debug_=self.debug) + self.clear_endp_counters() + + def clear_endp_counters(self): + """Clear all endpoint counters after CXs stop.""" + self.json_post("/cli-json/clear_endp_counters", { + "endp_name": "all" + }, debug_=self.debug) def stop(self): self.cx_profile.stop_cx() + self.clear_endp_counters() self.station_profile.admin_down() def remove_missing_cx(self): - """Drop CXs still marked missing out of created_cx so stop()/cleanup() skip them.""" - if not self.missing_cx_logged or not self.cx_profile.created_cx: + """Remove locally tracked CXs absent from the current Layer-3 CX list.""" + # Refresh Layer-3 CX availability before modifying the local CX profile. + if not self.cx_profile.created_cx: + return + try: + cx_response = self.json_get('/cx/all') + except Exception as error: + logger.warning("Unable to refresh the Layer-3 CX list before cleanup: %s", error) return - missing_cxs = set(self.missing_cx_logged).intersection(self.cx_profile.created_cx.keys()) + if not isinstance(cx_response, dict) or not cx_response: + logger.warning("Unable to refresh the Layer-3 CX list before cleanup; unexpected response: %s", cx_response) + return + created_cxs = set(self.cx_profile.created_cx.keys()) + metadata_keys = {'handler', 'uri', 'buttons', 'empty'} + available_cxs = set(cx_response.keys()) - metadata_keys + for value in cx_response.values(): + entries = value if isinstance(value, list) else [value] + for entry in entries: + if isinstance(entry, dict) and entry.get('name'): + available_cxs.add(entry['name']) + available_cxs.intersection_update(created_cxs) + missing_cxs = created_cxs - available_cxs if missing_cxs: logger.warning( - "Excluding %s missing CX(s) because endpoints were unavailable: %s", + "Excluding %s CX(s) missing from the current Layer-3 response: %s", len(missing_cxs), sorted(missing_cxs) ) for cx in missing_cxs: @@ -1346,11 +1444,10 @@ def pre_cleanup(self): def cleanup(self): if self.robo_ip: self.remove_missing_cx() - logger.info("self.cx_profile.created_cx %s", self.cx_profile.created_cx) logger.info("cleanup done") self.cx_profile.cleanup() - def get_layer3_endp_data(self): + def get_layer3_endp_data(self, active_cxs=None): """ Fetches Layer 3 endpoint data for all created cross connections. @@ -1364,8 +1461,9 @@ def get_layer3_endp_data(self): [4]: Status of the Device ("Run" or "Stopped") """ cx_list = list(self.cx_profile.created_cx.keys()) - # One URL fetching both A/B endpoints for every CX in a single request. - endpoint_names = [endpoint for cx in cx_list for endpoint in (cx + '-A', cx + '-B')] + active_cxs = set(active_cxs or self.current_iteration_cxs or cx_list) + endpoint_names = [endpoint for cx in cx_list if cx in active_cxs + for endpoint in (cx + '-A', cx + '-B')] monitor_url = '/endp/{}/list?fields=rx rate (last),rx drop %25,name,run'.format(','.join(endpoint_names)) endpoint_response = {} cx_response = {} @@ -1375,18 +1473,8 @@ def get_layer3_endp_data(self): except Exception as e: # Partial /endp response is expected for phantom clients; continue with default metrics. logger.error("Endpoint not fetched from API: %s", e) - logger.error("URL : %s", monitor_url) - logger.error("Response: %s", endpoint_response) - - if not endpoint_response: - # json_get() returned None/empty with no exception; surface it instead of - # silently treating every CX as missing with no clue why. - logger.warning("Empty response fetching endpoint data.") - logger.warning("URL : %s", monitor_url) - logger.warning("Response: %s", endpoint_response) self.last_monitor_url = monitor_url - self.last_monitor_response = endpoint_response # Normalize the two response shapes /endp/.../list can return into one name->metrics lookup. endpoint_data = endpoint_response.get('endpoint', []) if isinstance(endpoint_response, dict) else [] if isinstance(endpoint_data, dict): @@ -1408,14 +1496,18 @@ def get_layer3_endp_data(self): if isinstance(value, dict): metrics_by_endpoint[value.get('name', name)] = value + self.last_monitor_present_keys = sorted(metrics_by_endpoint.keys()) + rtt_by_cx = {} state_by_cx = {} + cx_object_by_name = {} # /cx/all keys its entries by index, not by name, so look up 'name' per value. if isinstance(cx_response, dict): for value in cx_response.values(): if isinstance(value, dict) and value.get('name'): rtt_by_cx[value['name']] = value.get('avg rtt', 0) state_by_cx[value['name']] = value.get("state", "Stopped") + cx_object_by_name[value['name']] = value throughput = {} # Suppress "not running" warnings for the first 10s so fresh CXs don't false-positive. @@ -1425,29 +1517,41 @@ def get_layer3_endp_data(self): endp_a, endp_b = cx + '-A', cx + '-B' a_metrics = metrics_by_endpoint.get(endp_a) b_metrics = metrics_by_endpoint.get(endp_b) - # Neither endpoint reported: log once when it first goes missing, not every poll. - if a_metrics is None and b_metrics is None: + cx_is_present = a_metrics is not None or b_metrics is not None + cx_is_active = cx in active_cxs + if cx_is_active and not cx_is_present: if cx not in self.missing_cx_logged: - logger.warning("CX '%s' is missing from monitoring data; continuing with the remaining devices.\nURL : %s\nResponse: %s", - cx, monitor_url, endpoint_response) + logger.warning("CX '%s' is missing from monitoring data; continuing with the remaining devices.\nURL : %s\nEndpoint keys present: %s", + cx, monitor_url, self.last_monitor_present_keys) self.missing_cx_logged.add(cx) - self.record_device_issue(cx, "CX missing from monitoring data") - elif cx in self.missing_cx_logged: - # It came back: clear the flag and log the recovery once. + self.cx_missing_until_running.add(cx) + self.record_device_issue(cx, "CX missing from monitoring data", self.last_monitor_present_keys) + elif cx_is_active and cx in self.missing_cx_logged: logger.info("CX '%s' data is available again.", cx) self.missing_cx_logged.discard(cx) - # Either endpoint reporting "run" counts the CX as running. running = any(bool(metrics and metrics.get('run')) for metrics in (a_metrics, b_metrics)) status = 'Run' if running else state_by_cx.get(cx, "Stopped") - if status != 'Run' and past_grace_period and cx not in self.cx_not_running_logged: - # Log the transition into "not running" once, not on every poll. - logger.warning("CX '%s' status is '%s', not running.", cx, status) - self.cx_not_running_logged.add(cx) - self.record_device_issue(cx, "CX status is '{}', not running".format(status)) - elif status == 'Run' and cx in self.cx_not_running_logged: - logger.info("CX '%s' status is back to running.", cx) - self.cx_not_running_logged.discard(cx) + previous_status = self.last_cx_status.get(cx) + if cx_is_active and cx_is_present: + # Ignore the expected initial Stopped/Waiting-to-Run transition. + if status == 'Run' and cx not in self.cx_has_run: + self.last_cx_status[cx] = status + self.cx_has_run.add(cx) + elif previous_status is None: + self.last_cx_status[cx] = status + elif status != previous_status: + missing_and_not_running = cx in self.cx_missing_until_running and status != 'Run' + if not missing_and_not_running: + issue = "Status ({}->{})".format(previous_status, status) + if status == 'Run': + logger.info("CX '%s' %s.", cx, issue) + elif past_grace_period: + logger.warning("CX '%s' %s.", cx, issue) + self.record_device_issue(cx, issue, cx_object_by_name.get(cx, {})) + self.last_cx_status[cx] = status + if status == 'Run': + self.cx_missing_until_running.discard(cx) throughput[index] = [ (a_metrics or {}).get('rx rate (last)', 0), @@ -1472,6 +1576,9 @@ def monitor(self, iteration, individual_df, device_names, incremental_capacity_l if self.cx_profile.created_cx is None: raise ValueError("Monitor needs a list of Layer 3 connections") + # Restrict missing and status checks to CXs used by this iteration. + self.current_iteration_cxs = list(device_names) + start_time = datetime.now() if self.monitor_start_time is None: self.monitor_start_time = start_time @@ -1482,13 +1589,20 @@ def monitor(self, iteration, individual_df, device_names, incremental_capacity_l # Don't start an interval if every CX is already missing; give recovery a chance first. if self.cx_profile.created_cx: - self.get_layer3_endp_data() - if self.should_stop_for_missing_cx(): - return individual_df, True - if self.missing_cx_logged and not self.pre_monitoring_missing_logged: - logger.warning("Missing before monitoring; continuing with %s device(s): %s\nURL : %s\nResponse: %s", - len(self.cx_profile.created_cx) - len(self.missing_cx_logged), - sorted(self.missing_cx_logged), self.last_monitor_url, self.last_monitor_response) + self.get_layer3_endp_data(self.current_iteration_cxs) + if self.should_stop_for_missing_cx(self.current_iteration_cxs): + if self.stopped_by_user: + individual_df = self.append_stopped_monitor_row( + individual_df, iteration, incremental_capacity_list, overall_start_time) + # Propagate missing CX failure to stop band-steering coordinates. + return individual_df, True if self.do_bandsteering else self.stop_test + self.monitoring_started_with_available_cx = True + missing_active_cxs = set(self.current_iteration_cxs).intersection(self.missing_cx_logged) + if missing_active_cxs and not self.pre_monitoring_missing_logged: + logger.warning("Missing before monitoring; continuing with %s device(s): %s\nURL : %s\nEndpoint keys present: %s", + len(self.current_iteration_cxs) - len(missing_active_cxs), + sorted(missing_active_cxs), self.last_monitor_url, + self.last_monitor_present_keys) self.pre_monitoring_missing_logged = True # Initialize variables for real-time connections data @@ -1519,10 +1633,10 @@ def monitor(self, iteration, individual_df, device_names, incremental_capacity_l logger.info("Stop already requested; ending monitoring interval.") test_stopped_by_user = True break - throughput[index] = self.get_layer3_endp_data() - if self.cx_profile.created_cx and self.should_stop_for_missing_cx(): - logger.error("No devices recovered; ending this monitoring interval with collected data.") - test_stopped_by_user = True + throughput[index] = self.get_layer3_endp_data(self.current_iteration_cxs) + if self.current_iteration_cxs and self.should_stop_for_missing_cx(self.current_iteration_cxs): + # Propagate missing CX failure to stop band-steering coordinates. + test_stopped_by_user = True if self.do_bandsteering else self.stop_test break # Check if next sleep would overshoot the end_time is_last_iteration = ((current_time + timedelta(seconds=1 if self.dowebgui else self.report_timer)) >= end_time) @@ -1616,6 +1730,7 @@ def monitor(self, iteration, individual_df, device_names, incremental_capacity_l if data["status"] != "Running": logger.warning('Test is stopped by the user') test_stopped_by_user = True + self.stopped_by_user = True if self.do_bandsteering: self.actual_monitoring_duration_seconds += (datetime.now() - start_time).total_seconds() return individual_df, test_stopped_by_user @@ -1815,9 +1930,6 @@ def monitor(self, iteration, individual_df, device_names, incremental_capacity_l for i in range(len(upload_throughput)): connections_upload.update({keys[i]: float(f"{(upload_throughput[i]):.2f}")}) - logger.info("connections download {}".format(connections_download)) - logger.info("connections upload {}".format(connections_upload)) - self.actual_monitoring_duration_seconds += (datetime.now() - start_time).total_seconds() return individual_df, test_stopped_by_user @@ -1848,6 +1960,9 @@ def monitor_for_robo(self, iteration, individual_df, device_names, incremental_c if self.cx_profile.created_cx is None: raise ValueError("Monitor needs a list of Layer 3 connections") + # Restrict missing and status checks to CXs used by this robot iteration. + self.current_iteration_cxs = list(device_names) + start_time = datetime.now() if self.monitor_start_time is None: self.monitor_start_time = start_time @@ -1857,9 +1972,14 @@ def monitor_for_robo(self, iteration, individual_df, device_names, incremental_c self.overall = [] if self.cx_profile.created_cx: - self.get_layer3_endp_data() - if self.should_stop_for_missing_cx(): + self.get_layer3_endp_data(self.current_iteration_cxs) + if self.should_stop_for_missing_cx(self.current_iteration_cxs): + if self.stopped_by_user: + individual_df = self.append_stopped_monitor_row( + individual_df, iteration, incremental_capacity_list, overall_start_time) + # End robot monitoring so collected data can proceed to report generation. return individual_df, True + self.monitoring_started_with_available_cx = True # Initialize variables for real-time connections data index = -1 @@ -1925,9 +2045,9 @@ def monitor_for_robo(self, iteration, individual_df, device_names, incremental_c logger.info("Stop already requested; ending monitoring interval.") test_stopped_by_user = True break - throughput[index] = self.get_layer3_endp_data() - if self.cx_profile.created_cx and self.should_stop_for_missing_cx(): - logger.error("No devices recovered; ending this monitoring interval with collected data.") + throughput[index] = self.get_layer3_endp_data(self.current_iteration_cxs) + if self.current_iteration_cxs and self.should_stop_for_missing_cx(self.current_iteration_cxs): + # End robot monitoring so collected data can proceed to report generation. test_stopped_by_user = True break # Check if next sleep would overshoot the end_time @@ -2068,6 +2188,7 @@ def monitor_for_robo(self, iteration, individual_df, device_names, incremental_c if data["status"] != "Running": logger.warning('Test is stopped by the user') test_stopped_by_user = True + self.stopped_by_user = True break # Adjust time_gap based on elapsed time since start (for webui) @@ -2309,9 +2430,6 @@ def monitor_for_robo(self, iteration, individual_df, device_names, incremental_c for i in range(len(upload_throughput)): connections_upload.update({keys[i]: float(f"{(upload_throughput[i]):.2f}")}) - logger.info("connections download {}".format(connections_download)) - logger.info("connections upload {}".format(connections_upload)) - self.actual_monitoring_duration_seconds += (datetime.now() - start_time).total_seconds() return individual_df, test_stopped_by_user @@ -2888,6 +3006,10 @@ def generate_report(self, iterations_before_test_stopped_by_user, incremental_ca data_iter = data[data['Iteration'] == i + 1] avg_rtt_data = [] + if data_iter.empty: + logger.warning("Skipping report section for iteration %s because no monitoring data was collected.", i + 1) + continue + # for sig in self.signal_list[0:int(incremental_capacity_list[i])]: # signal_data.append(int(sig)*(-1)) # rssi_signal_data.append(signal_data) @@ -3358,6 +3480,10 @@ def generate_report(self, iterations_before_test_stopped_by_user, incremental_ca data_iter = data[data['Iteration'] == i + 1] avg_rtt_data = [] + if data_iter.empty: + logger.warning("Skipping report section for iteration %s because no monitoring data was collected.", i + 1) + continue + # Fetch devices_on_running from real_client_list devices_on_running.append(self.real_client_list[data1[i][-1] - 1].split(" ")[-1]) # If the device fails to configure, skip its data in the report @@ -3832,6 +3958,12 @@ def generate_report_robo(self, iterations_before_test_stopped_by_user, increment data_iter = data[data['Iteration'] == i + 1] avg_rtt_data = [] + if data_iter.empty: + logger.warning( + "Skipping report section for iteration %s at coordinate %s because no monitoring data was collected.", + i + 1, coordinate) + continue + # for sig in self.signal_list[0:int(incremental_capacity_list[i])]: # signal_data.append(int(sig)*(-1)) # rssi_signal_data.append(signal_data) @@ -5207,6 +5339,10 @@ def main(): throughput.perform_robo(args, clients_to_run) exit(1) + if not throughput.precheck_all_created_cx_endpoints(): + throughput.stop() + return + individual_dataframe_column = [] to_run_cxs, to_run_cxs_len, created_cx_lists_keys, incremental_capacity_list = throughput.get_incremental_capacity_list() @@ -5246,6 +5382,9 @@ def main(): if args.do_interopability and i != 0: throughput.stop_specific(to_run_cxs[i - 1]) time.sleep(5) + elif not args.do_interopability and i != 0: + # Reset the previous cumulative CX set before the next capacity. + throughput.stop_specific(created_cx_lists_keys[:incremental_capacity_list[i - 1]]) if args.interopability_config: if args.do_interopability and i == 0: # To disconnect all the selected devices at the starting selected @@ -5255,10 +5394,17 @@ def main(): # To configure device which is under test is_device_configured = throughput.configure_specific([device_to_run_resource]) if is_device_configured: - throughput.start_specific(to_run_cxs[i]) + # Start cumulative CXs for capacity tests or isolated CXs for interoperability. + if args.do_interopability: + throughput.start_specific(to_run_cxs[i]) + else: + throughput.start_specific(created_cx_lists_keys[:incremental_capacity_list[i]]) # Determine device names based on the current iteration - device_names = created_cx_lists_keys[:to_run_cxs_len[i][-1]] + if args.do_interopability and args.load_type != "wc_intended_load": + device_names = list(to_run_cxs[i]) + else: + device_names = created_cx_lists_keys[:to_run_cxs_len[i][-1]] # Monitor throughput and capture all dataframes and test stop status all_dataframes, test_stopped_by_user = throughput.monitor(i, individual_df, device_names, incremental_capacity_list, overall_start_time, overall_end_time, is_device_configured) @@ -5282,6 +5428,7 @@ def main(): throughput.stop() if args.postcleanup: throughput.cleanup() + throughput.ensure_monitoring_data_collected() iot_summary = None if args.iot_test and args.iot_testname: # Load IoT summary data from the specified JSON file