mirror of
https://github.com/Telecominfraproject/wlan-lanforge-scripts.git
synced 2025-12-24 22:37:03 +00:00
jbr_jag_test.py: updated to work with agreed pathing
- corrects import statements to work from sandbox directory - corrects renamed debug parameters - now uses LFSession Signed-off-by: Jed Reynolds <jed@bitratchet.com>
This commit is contained in:
Jed Reynolds
@@ -11,7 +11,7 @@ To enable using lf_json_autogen in other parts of the codebase, set LF_USE_AUTOG
|
||||
$ LF_USE_AUTOGEN=1 ./jbr_jag_test.py --test set_port --host ct521a-lion
|
||||
|
||||
NOTES:
|
||||
|
||||
this has not been worked on in a while, many parameters have changed
|
||||
|
||||
TO DO NOTES:
|
||||
|
||||
@@ -21,35 +21,42 @@ import sys
|
||||
if sys.version_info[0] != 3:
|
||||
print("This script requires Python3")
|
||||
exit()
|
||||
import os
|
||||
|
||||
sys.path.append(os.path.join(os.path.abspath(__file__ + "/../../../")))
|
||||
import importlib
|
||||
|
||||
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
|
||||
|
||||
lanforge_client = importlib.import_module("lanforge_client")
|
||||
|
||||
import lanforge_client.lanforge_api
|
||||
from lanforge_client.lanforge_api import LFSession
|
||||
from lanforge_client.lanforge_api import LFJsonQuery as LFGet
|
||||
from lanforge_client.lanforge_api import LFJsonCommand as LFPost
|
||||
|
||||
realm = importlib.import_module("py-json.realm")
|
||||
Realm = realm.Realm
|
||||
station_profile = importlib.import_module("py-json.station_profile")
|
||||
StationProfile = station_profile.StationProfile
|
||||
|
||||
# ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- #
|
||||
class TestStation(Realm):
|
||||
def __init__(self,
|
||||
host="localhost",
|
||||
port=8080,
|
||||
_debug_on=True,
|
||||
debug_=True,
|
||||
_exit_on_error=True,
|
||||
_exit_on_fail=False):
|
||||
super().__init__(lfclient_host=host,
|
||||
lfclient_port=port,
|
||||
debug_=_debug_on,
|
||||
debug_=debug_,
|
||||
_exit_on_error=_exit_on_error,
|
||||
_exit_on_fail=_exit_on_fail)
|
||||
|
||||
def run(self, get_request: LFG = None, post_request: LFP = None):
|
||||
def run(self, get_request: LFGet = None, post_request: LFPost = None):
|
||||
station_profile = StationProfile(
|
||||
"http://%s:%s" % (self.lfclient_host, self.lfclient_port),
|
||||
self,
|
||||
@@ -67,12 +74,12 @@ class TestStation(Realm):
|
||||
|
||||
print("Checking for previous station:")
|
||||
try:
|
||||
response = get_request.get_port(eid_list="1.1.sta000", requested_col_names=['_links', 'alias'],
|
||||
debug_=self.debug)
|
||||
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)
|
||||
post_request.post_rm_vlan(shelf=1, resource=1, port="sta000", debug=self.debug)
|
||||
else:
|
||||
print("Previous station not seen")
|
||||
|
||||
@@ -82,7 +89,7 @@ class TestStation(Realm):
|
||||
sta_names_=['sta000'],
|
||||
debug=True)
|
||||
print("Created station:")
|
||||
response = get_request.get_port("1.1.sta000")
|
||||
response = get_request.get_port(eid_list=["1.1.sta000"])
|
||||
if response:
|
||||
pprint.pprint(response)
|
||||
else:
|
||||
@@ -107,14 +114,15 @@ def main():
|
||||
if not args.test:
|
||||
print("No test requested")
|
||||
exit(1)
|
||||
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)
|
||||
|
||||
session = LFSession(lfclient_url=args.host,
|
||||
debug=True,
|
||||
connection_timeout_sec=1.0,
|
||||
require_session=True,
|
||||
exit_on_error=True)
|
||||
|
||||
post_request = LFPost(session_obj=session)
|
||||
get_request = LFGet(session_obj=session)
|
||||
|
||||
if args.test.endswith("get_port"):
|
||||
test_get_port(args, get_request=get_request, post_request=post_request)
|
||||
@@ -124,22 +132,22 @@ def main():
|
||||
|
||||
# ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- #
|
||||
def test_get_port(args=None,
|
||||
get_request: LFG = None,
|
||||
post_request: LFP = None):
|
||||
get_request: LFGet = None,
|
||||
post_request: LFPost = 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)
|
||||
debug=True)
|
||||
pprint.pprint(result)
|
||||
|
||||
|
||||
# ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- ---- #
|
||||
def test_set_port(args=None,
|
||||
get_request: LFG = None,
|
||||
post_request: LFP = None):
|
||||
get_request: LFGet = None,
|
||||
post_request: LFPost = None):
|
||||
print("test_set_port")
|
||||
if not args:
|
||||
raise ValueError("test_set_port needs args")
|
||||
@@ -147,20 +155,20 @@ def test_set_port(args=None,
|
||||
my_current_flags = 0
|
||||
my_interest_flags = 0
|
||||
try:
|
||||
my_current_flags = LFP.set_flags(LFP.SetPortCurrentFlags,
|
||||
0,
|
||||
['if_down', 'use_dhcp'])
|
||||
my_current_flags = LFPost.set_flags(LFPost.SetPortCurrentFlags,
|
||||
0,
|
||||
['if_down', 'use_dhcp'])
|
||||
|
||||
my_current_flags = LFP.set_flags(LFP.SetPortCurrentFlags,
|
||||
my_current_flags,
|
||||
[
|
||||
LFP.SetPortCurrentFlags.if_down,
|
||||
LFP.SetPortCurrentFlags.use_dhcp
|
||||
my_current_flags = LFPost.set_flags(LFPost.SetPortCurrentFlags,
|
||||
my_current_flags,
|
||||
[
|
||||
LFPost.SetPortCurrentFlags.if_down,
|
||||
LFPost.SetPortCurrentFlags.use_dhcp
|
||||
])
|
||||
|
||||
my_interest_flags = LFP.set_flags(LFP.SetPortInterest,
|
||||
0,
|
||||
[
|
||||
my_interest_flags = LFPost.set_flags(LFPost.SetPortInterest,
|
||||
0,
|
||||
[
|
||||
'current_flags',
|
||||
'ifdown',
|
||||
'dhcp'
|
||||
@@ -174,24 +182,24 @@ def test_set_port(args=None,
|
||||
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)
|
||||
debug=True)
|
||||
|
||||
my_current_flags = LFP.clear_flags(LFP.SetPortCurrentFlags,
|
||||
my_current_flags,
|
||||
flag_names=LFP.SetPortCurrentFlags.use_dhcp)
|
||||
my_current_flags = LFP.clear_flags(LFP.SetPortCurrentFlags,
|
||||
my_current_flags,
|
||||
flag_names=[LFP.SetPortCurrentFlags.if_down])
|
||||
my_current_flags = LFPost.clear_flags(LFPost.SetPortCurrentFlags,
|
||||
my_current_flags,
|
||||
flag_names=LFPost.SetPortCurrentFlags.use_dhcp)
|
||||
my_current_flags = LFPost.clear_flags(LFPost.SetPortCurrentFlags,
|
||||
my_current_flags,
|
||||
flag_names=[LFPost.SetPortCurrentFlags.if_down])
|
||||
|
||||
my_interest_flags = LFP.set_flags(LFP.SetPortInterest,
|
||||
0,
|
||||
[
|
||||
my_interest_flags = LFPost.set_flags(LFPost.SetPortInterest,
|
||||
0,
|
||||
[
|
||||
'current_flags',
|
||||
'ifdown',
|
||||
'dhcp',
|
||||
LFP.SetPortInterest.ip_address,
|
||||
LFP.SetPortInterest.ip_gateway,
|
||||
LFP.SetPortInterest.ip_Mask,
|
||||
LFPost.SetPortInterest.ip_address,
|
||||
LFPost.SetPortInterest.ip_gateway,
|
||||
LFPost.SetPortInterest.ip_Mask,
|
||||
])
|
||||
|
||||
post_request.post_set_port(alias=None, # A user-defined name for this interface. Can be BLANK or NA.
|
||||
@@ -207,7 +215,7 @@ def test_set_port(args=None,
|
||||
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)
|
||||
debug=True)
|
||||
|
||||
result = get_request.get_port(eid_list="1.1.eth2",
|
||||
requested_col_names=["_links",
|
||||
@@ -220,7 +228,7 @@ def test_set_port(args=None,
|
||||
"PORT_SUPPORTED_FLAGS_H",
|
||||
"PORT_CUR_FLAGS_L",
|
||||
"PORT_CUR_FLAGS_H"],
|
||||
debug_=True)
|
||||
debug=True)
|
||||
pprint.pprint(result)
|
||||
except Exception as x:
|
||||
traceback.print_tb(x)
|
||||
|
||||
Reference in New Issue
Block a user