Update tests infra

tests/conftest.py

@@ -70,9 +70,8 @@ def settings():
 @pytest.fixture(scope="session")
 def api():
     # handle to make API calls
-    api = snappi.api(
-        location=utl.settings.location, ext=utl.settings.ext
-    )
+    api = snappi.api(location=utl.settings.location, ext=utl.settings.ext)
+    utl.configure_credentials(api, utl.settings.username, utl.settings.psd)
     yield api
     if getattr(api, "assistant", None) is not None:
         api.assistant.Session.remove()
@@ -82,9 +81,10 @@ def api():
 def cvg_api():
     # handle to make Convergence API calls
     api = snappi_convergence.api(
-        location=utl.settings.location, ext=utl.settings.ext
+        location=utl.settings.location,
+        ext=utl.settings.ext,
     )
-
+    api.configure_credentials(utl.settings.username, utl.settings.psd)
     yield api
     if getattr(api, "assistant", None) is not None:
         api.assistant.Session.remove()

tests/settings.json

@@ -1,5 +1,6 @@
 {
-    "username": "common",
+    "username": "admin",
+    "psd": "admin",
     "location": "https://localhost:443",
     "ports": [
         "localhost;1;1",

tests/utils/common.py

@@ -61,6 +61,11 @@ def load_dict_from_json_file(path):
         return json.load(fp, object_hook=byteify)
 
 
+def configure_credentials(api, usr, psd):
+    api.username = usr
+    api.password = psd
+
+
 class Settings(object):
     """
     Singleton for global settings
@@ -69,6 +74,7 @@ class Settings(object):
     def __init__(self):
         # these not be defined and are here only for documentation
         self.username = None
+        self.psd = None
         self.location = None
         self.ports = None
         self.speed = None
@@ -123,40 +129,42 @@ def start_traffic(api, cfg, start_capture=True):
     capture_names = get_capture_port_names(cfg)
     if capture_names and start_capture:
         print("Starting capture on ports %s ..." % str(capture_names))
-        cs = api.capture_state()
-        cs.state = cs.START
-        api.set_capture_state(cs)
+        cs = api.control_state()
+        cs.port.capture.state = cs.port.capture.START
+        api.set_control_state(cs)
+
     print("Starting all protocols ...")
-    ps = api.protocol_state()
-    ps.state = ps.START
-    api.set_protocol_state(ps)
+    cs = api.control_state()
+    cs.protocol.all.state = cs.protocol.all.START
+    api.set_control_state(cs)
+
 
     print("Starting transmit on all flows ...")
-    ts = api.transmit_state()
-    ts.state = ts.START
-    api.set_transmit_state(ts)
+    cs = api.control_state()
+    cs.traffic.flow_transmit.state = cs.traffic.flow_transmit.START
+    api.set_control_state(cs)
 
 
 def stop_traffic(api, cfg, stop_capture=True):
     """
     Stops flows
     """
     print("Stopping transmit on all flows ...")
-    ts = api.transmit_state()
-    ts.state = ts.STOP
-    api.set_transmit_state(ts)
+    cs = api.control_state()
+    cs.traffic.flow_transmit.state = cs.traffic.flow_transmit.STOP
+    api.set_control_state(cs)
 
-    print("Starting all protocols ...")
-    ps = api.protocol_state()
-    ps.state = ps.STOP
-    api.set_protocol_state(ps)
+    print("Stopping all protocols ...")
+    cs = api.control_state()
+    cs.protocol.all.state = cs.protocol.all.STOP
+    api.set_control_state(cs)
 
     capture_names = get_capture_port_names(cfg)
     if capture_names and stop_capture:
         print("Stopping capture on ports %s ..." % str(capture_names))
-        cs = api.capture_state()
-        cs.state = cs.STOP
-        api.set_capture_state(cs)
+        cs = api.control_state()
+        cs.port.capture.state = cs.port.capture.STOP
+        api.set_control_state(cs)
 
 
 def seconds_elapsed(start_seconds):