diff --git a/py-scripts/sandbox/jbr_jag_test.py b/py-scripts/sandbox/jbr_jag_test.py index 44f088e5..ed0efd73 100755 --- a/py-scripts/sandbox/jbr_jag_test.py +++ b/py-scripts/sandbox/jbr_jag_test.py @@ -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)