jbr_jag_test.py: incorporates attempting to use a station profile in test

StationProfile does not have sufficient logic to put LFJsonGet or LFJsonPost yet

Signed-off-by: Jed Reynolds <jed@candelatech.com>
This commit is contained in:
Jed Reynolds
2021-08-17 15:43:52 -07:00
parent 2301be272f
commit b1ac042179

View File

@@ -22,10 +22,73 @@ if sys.version_info[0] != 3:
sys.path.insert(1, "../../py-json")
import argparse
import pprint
import traceback
from LANforge import lf_json_autogen
from LANforge.lf_json_autogen import LFJsonGet as LFG
from LANforge.lf_json_autogen import LFJsonPost as LFP
# import LANforge.lfcli_base
# from LANforge.lfcli_base import LFCliBase
from realm import Realm
from station_profile import StationProfile
# ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- #
class TestStation(Realm):
def __init__(self,
host="localhost",
port=8080,
_debug_on=True,
_exit_on_error=True,
_exit_on_fail=False):
super().__init__(lfclient_host=host,
lfclient_port=port,
debug_=_debug_on,
_exit_on_error=_exit_on_error,
_exit_on_fail=_exit_on_fail)
def run(self, get_request: LFG = None, post_request: LFP = None):
station_profile = StationProfile(
"http://%s:%s" % (self.lfclient_host, self.lfclient_port),
self,
ssid="jedway-r7800-5G",
ssid_pass="jedway-r7800-5G",
security="wpa2",
number_template_="000",
mode=0,
up=True,
resource=1,
shelf=1,
dhcp=True,
debug_=self.debug,
use_ht160=False)
print("Checking for previous station:")
try:
response = get_request.get_port(eid_list="1.1.sta000", requested_col_names=['_links', 'alias'],
debug_=self.debug)
if response:
pprint.pprint(response)
print("Deleting previous station:")
post_request.post_rm_vlan(shelf=1, resource=1, port="sta000", debug_=self.debug)
else:
print("Previous station not seen")
print("Creating station:")
station_profile.create(radio="1.1.wiphy0",
num_stations=1,
sta_names_=['sta000'],
debug=True)
print("Created station:")
response = get_request.get_port("1.1.sta000")
if response:
pprint.pprint(response)
else:
print("Station did not get created.")
except Exception as x:
traceback.print_tb(x)
print(x.__repr__())
exit(1)
# ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- #
@@ -41,23 +104,29 @@ def main():
if not args.test:
print("No test requested")
exit(1)
if args.test.endswith("get_port"):
test_get_port(args)
if args.test.endswith("set_port"):
test_set_port(args)
# ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- #
def test_get_port(args=None):
print("test_get_port")
if not args:
raise ValueError("test_get_port needs args")
post_request = lf_json_autogen.LFJsonPost(lfclient_host=args.host,
lfclient_port=8080,
debug_=True,
_exit_on_error=True)
get_request = LFG(lfclient_host=args.host,
lfclient_port=8080,
debug_=True,
_exit_on_error=True)
if args.test.endswith("get_port"):
test_get_port(args, get_request=get_request, post_request=post_request)
if args.test.endswith("set_port"):
test_set_port(args, get_request=get_request, post_request=post_request)
# ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- #
def test_get_port(args=None,
get_request: LFG = None,
post_request: LFP = None):
print("test_get_port")
if not args:
raise ValueError("test_get_port needs args")
result = get_request.get_port(eid_list=["1.1.eth0", "1.1.eth1", "1.1.eth2"],
requested_col_names=(),
debug_=True)
@@ -65,38 +134,27 @@ def test_get_port(args=None):
# ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- #
def test_set_port(args=None):
def test_set_port(args=None,
get_request: LFG = None,
post_request: LFP = None):
print("test_set_port")
if not args:
raise ValueError("test_set_port needs args")
post_request = lf_json_autogen.LFJsonPost(lfclient_host=args.host,
lfclient_port=8080,
debug_=True,
_exit_on_error=True)
my_current_flags = 0
my_interest_flags = 0;
my_interest_flags = 0
try:
my_current_flags = LFP.set_flags(LFP.SetPortCurrentFlags,
0,
['if_down', 'use_dhcp'])
except Exception as x:
import traceback
traceback.print_tb(x)
print(x.__repr__())
exit(1)
try:
my_current_flags = LFP.set_flags(LFP.SetPortCurrentFlags,
0,
my_current_flags,
[
LFP.SetPortCurrentFlags.if_down,
LFP.SetPortCurrentFlags.use_dhcp
])
except Exception as x:
import traceback
traceback.print_tb(x)
print(x.__repr__())
exit(1)
try:
my_interest_flags = LFP.set_flags(LFP.SetPortInterest,
0,
[
@@ -104,18 +162,16 @@ def test_set_port(args=None):
'ifdown',
'dhcp'
])
result = post_request.post_set_port(alias=None, # A user-defined name for this interface. Can be BLANK or NA.
current_flags=my_current_flags, # See above, or NA.
current_flags_msk=my_current_flags,
mac='NA',
# This sets 'interest' for flags 'Enable RADIUS service' and higher. See above, or NA.
interest=my_interest_flags,
port='eth2', # Port number for the port to be modified.
report_timer=2000,
resource=1, # Resource number for the port to be modified.
shelf=1, # Shelf number for the port to be modified.
debug_=True)
post_request.post_set_port(current_flags=my_current_flags, # See above, or NA.
current_flags_msk=my_current_flags,
mac='NA',
# This sets 'interest' for flags 'Enable RADIUS service' and higher. See above, or NA.
interest=my_interest_flags,
port='eth2', # Port number for the port to be modified.
report_timer=2000,
resource=1, # Resource number for the port to be modified.
shelf=1, # Shelf number for the port to be modified.
debug_=True)
my_current_flags = LFP.clear_flags(LFP.SetPortCurrentFlags,
my_current_flags,
@@ -135,25 +191,20 @@ def test_set_port(args=None):
LFP.SetPortInterest.ip_Mask,
])
result = post_request.post_set_port(alias=None, # A user-defined name for this interface. Can be BLANK or NA.
current_flags=my_current_flags, # See above, or NA.
current_flags_msk=my_current_flags,
mac='NA',
ip_addr='10.32.23.1',
netmask='255.255.255.0',
gateway='0.0.0.0',
# This sets 'interest' for flags 'Enable RADIUS service' and higher. See above, or NA.
interest=my_interest_flags,
port='eth2', # Port number for the port to be modified.
report_timer=2000,
resource=1, # Resource number for the port to be modified.
shelf=1, # Shelf number for the port to be modified.
debug_=True)
get_request = LFG(lfclient_host=args.host,
lfclient_port=8080,
debug_=True,
_exit_on_error=True)
post_request.post_set_port(alias=None, # A user-defined name for this interface. Can be BLANK or NA.
current_flags=my_current_flags, # See above, or NA.
current_flags_msk=my_current_flags,
mac='NA',
ip_addr='10.32.23.1',
netmask='255.255.255.0',
gateway='0.0.0.0',
# This sets 'interest' for flags 'Enable RADIUS service' and higher. See above, or NA.
interest=my_interest_flags,
port='eth2', # Port number for the port to be modified.
report_timer=2000,
resource=1, # Resource number for the port to be modified.
shelf=1, # Shelf number for the port to be modified.
debug_=True)
result = get_request.get_port(eid_list="1.1.eth2",
requested_col_names=["_links",
@@ -165,15 +216,27 @@ def test_set_port(args=None):
"PORT_SUPPORTED_FLAGS_L",
"PORT_SUPPORTED_FLAGS_H",
"PORT_CUR_FLAGS_L",
"PORT_CUR_FLAGS_H" ],
"PORT_CUR_FLAGS_H"],
debug_=True)
pprint.pprint(result)
except Exception as x:
import traceback
traceback.print_tb(x)
print(x.__repr__())
exit(1)
# ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- #
# Create a station using the station_profile object
# ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- #
try:
station_test = TestStation(host=args.host,
port=8080,
_debug_on=False)
station_test.run(get_request=get_request, post_request=post_request)
except Exception as x:
traceback.print_tb(x)
print(x.__repr__())
exit(1)
# ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- #