scottk/wlan-lanforge-scripts
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-lanforge-scripts.git synced 2025-11-01 03:07:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update py_scripts to be in line with pip standards

Browse Source 
Signed-off-by: Matthew Stidham <stidmatt@gmail.com>
This commit is contained in:
Matthew Stidham
2021-09-08 16:57:10 -07:00
parent 0be47c82de
commit 37df48594e
31 changed files with 260 additions and 235 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
py-scripts/__init__.py
View File

@@ -0,0 +1,75 @@
#from .connection_test import ConnectionTest
from .create_bond import CreateBond
from .create_bridge import CreateBridge
from .create_chamberview import CreateChamberview
from .create_l3 import CreateL3
from .create_l4 import CreateL4
from .create_macvlan import CreateMacVlan
from .create_qvlan import CreateQVlan
from .create_station import CreateStation
from .create_vap import CreateVAP
from .csv_convert import CSVParcer
from .csv_processor import L3CSVParcer
from .csv_to_influx import CSVtoInflux
from .csv_to_grafana import UseGrafana
from .download_test import DownloadTest
from .event_breaker import EventBreaker
from .event_flood import EventBreaker as EventFlood
from .example_security_connection import IPv4Test
from .ghost_profile import UseGhost
from .grafana_profile import UseGrafana
from .influx import RecordInflux
from .layer3_test import Layer3Test
from .layer4_test import HTTPTest
from .lf_ap_auto_test import ApAutoTest
from .lf_atten_mod_test import CreateAttenuator
from .lf_csv import lf_csv, lf_kpi_csv
from .lf_dataplane_test import DataplaneTest
from .lf_dfs_test import FileAdapter, CreateCtlr, L3VariableTime
from .lf_dut_sta_vap_test import Login_DUT, LoadScenario, CreateSTA_CX
from .lf_ftp_test import ftp_test
from .lf_ftp import FtpTest
from .lf_graph import lf_bar_graph, lf_stacked_graph, lf_horizontal_stacked_graph, lf_scatter_graph, lf_line_graph
from .lf_mesh_test import MeshTest
from .lf_multipsk import MultiPsk
from .lf_report import lf_report
from .lf_rvr_test import RvrTest
from .lf_rx_sensitivity_test import RxSensitivityTest
from .lf_sniff_radio import SniffRadio
#from .lf_snp_test import SAME CLASS NAMES AS LF_DFS_TEST
#from .lf_tr398_test import DataPlaneTest
from .lf_webpage import HttpDownload
from .lf_wifi_capacity_test import WiFiCapacityTest
from .measure_station_time_up import MeasureTimeUp
from .modify_station import ModifyStation
from .modify_vap import ModifyVAP
from .run_cv_scenario import RunCvScenario
from .sta_connect import StaConnect
from .sta_connect2 import StaConnect2
from .sta_connect_bssid_mac import client_connect
from .station_layer3 import STATION
from .stations_connected import StationsConnected
from .test_1k_clients_jedtest import Test1KClients
from .test_client_admission import LoadLayer3
from .test_fileio import FileIOTest
from .test_generic import GenTest
from .test_ip_connection import ConnectTest
from .test_ip_variable_time import IPVariableTime
from .test_ipv4_ttls import TTLSTest
from .test_ipv4_ps import IPV4VariableTime
#from .test_l3_longevity import L3VariableTime ALSO IN LF_DFS_TEST
from .test_l3_powersave_traffic import L3PowersaveTraffic
#from .test_l3_scenario_throughput import
from .test_l3_unicast_traffic_gen import L3VariableTimeLongevity
from .test_l3_WAN_LAN import VRTest
from .test_l4 import IPV4L4
from .test_status_msg import TestStatusMessage
from .test_wanlink import LANtoWAN
from .test_wpa_passphrases import WPAPassphrases
from .testgroup import TestGroup
from .testgroup2 import TestGroup2
from .tip_station_powersave import TIPStationPowersave
from .vap_stations_example import VapStations
from .video_rates import VideoRates
from .wlan_capacity_calculator import main as WlanCapacityCalculator
from .ws_generic_monitor_test import WS_Listener

4
py-scripts/create_bond.py
View File

@@ -20,12 +20,8 @@ if sys.version_info[0] != 3:
if 'py-json' not in sys.path:
sys.path.append(os.path.join(os.path.abspath('..'), 'py-json'))
import LANforge
from LANforge.lfcli_base import LFCliBase
from LANforge import LFUtils
from realm import Realm
import time
import pprint
class CreateBond(LFCliBase):

2
py-scripts/create_bridge.py
View File

@@ -14,11 +14,9 @@ if sys.version_info[0] != 3:
if 'py-json' not in sys.path:
sys.path.append(os.path.join(os.path.abspath('..'), 'py-json'))
import LANforge
from LANforge.lfcli_base import LFCliBase
from LANforge import LFUtils
from realm import Realm
import time
import pprint

3
py-scripts/create_vap.py
View File

@@ -14,12 +14,9 @@ if sys.version_info[0] != 3:
if 'py-json' not in sys.path:
sys.path.append(os.path.join(os.path.abspath('..'), 'py-json'))
import LANforge
from LANforge.lfcli_base import LFCliBase
from LANforge import LFUtils
import realm
from realm import Realm
import time
import pprint

4
py-scripts/layer3_test.py
View File

@@ -9,10 +9,8 @@ if sys.version_info[0] != 3:
if 'py-json' not in sys.path:
sys.path.append('../py-json')
from LANforge import LFUtils
from LANforge import lfcli_base
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
import realm
from realm import Realm
import argparse
import datetime
@@ -42,7 +40,7 @@ class Layer3Test(LFCliBase):
self.traffic_type = traffic_type
self.side_a_speed = side_a_speed
self.side_b_speed = side_b_speed
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.local_realm = Realm(lfclient_host=self.host, lfclient_port=self.port)
self.station_profile = self.local_realm.new_station_profile()
self.cx_profile = self.local_realm.new_l3_cx_profile()

10
py-scripts/layer4_test.py
View File

@@ -17,17 +17,13 @@ if sys.version_info[0] != 3:
if 'py-json' not in sys.path:
sys.path.append('../py-json')
from LANforge import LFUtils
from LANforge import lfcli_base
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
import realm
from realm import PortUtils
from realm import PortUtils, Realm
import argparse
import datetime
import time
from test_utility import CreateHTML
from test_utility import RuntimeUpdates
import pdfkit
from test_utility import CreateHTML, RuntimeUpdates
import json
import re
import os
@@ -60,7 +56,7 @@ class HTTPTest(LFCliBase):
self.urls_ps = url_ps
self.test_update =_test_update
self.test_update.send_update({"test_status": '1', "duration_left": "initializing...", "data": 'None'})
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.local_realm = Realm(lfclient_host=self.host, lfclient_port=self.port)
self.test_duration = self.local_realm.parse_time(duration)
self.station_profile = self.local_realm.new_station_profile()
self.port_util = PortUtils(self.local_realm)

6
py-scripts/lf_atten_mod_test.py
View File

@@ -32,16 +32,14 @@ if 'py-json' not in sys.path:
import argparse
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
from LANforge import LFUtils
import realm
import time
from realm import Realm
class CreateAttenuator(LFCliBase):
def __init__(self, host, port, serno, idx, val,
_debug_on=False,
_exit_on_error=False,
_exit_on_fail=False):
super().__init__(host, port, _local_realm=realm.Realm(host, port), _debug=_debug_on, _exit_on_fail=_exit_on_fail)
super().__init__(host, port, _local_realm=Realm(host, port), _debug=_debug_on, _exit_on_fail=_exit_on_fail)
self.host = host
self.port = port
self.serno = serno

261
py-scripts/lf_dut_sta_vap_test.py
View File

@@ -27,47 +27,42 @@ import sys
if sys.version_info[0] != 3:
print("This script requires Python 3")
exit(1)
if 'py-json' not in sys.path:
sys.path.append('../py-json')
if 'py_json' not in sys.path:
sys.path.append('../py_json')
import argparse
import time
from LANforge import LFUtils
from LANforge import lfcli_base
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
import realm
from realm import Realm
import logging
import paramiko as pm
from paramiko.ssh_exception import NoValidConnectionsError as exception
import xlsxwriter
from bokeh.io import output_file, show
from bokeh.io import show
from bokeh.plotting import figure
from bokeh.models import LinearAxis, Range1d
from bokeh.models import HoverTool
from bokeh.layouts import row
# Specifically for Measuring CPU Core Temperatures
class Login_DUT:
def __init__(self, threadID, name, HOST):
self.threadID = threadID
self.name = name
self.host=HOST
self.USERNAME = "lanforge"
self.PASSWORD = "lanforge"
self.CLIENT= pm.SSHClient()
self.LF1= self.Connect()
self.data_core1=[]
self.data_core2=[]
if self.CLIENT == 0:
exit()
print("Connected to " +HOST+" DUT to Measure the Core Temperature")
self.threadID = threadID
self.name = name
self.host=HOST
self.USERNAME = "lanforge"
self.PASSWORD = "lanforge"
self.CLIENT= pm.SSHClient()
self.LF1= self.Connect()
self.data_core1=[]
self.data_core2=[]
if self.CLIENT == 0:
exit()
print("Connected to " +HOST+" DUT to Measure the Core Temperature")
def run(self):
stdin, stdout, stderr= self.CLIENT.exec_command("sensors")
out_lines = stdout.readlines()
@@ -75,14 +70,14 @@ class Login_DUT:
print(out_lines[len(out_lines)-3], out_lines[len(out_lines)-2])
self.data_core1.append(out_lines[len(out_lines)-3])
self.data_core2.append(out_lines[len(out_lines)-2])
def Connect(self):
self.CLIENT.load_system_host_keys()
self.CLIENT.set_missing_host_key_policy(pm.AutoAddPolicy())
try:
self.CLIENT.connect(self.host, username=self.USERNAME, password=self.PASSWORD,timeout=10)
return None
return None
except exception as error:
self.CLIENT = 0;
return None
@@ -100,14 +95,14 @@ class LoadScenario(LFCliBase):
# Class to create stations and run L3 Cross connects and run them for given time. It also stores the endpoint names for measuring throughput
class CreateSTA_CX(LFCliBase):
def __init__(self, host, port, radio, num_sta, sta_id, ssid, security, password, upstream, protocol, min_bps, max_bps, security_debug_on=True, _exit_on_error=True, _exit_on_fail=True):
super().__init__(host, port, _debug=security_debug_on, _exit_on_fail=_exit_on_fail)
self.host = host
self.port = port
self.radio = radio
self.num_sta = num_sta
self.sta_id = sta_id
@@ -136,7 +131,7 @@ class CreateSTA_CX(LFCliBase):
self.endp=[]
for i in range(sta_id,sta_id+num_sta):
self.sta_list.append("sta00")
#portDhcpUpRequest
'''
upstream_dhcp = LFRequest.LFRequest("http://"+str(host)+":"+str(port)+"/"+"/cli-form/set_port")
@@ -156,36 +151,36 @@ class CreateSTA_CX(LFCliBase):
self.station_profile.admin_up()
#Wait for a while
time.sleep(15)
#Setting up the Parameters for CX
self.cx_profile.side_a_min_bps = self.min_bps
self.cx_profile.side_b_min_bps = self.min_bps
self.cx_profile.side_a_max_bps = self.max_bps
self.cx_profile.side_b_max_bps = self.max_bps
self.cx_profile.side_a_min_pdu = 'Auto'
self.cx_profile.side_b_min_pdu = 'Auto'
self.cx_profile.report_timer = 1000
self.cx_profile.side_a_min_pkt='Same'
self.cx_profile.side_a_max_pkt='Same'
#Create Connections of Given Parameters
self.cx_profile.create(self.protocol, side_a="1.1."+self.upstream, side_b=list(self.local_realm.find_ports_like("sta0+")))
time.sleep(15)
# Getting all the Endpoint Names for measuring Throughput Later
for i in self.cx_profile.get_cx_names():
self.cx_names.append(i)
self.cx_names.append(i)
for j in self.cx_names:
x=self.local_realm.json_get("/cx/"+j)
self.endp.append(x.get(j).get('endpoints')[1])
#print(self.endp)
return 0
def start(self):
#self.station_profile.admin_up()
self.cx_profile.start_cx()
time.sleep(5)
return 0
@@ -200,7 +195,7 @@ class CreateSTA_CX(LFCliBase):
def cleanup(self):
# Removing Connections
self.local_realm.cleanup_cxe_prefix(self.cx_profile.name_prefix)
vap = self.local_realm.find_ports_like("vap+")
bridges = self.local_realm.find_ports_like("br+")
station_map = self.local_realm.find_ports_like("sta+")
@@ -212,7 +207,7 @@ class CreateSTA_CX(LFCliBase):
for eid,record in vap.items():
self.local_realm.remove_vlan_by_eid(eid)
time.sleep(0.03)
#Removing stations
for eid,record in station_map.items():
self.local_realm.remove_vlan_by_eid(eid)
@@ -228,13 +223,13 @@ class CreateSTA_CX(LFCliBase):
LFUtils.waitUntilPortsDisappear(base_url=self.local_realm.lfclient_url, port_list=del_sta_names, debug=True)
print("Ports Successfully Cleaned up")
return 0
except:
except:
print("Ports Successfully Cleaned up")
time.sleep(5)
return 0
# Generates XLSX Report
# Generates XLSX Report
def GenerateReport(throughput_sta, throughput_vap, core1_temp, core2_temp, duration,name):
workbook = xlsxwriter.Workbook(name)
worksheet = workbook.add_worksheet()
@@ -277,7 +272,7 @@ def GenerateReport(throughput_sta, throughput_vap, core1_temp, core2_temp, durat
# Plotting Function for Parameters
def plot(throughput_sta, throughput_vap, core1_temp, core2_temp, Time):
s1 = figure()
s1.title.text = "WIFI Throughput vs Temperature Plot"
s1.xaxis.axis_label = "Time in Seconds"
@@ -288,10 +283,10 @@ def plot(throughput_sta, throughput_vap, core1_temp, core2_temp, Time):
s1.line( Time, throughput_vap, color='blue')
#s1.circle(Time, throughput_vap, color='blue')
s1.extra_y_ranges = {"Temperature": Range1d(start=0, end=150)}
s1.add_layout(LinearAxis(y_range_name="Temperature", axis_label="Temperature in Degree Celsius"), 'right')
s1.line(Time, core1_temp, y_range_name='Temperature', color='red')
#s1.circle(Time, core1_temp, y_range_name='Temperature', color='red')
@@ -300,7 +295,7 @@ def plot(throughput_sta, throughput_vap, core1_temp, core2_temp, Time):
show(s1)
# Creates the Instance for LFCliBase
class VAP_Measure(LFCliBase):
def __init__(self, lfclient_host, lfclient_port):
@@ -312,7 +307,7 @@ class VAP_Measure(LFCliBase):
def main():
parser = argparse.ArgumentParser(description="Test Scenario of DUT Temperature measurement along with simultaneous throughput on VAP as well as stations")
parser.add_argument("-m", "--manager", type=str, help="Enter the address of Lanforge Manager (By default localhost)")
parser.add_argument("-sc", "--scenario", type=str, help="Enter the Name of the Scenario you want to load (by Default DFLT)")
parser.add_argument("-r", "--radio", type=str, help="Enter the radio on which you want to create a station/s on ")
@@ -328,112 +323,112 @@ def main():
parser.add_argument("-t", "--duration", type=int, help="Enter the Time for which you want to run test (In Minutes)")
parser.add_argument("-o", "--report_name", type=str, help="Enter the Name of the Output file ('Report.xlsx')")
args = None
try:
args = parser.parse_args()
# Lanforge Manager IP Address
if (args.manager is None):
manager = "localhost"
if (args.manager is not None):
manager = args.manager
# Scenario Name
if (args.scenario is not None):
scenario = args.scenario
# Radio Name
if (args.radio is not None):
radio = args.radio
# Number of Stations
if (args.num_sta is None):
num_sta = 0
if (args.num_sta is not None):
num_sta = args.num_sta
# Station ID
if (args.sta_id is None):
sta_id = '0'
if (args.sta_id is not None):
sta_id = args.sta_id
# SSID
if (args.ssid is not None):
ssid = args.ssid
if (args.ssid is not None):
ssid = args.ssid
args = parser.parse_args()
# Lanforge Manager IP Address
if (args.manager is None):
manager = "localhost"
if (args.manager != None):
manager = args.manager
# Security (Open by Default)
if (args.security is None):
security = 'open'
if (args.security is not None):
security = args.security
# Scenario Name
if (args.scenario != None):
scenario = args.scenario
# Radio Name
if (args.radio != None):
radio = args.radio
# Password (if Security is not Open)
if (args.password is not None):
password = args.password
if (args.password is 'open'):
password = "[Blank]"
if (args.password is None):
password = "[Blank]"
# Upstream Port (By default br0000)
if (args.upstream is None):
upstream = 'br0000'
if (args.upstream is not None):
upstream = args.upstream
# Protocol (By Default lf_udp)
if (args.protocol is not None):
protocol = args.protocol
if (args.protocol is None):
protocol = 'lf_udp'
#Min BPS
if (args.min_mbps is not None):
min_bps = int(args.min_mbps)*1000000
if (args.min_mbps is None):
min_bps = int(1000)*1000000
if (args.max_mbps is None ):
max_bps = int(1000)*1000000
# Number of Stations
if (args.num_sta is None):
num_sta = 0
if (args.num_sta != None):
num_sta = args.num_sta
if (args.min_mbps is not None):
min_bps = int(args.min_mbps)*1000000
if (args.max_mbps is not None and args.max_mbps is not "same"):
max_bps = int(args.max_mbps)*1000000
if (args.max_mbps is not None and args.max_mbps is "same"):
max_bps = args.min_mbps
if (args.duration is not None):
duration = (args.duration * 60)/5
if (args.report_name is not None):
report_name = args.report_name
if (args.duration is None):
duration = (1 * 60)/5
if (args.report_name is None):
report_name = "report.xlsx"
# Station ID
if (args.sta_id is None):
sta_id = '0'
if (args.sta_id != None):
sta_id = args.sta_id
# SSID
if (args.ssid != None):
ssid = args.ssid
if (args.ssid != None):
ssid = args.ssid
# Security (Open by Default)
if (args.security is None):
security = 'open'
if (args.security != None):
security = args.security
# Password (if Security is not Open)
if (args.password != None):
password = args.password
if (args.password == 'open'):
password = "[Blank]"
if (args.password is None):
password = "[Blank]"
# Upstream Port (By default br0000)
if (args.upstream is None):
upstream = 'br0000'
if (args.upstream != None):
upstream = args.upstream
# Protocol (By Default lf_udp)
if (args.protocol != None):
protocol = args.protocol
if (args.protocol is None):
protocol = 'lf_udp'
#Min BPS
if (args.min_mbps != None):
min_bps = int(args.min_mbps)*1000000
if (args.min_mbps is None):
min_bps = int(1000)*1000000
if (args.max_mbps is None ):
max_bps = int(1000)*1000000
if (args.min_mbps != None):
min_bps = int(args.min_mbps)*1000000
if (args.max_mbps != None and args.max_mbps != "same"):
max_bps = int(args.max_mbps)*1000000
if (args.max_mbps != None and args.max_mbps == "same"):
max_bps = args.min_mbps
if (args.duration != None):
duration = (args.duration * 60)/5
if (args.report_name != None):
report_name = args.report_name
if (args.duration is None):
duration = (1 * 60)/5
if (args.report_name is None):
report_name = "report.xlsx"
except Exception as e:
logging.exception(e)
exit(2)
logging.exception(e)
exit(2)
# Start DUT
#Loading the Scenario on Lanforge_1 (Here Considered as DUT) [Created VAP With SSID 'lexusap' on wiphy0 with eth1 as backhaul]
Scenario_1 = LoadScenario("192.168.200.18", 8080, "Lexus_DUT")
dut_traffic_profile = CreateSTA_CX("192.168.200.18", 8080, "wiphy1", 1, 0, 'lanforge_ap', 'open', password, 'br0000', 'lf_udp', min_bps, max_bps)
dut_traffic_profile.build()
print("DUT All Set... Lets setup Lanforge")
#Loading the Scenario on Lanforge_2 (Here Considered as LANFORGE Test) [Created VAP With SSID 'lanforge_ap' on wiphy0 with eth2 as backhaul]
DB_Lanforge_2 = "LANforge_TEST"
Scenario_2 = LoadScenario(manager, 8080, scenario)
lf_traffic_profile = CreateSTA_CX(manager, 8080, radio, num_sta, sta_id, ssid, security, password, upstream, protocol, min_bps, max_bps)
lf_traffic_profile.build()
@@ -456,7 +451,7 @@ def main():
#List for Storing the Total Throughput
throughput_sta =[]
throughput_vap =[]
# This loop will get the Data from All the endpoints and sum up to give total Throughput over time
for i in range(0,int(duration)):
temp=0
@@ -473,9 +468,9 @@ def main():
dut_traffic_profile.cleanup()
lf_traffic_profile.cleanup()
GenerateReport(throughput_sta, throughput_vap, dut_temp_obj.data_core1, dut_temp_obj.data_core2, duration, report_name)
if __name__ == '__main__':
main()

8
py-scripts/lf_ftp.py
View File

@@ -15,12 +15,10 @@ if 'py-json' not in sys.path:
from LANforge import LFUtils
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
import realm
from realm import Realm, PortUtils
import argparse
import datetime
from datetime import datetime
import time
import os
import matplotlib.patches as mpatches
from lf_report import *
from lf_graph import *
@@ -52,10 +50,10 @@ class FtpTest(LFCliBase):
self.duration = duration
self.traffic_duration = traffic_duration
self.ssh_port = ssh_port
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.local_realm = Realm(lfclient_host=self.host, lfclient_port=self.port)
self.station_profile = self.local_realm.new_station_profile()
self.cx_profile = self.local_realm.new_http_profile()
self.port_util = realm.PortUtils(self.local_realm)
self.port_util = PortUtils(self.local_realm)
self.cx_profile.requests_per_ten = self.requests_per_ten
print("Test is Initialized")

6
py-scripts/lf_ftp_test.py
View File

@@ -38,7 +38,7 @@ if 'py-json' not in sys.path:
from LANforge import LFUtils
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
import realm
from realm import Realm, PortUtils
import argparse
from datetime import datetime
import time
@@ -64,10 +64,10 @@ class ftp_test(LFCliBase):
self.band=band
self.file_size=file_size
self.direction=direction
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.local_realm = Realm(lfclient_host=self.host, lfclient_port=self.port)
self.station_profile = self.local_realm.new_station_profile()
self.cx_profile = self.local_realm.new_http_profile()
self.port_util = realm.PortUtils(self.local_realm)
self.port_util = PortUtils(self.local_realm)
self.cx_profile.requests_per_ten = self.requests_per_ten
print("Test is Initialized")

3
py-scripts/lf_graph.py
View File

@@ -19,11 +19,8 @@ INCLUDE_IN_README
'''
import matplotlib.pyplot as plt
import matplotlib as mpl
import numpy as np
import pandas as pd
import pdfkit
import math
from matplotlib.colors import ListedColormap
from lf_csv import lf_csv

4
py-scripts/lf_multipsk.py
View File

@@ -33,9 +33,7 @@ if sys.version_info[0] != 3:
if 'py-json' not in sys.path:
sys.path.append(os.path.join(os.path.abspath('..'), 'py-json'))
from LANforge.lfcli_base import LFCliBase
from LANforge import LFUtils
import realm
from realm import Realm
@@ -66,7 +64,7 @@ class MultiPsk(Realm):
self.resource = resource
self.sta_prefix = sta_prefix
self.debug = debug_
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.local_realm = Realm(lfclient_host=self.host, lfclient_port=self.port)
self.station_profile = self.local_realm.new_station_profile()
def build(self):

3
py-scripts/lf_sniff_radio.py
View File

@@ -25,7 +25,6 @@ import argparse
import time
from LANforge.LFUtils import *
from realm import Realm
import realm
class SniffRadio(Realm):
@@ -42,7 +41,7 @@ class SniffRadio(Realm):
self.lfclient_host = lfclient_host
self.lfclient_port = lfclient_port
self.debug = debug_on_
self.local_realm = realm.Realm(lfclient_host=self.lfclient_host,
self.local_realm = Realm(lfclient_host=self.lfclient_host,
lfclient_port=self.lfclient_port,
debug_=self.debug)
self.monitor = self.local_realm.new_wifi_monitor_profile()

6
py-scripts/lf_webpage.py
View File

@@ -17,9 +17,7 @@ if 'py-json' not in sys.path:
sys.path.append('../py-json')
from LANforge import LFUtils
from LANforge.LFUtils import *
import realm
from realm import Realm
from realm import PortUtils
from realm import Realm, PortUtils
from lf_report import *
from lf_graph import *
@@ -44,7 +42,7 @@ class HttpDownload(Realm):
self.debug = _debug_on
print(bands)
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.local_realm = Realm(lfclient_host=self.host, lfclient_port=self.port)
self.station_profile = self.local_realm.new_station_profile()
self.http_profile = self.local_realm.new_http_profile()
self.http_profile.requests_per_ten = self.target_per_ten

2
py-scripts/run_cv_scenario.py
View File

@@ -14,10 +14,8 @@ if 'py-json' not in sys.path:
import argparse
from LANforge import LFUtils
from LANforge import lfcli_base
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
import realm
from realm import Realm
"""

3
py-scripts/sta_connect.py
View File

@@ -16,11 +16,8 @@ if 'py-json' not in sys.path:
import argparse
from LANforge import LFUtils
# from LANforge import LFCliBase
from LANforge import lfcli_base
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
import realm
from realm import Realm
import time

2
py-scripts/sta_connect2.py
View File

@@ -391,7 +391,7 @@ class StaConnect2(LFCliBase):
def main():
lfjson_host = "localhost"
lfjson_port = 8080
parser = argparse.ArgumentParser(
parser = Realm.create_basic_argparse(
description="""LANforge Unit Test: Connect Station to AP
Example:
./sta_connect2.py --dest 192.168.100.209 --dut_ssid OpenWrt-2 --dut_bssid 24:F5:A2:08:21:6C

5
py-scripts/test_client_admission.py
View File

@@ -12,10 +12,7 @@ import time
if 'py-json' not in sys.path:
sys.path.append('../py-json')
from LANforge import LFUtils
from LANforge import lfcli_base
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
import realm
from realm import Realm
@@ -33,7 +30,7 @@ class LoadLayer3(Realm):
self.name_prefix = name_prefix
self.upstream = upstream
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.local_realm = Realm(lfclient_host=self.host, lfclient_port=self.port)
self.station_profile = self.local_realm.new_station_profile()
self.station_profile.ssid = self.ssid
self.station_profile.ssid_pass = self.paswd,

6
py-scripts/test_fileio.py
View File

@@ -40,13 +40,11 @@ if 'py-json' not in sys.path:
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
from LANforge import LFUtils
from LANforge import add_file_endp
from LANforge.add_file_endp import *
import argparse
import realm
from realm import Realm
import time
import datetime
import pprint
import os
@@ -138,7 +136,7 @@ class FileIOTest(LFCliBase):
# self.min_write_rate_bps = self.parse_size_bps(min_write_rate_bps)
#self.max_write_rate_bps = self.parse_size_bps(max_write_rate_bps)
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.local_realm = Realm(lfclient_host=self.host, lfclient_port=self.port)
self.wo_profile = self.local_realm.new_fio_endp_profile()
self.mvlan_profile = self.local_realm.new_mvlan_profile()

5
py-scripts/test_ip_connection.py
View File

@@ -32,10 +32,9 @@ if sys.version_info[0] != 3:
if 'py-json' not in sys.path:
sys.path.append(os.path.join(os.path.abspath('..'), 'py-json'))
import LANforge
from LANforge.lfcli_base import LFCliBase
from LANforge import LFUtils
import realm
from realm import Realm
import time
import pprint
@@ -61,7 +60,7 @@ class ConnectTest(LFCliBase):
super().__init__(_host,
_port,
_proxy_str=_proxy_str,
_local_realm=realm.Realm(lfclient_host=_host,
_local_realm=Realm(lfclient_host=_host,
lfclient_port=_port,
_exit_on_error=_exit_on_error,
_exit_on_fail=_exit_on_fail,

6
py-scripts/test_ipv4_ps.py
View File

@@ -13,7 +13,7 @@ if 'py-json' not in sys.path:
import argparse
from LANforge.lfcli_base import LFCliBase
from LANforge import LFUtils
import realm
from realm import Realm, WifiMonitor
import time
import datetime
@@ -40,12 +40,12 @@ class IPV4VariableTime(LFCliBase):
self.debug = _debug_on
self.name_prefix = name_prefix
self.test_duration = test_duration
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.local_realm = Realm(lfclient_host=self.host, lfclient_port=self.port)
self.station_profile = self.local_realm.new_station_profile()
self.cx_profile = self.local_realm.new_l3_cx_profile()
self.vap_profile = self.local_realm.new_vap_profile()
self.vap_profile.vap_name = "vap0000"
self.monitor = realm.WifiMonitor(self.lfclient_url, self.local_realm, debug_=_debug_on)
self.monitor = WifiMonitor(self.lfclient_url, self.local_realm, debug_=_debug_on)
self.station_profile.lfclient_url = self.lfclient_url
self.station_profile.ssid = self.ssid

5
py-scripts/test_ipv4_ttls.py
View File

@@ -11,10 +11,9 @@ if sys.version_info[0] != 3:
if 'py-json' not in sys.path:
sys.path.append(os.path.join(os.path.abspath('..'), 'py-json'))
import LANforge
from LANforge.lfcli_base import LFCliBase
from LANforge import LFUtils
import realm
from realm import Realm
import time
import pprint
from test_ip_variable_time import IPVariableTime
@@ -109,7 +108,7 @@ class TTLSTest(LFCliBase):
self.timeout = 120
self.number_template = number_template
self.debug = _debug_on
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.local_realm = Realm(lfclient_host=self.host, lfclient_port=self.port)
self.station_profile = self.local_realm.new_station_profile()
self.vap = vap
self.upstream_port = "eth1"

4
py-scripts/test_l3_WAN_LAN.py
View File

@@ -13,7 +13,7 @@ if 'py-json' not in sys.path:
import argparse
from LANforge.lfcli_base import LFCliBase
from LANforge import LFUtils
import realm
from realm import Realm
import time
import datetime
@@ -42,7 +42,7 @@ class VRTest(LFCliBase):
self.debug = _debug_on
self.name_prefix = name_prefix
self.test_duration = test_duration
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.local_realm = Realm(lfclient_host=self.host, lfclient_port=self.port)
self.station_profile = self.local_realm.new_station_profile()
self.cx_profile = self.local_realm.new_l3_cx_profile()
self.vr_profile = self.local_realm.new_vr_profile()

16
py-scripts/test_l3_powersave_traffic.py
View File

@@ -10,12 +10,10 @@ if sys.version_info[0] != 3:
if 'py-json' not in sys.path:
sys.path.append(os.path.join(os.path.abspath('..'), 'py-json'))
import argparse
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
from LANforge import LFUtils
import l3_cxprofile
import realm
from l3_cxprofile import L3CXProfile
from realm import StationProfile, WifiMonitor, Realm
import time
import datetime
@@ -36,28 +34,28 @@ class L3PowersaveTraffic(LFCliBase):
self.sta_list = station_list
self.prefix = prefix
self.debug = _debug_on
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port, debug_=False)
self.local_realm = Realm(lfclient_host=self.host, lfclient_port=self.port, debug_=False)
# upload
self.cx_prof_upload = l3_cxprofile.L3CXProfile(self.host, self.port, self.local_realm,
self.cx_prof_upload = L3CXProfile(self.host, self.port, self.local_realm,
side_a_min_bps=side_a_min_rate, side_b_min_bps=0,
side_a_max_bps=side_a_max_rate, side_b_max_bps=0,
side_a_min_pdu=pdu_size, side_a_max_pdu=pdu_size,
side_b_min_pdu=0, side_b_max_pdu=0, debug_=False)
# download
self.cx_prof_download = l3_cxprofile.L3CXProfile(self.host, self.port, self.local_realm,
self.cx_prof_download = L3CXProfile(self.host, self.port, self.local_realm,
side_a_min_bps=0, side_b_min_bps=side_b_min_rate,
side_a_max_bps=0, side_b_max_bps=side_b_max_rate,
side_a_min_pdu=0, side_a_max_pdu=0,
side_b_min_pdu=pdu_size, side_b_max_pdu=pdu_size, debug_=False)
self.test_duration = test_duration
self.station_profile = realm.StationProfile(self.lfclient_url, self.local_realm, ssid=self.ssid,
self.station_profile = StationProfile(self.lfclient_url, self.local_realm, ssid=self.ssid,
ssid_pass=self.password,
security=self.security, number_template_=self.prefix, mode=0,
up=True,
dhcp=True,
debug_=False)
self.new_monitor = realm.WifiMonitor(self.lfclient_url, self.local_realm, debug_=_debug_on)
self.new_monitor = WifiMonitor(self.lfclient_url, self.local_realm, debug_=_debug_on)
def build(self):
self.station_profile.use_security("open", ssid=self.ssid, passwd=self.password)

4
py-scripts/test_l3_unicast_traffic_gen.py
View File

@@ -13,7 +13,7 @@ if 'py-json' not in sys.path:
import argparse
from LANforge.lfcli_base import LFCliBase
from LANforge import LFUtils
import realm
from realm import Realm
import time
import datetime
@@ -44,7 +44,7 @@ class L3VariableTimeLongevity(LFCliBase):
self.radios = radios # from the command line
self.radio_list = radio_name_list
self.number_of_stations_per_radio_list = number_of_stations_per_radio_list
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.local_realm = Realm(lfclient_host=self.host, lfclient_port=self.port)
self.cx_profile = self.local_realm.new_l3_cx_profile()
self.station_profiles = []

6
py-scripts/test_l4.py
View File

@@ -65,11 +65,9 @@ if 'py-json' not in sys.path:
import argparse
from LANforge.lfcli_base import LFCliBase
from LANforge import LFUtils
import realm
import time
import datetime
from realm import TestGroupProfile
from port_utils import PortUtils
from realm import Realm, PortUtils
class IPV4L4(LFCliBase):
@@ -119,7 +117,7 @@ class IPV4L4(LFCliBase):
self.num_tests = int(num_tests)
self.target_requests_per_ten = int(target_requests_per_ten)
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.local_realm = Realm(lfclient_host=self.host, lfclient_port=self.port)
self.station_profile = self.local_realm.new_station_profile()
self.cx_profile = self.local_realm.new_l4_cx_profile()

2
py-scripts/test_wpa_passphrases.py
View File

@@ -7,6 +7,8 @@ Copyright 2021 Candela Technologies Inc
License: Free to distribute and modify. LANforge systems must be licensed.
'''
import sys
import os
if sys.version_info[0] != 3:
print("This script requires Python 3")

5
py-scripts/testgroup.py
View File

@@ -10,9 +10,8 @@ if 'py-json' not in sys.path:
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
from LANforge import LFUtils
import argparse
import realm
from realm import Realm
class TestGroup(LFCliBase):
def __init__(self, host, port,
@@ -24,7 +23,7 @@ class TestGroup(LFCliBase):
list_groups=None,
show_group=None):
self.local_realm = realm.Realm(lfclient_host=host, lfclient_port=port)
self.local_realm = Realm(lfclient_host=host, lfclient_port=port)
self.tg_profile = self.local_realm.new_test_group_profile()
if group_name is None and list_groups is None and (tg_action is not None or cx_action is not None or
add_cx_list is not None or rm_cx_list is not None or show_group is not None):

6
py-scripts/tip_station_powersave.py
View File

@@ -14,7 +14,7 @@ import argparse
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
from LANforge import LFUtils
import realm
from realm import Realm, PacketFilter
import time
import datetime
@@ -69,8 +69,8 @@ class TIPStationPowersave(LFCliBase):
self.powersave_sta_radio = powersave_station_radio_
self.sta_mac_map = {}
self.debug = debug_on_
self.packet_filter = realm.PacketFilter()
self.local_realm = realm.Realm(lfclient_host=self.host,
self.packet_filter = PacketFilter()
self.local_realm = Realm(lfclient_host=self.host,
lfclient_port=self.port,
debug_=self.debug)

3
py-scripts/vap_stations_example.py
View File

@@ -8,12 +8,9 @@ if sys.version_info[0] != 3:
exit(1)
if 'py-json' not in sys.path:
sys.path.append('../py-json')
import traceback
from LANforge import LFUtils
from LANforge.LFUtils import *
from LANforge.lfcli_base import LFCliBase
from generic_cx import GenericCx
mgrURL = "http://localhost:8080/"
staName = "sta0"

20
py-scripts/wlan_capacity_calculator.py
View File

@@ -12,19 +12,19 @@ import os
from pip._internal.utils import logging
if 'py-json' not in sys.path:
sys.path.append(os.path.join(os.path.abspath('..'), 'py-json'))
import wlan_theoretical_sta
if 'py_json' not in sys.path:
sys.path.append(os.path.join(os.path.abspath('..'), 'py_json'))
from wlan_theoretical_sta import abg11_calculator, n11_calculator, ac11_calculator
# main method
def main():
parse = wlan_theoretical_sta.abg11_calculator.create_argparse( prog='wlan_capacity_calculator.py',
formatter_class=argparse.RawTextHelpFormatter,
epilog='''\
parse = abg11_calculator.create_argparse( prog='wlan_capacity_calculator.py',
formatter_class=argparse.RawTextHelpFormatter,
epilog='''\
This python script calculates the theoretical value of three different stations( 11abg/11n/11ac)''',
description='''\
description='''\
wlan_capacity_calculator.py
---------------------------------------------------------------------------
@@ -226,20 +226,20 @@ def main():
# Select station(802.11a/b/g/n/ac standards)
if "11abg" in Calculator_name:
Station1 = wlan_theoretical_sta.abg11_calculator(traffic_name, phy_name, encryption_name, qos_name, mac_name, basic_name,
Station1 = abg11_calculator(traffic_name, phy_name, encryption_name, qos_name, mac_name, basic_name,
preamble_name, slot_name, codec_name, rts_name, cts_name)
Station1.calculate()
Station1.get_result()
if "11n" in Calculator_name:
Station2 = wlan_theoretical_sta.n11_calculator(traffic_name, data_name, channel_name, guard_name, highest_name, encryption_name,
Station2 = n11_calculator(traffic_name, data_name, channel_name, guard_name, highest_name, encryption_name,
qos_name, ip_name,
mc_name, basic_name, mac_name,
codec_name, plcp_name, cwin_name, rts_name, cts_name)
Station2.calculate()
Station2.get_result()
if "11ac" in Calculator_name:
Station3 = wlan_theoretical_sta.ac11_calculator(traffic_name, data_name, spatial_name, channel_name, guard_name, highest_name,
Station3 = ac11_calculator(traffic_name, data_name, spatial_name, channel_name, guard_name, highest_name,
encryption_name
, qos_name, ip_name, mc_name, basic_name, mac_name,
codec_name, cwin_name, rtscts_name)