Using the realm.py library we will write a script that will allow us to automate the creation of stations and Layer 4 cross connects. We will also be able to start and stop traffic over the cross connects using the script. Station and Cross Connect creation is covered in the Realm Scripting Cookbook. Requires LANforge 5.4.2. | |
requests_per_ten = 600
will set our URL request rate to 1 per second. There is no limit to what can be used as the rate but common rates are: l4_cx_profile.url = "dl http://10.40.0.1 /dev/null"
class IPV4L4(LFCliBase):
def __init__(self, host, port, ssid, security, password, url, requests_per_ten, station_list,
target_requests_per_ten=600, number_template="00000", resource=1, num_tests=1,
_debug_on=False,
_exit_on_error=False,
_exit_on_fail=False):
super().__init__(host, port, _debug=_debug_on, _halt_on_error=_exit_on_error, _exit_on_fail=_exit_on_fail)
self.host = host
self.port = port
self.ssid = ssid
self.security = security
self.password = password
self.url = url
self.requests_per_ten = requests_per_ten
self.number_template = number_template
self.sta_list = station_list
self.resource = resource
self.num_tests = num_tests
self.target_requests_per_ten = target_requests_per_ten
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.cx_profile = self.local_realm.new_l4_cx_profile()
self.cx_profile.url = self.url
self.cx_profile.requests_per_ten = self.requests_per_ten
# Station Profile init
def build(self):
# Build stations
self.station_profile.use_security(self.security, self.ssid, self.password)
print("Creating stations")
self.station_profile.create(resource=1, radio="wiphy0", sta_names_=self.sta_list, debug=self.debug)
temp_sta_list = []
for station in range(len(self.sta_list)):
temp_sta_list.append(str(self.resource) + "." + self.sta_list[station])
self.l4_profile.create(ports=temp_sta_list, sleep_time=.5, debug_=self.debug, suppress_related_commands_=None)
def start(self, print_pass=False, print_fail=False):
temp_stas = self.sta_list.copy()
temp_stas.append("eth1")
cur_time = datetime.datetime.now()
interval_time = cur_time + datetime.timedelta(minutes=1)
passes = 0
expected_passes = 0
self.station_profile.admin_up(1)
self.local_realm.wait_for_ip(self.resource, temp_stas)
self.l4_profile.start_cx()
print("Starting test")
for test in range(self.num_tests):
expected_passes += 1
while cur_time < interval_time:
time.sleep(1)
cur_time = datetime.datetime.now()
if self.l4_profile.check_errors(self.debug):
if self.__check_request_rate():
passes += 1
else:
self._fail("FAIL: Request rate did not exceed 90% target rate", print_fail)
break
else:
self._fail("FAIL: Errors found getting to %s " % self.url, print_fail)
break
interval_time = cur_time + datetime.timedelta(minutes=1)
if passes == expected_passes:
self._pass("PASS: All tests passes", print_pass)