Merge branch 'master' into feature/wifi-3162--add-clustersysteminfo-check

.github/workflows/ow_sdk_docker-compose.yml

@@ -0,0 +1,11 @@
+name: 2.x SDK testing with Docker Compose deployment
+
+on:
+  workflow_dispatch:
+
+jobs:
+  test:
+    runs-on: ubuntu-20.04
+    steps:
+      - name: Run echo
+        run: echo "This is a dummy workflow."

libs/apnos/apnos.py

@@ -337,8 +337,11 @@ class APNOS:
                 active = output.decode('utf-8').splitlines()[4].split(":")[1].replace(" ", "").replace(",", "")
             client.close()
         except Exception as e:
-            pytest.exit("ubus call ucentral status: error" + output)
-            print(e)
+            if output.__contains__('"connected":'):
+                pass
+            else:
+                pytest.exit("ubus call ucentral status: error" + str(output))
+                print(e)
             connected, latest, active = "Error", "Error", "Error"
         return connected, latest, active
 

libs/lanforge/lf_tests.py

@@ -763,7 +763,7 @@ if __name__ == '__main__':
         "influx_tag": ["basic-03", "ec420"],
     }
     lanforge_data = {
-                "ip": "192.168.200.80",
+                "ip": "192.168.200.10",
                 "port": 8080,
                 "ssh_port": 22,
                 "2.4G-Radio": ["wiphy0"],
@@ -777,8 +777,20 @@ if __name__ == '__main__':
                 "AX-Station-Name": "ax"
             }
     obj = RunTest(lanforge_data=lanforge_data, debug=False, influx_params=influx_params)
-    obj.Client_Connect(ssid="ssid_wpa_5g_br", passkey="something", security="wpa", station_name=['sta0000'])
-    obj.dataplane(station_name=["sta0000"])
+    upstream = lanforge_data['upstream']
+    data = obj.staConnect.json_get("/port/all")
+    for i in data["interfaces"]:
+        if list(i.keys())[0] == "1.1.eth1.10":
+            print(i)
+    # print(dict(list(data['interfaces'])).keys())
+    # print(obj.staConnect.json_get("/port/" + upstream.split(".")[0] +
+    #                         "/" + upstream.split(".")[1] +
+    #                         "/" + upstream.split(".")[2] + "/" + "10"))
+    # print("/port/" + upstream.split(".")[0] +
+    #                         "/" + upstream.split(".")[1] +
+    #                         "/" + upstream.split(".")[2] + "/" + "100")
+    # obj.Client_Connect(ssid="ssid_wpa_5g_br", passkey="something", security="wpa", station_name=['sta0000'])
+    # obj.dataplane(station_name=["sta0000"])
     # a = obj.staConnect.json_get("/events/since/")
     # print(a)
     # print(obj.eap_connect.json_get("port/1/1/sta0000?fields=ap,ip"))

libs/lanforge/lf_tools.py

@@ -18,7 +18,7 @@ sys.path.append(f"../lanforge/lanforge-scripts/py-scripts/tip-cicd-sanity")
 sys.path.append(f'../libs')
 sys.path.append(f'../libs/lanforge/')
 import allure
-
+from sta_connect2 import StaConnect2
 from create_chamberview import CreateChamberview
 from create_chamberview_dut import DUT
 import time
@@ -88,6 +88,7 @@ class ChamberView:
             self.exit_on_error = False
             self.dut_idx_mapping = {}
             self.ssid_list = []
+            self.staConnect = StaConnect2(self.lanforge_ip, self.lanforge_port, debug_=self.debug)
             self.raw_line = [
             ["profile_link " + self.upstream_resources + " upstream-dhcp 1 NA NA " + self.upstream_port.split(".")
             [2] + ",AUTO -1 NA"],
@@ -148,10 +149,17 @@ class ChamberView:
         return self.CreateChamberview, self.scenario_name
 
     def add_vlan(self, vlan_ids=[]):
+        data = self.staConnect.json_get("/port/all")
+        flag = 0
         for vlans in vlan_ids:
-            self.raw_line.append(["profile_link 1.1 " + "vlan-100 1 " + self.upstream_port
-                                  + " NA " + self.upstream_port.split(".")[2] + ",AUTO -1 " + str(vlans)])
-        self.Chamber_View()
+            for i in data["interfaces"]:
+                if list(i.keys())[0] != self.upstream_port + "." + str(vlans):
+                    flag = 1
+        if flag == 1:
+            for vlans in vlan_ids:
+                self.raw_line.append(["profile_link 1.1 " + "vlan-100 1 " + self.upstream_port
+                                      + " NA " + self.upstream_port.split(".")[2] + ",AUTO -1 " + str(vlans)])
+            self.Chamber_View()
 
     def add_stations(self, band="2G", num_stations="max", dut="NA", ssid_name=[]):
         idx = 0

tests/e2e/basic/conftest.py

@@ -47,10 +47,9 @@ def create_lanforge_chamberview_dut(lf_tools):
 
 
 @pytest.fixture(scope="class")
-def setup_profiles(request, setup_controller, testbed, get_equipment_ref, fixtures_ver,
+def setup_profiles(request, setup_controller, testbed, get_equipment_ref, fixtures_ver, reset_scenario_lf,
                    instantiate_profile, get_markers, create_lanforge_chamberview_dut, lf_tools,
                    get_security_flags, get_configuration, radius_info, get_apnos, radius_accounting_info):
-    lf_tools.reset_scenario()
     param = dict(request.param)
 
     # VLAN Setup
@@ -84,8 +83,6 @@ def setup_profiles(request, setup_controller, testbed, get_equipment_ref, fixtur
     yield return_var
 
 
-
-
 @pytest.fixture(scope="session")
 def station_names_twog(request, get_configuration):
     station_names = []
@@ -101,6 +98,7 @@ def station_names_fiveg(request, get_configuration):
         station_names.append(get_configuration["traffic_generator"]["details"]["5G-Station-Name"] + "0" + str(i))
     yield station_names
 
+
 @pytest.fixture(scope="session")
 def station_names_ax(request, get_configuration):
     station_names = []
@@ -108,6 +106,7 @@ def station_names_ax(request, get_configuration):
         station_names.append(get_configuration["traffic_generator"]["details"]["AX-Station-Name"] + "0" + str(i))
     yield station_names
 
+
 @pytest.fixture(scope="session")
 def num_stations(request):
     num_sta = int(request.config.getini("num_stations"))
@@ -131,3 +130,14 @@ def get_vlan_list(get_apnos, get_configuration):
     vlan_list = list(ap_ssh.get_vlan())
     vlan_list.sort()
     yield vlan_list
+
+
+@pytest.fixture(scope="session")
+def reset_scenario_lf(request, lf_tools):
+    lf_tools.reset_scenario()
+
+    def teardown_session():
+        lf_tools.reset_scenario()
+
+    request.addfinalizer(teardown_session)
+    yield ""

tests/test_connectivity.py

@@ -52,7 +52,7 @@ class TestResources(object):
 class TestFMS(object):
 
     @pytest.mark.get_firmware_list
-    def test_fms_version_list(self, fixtures_ver, get_configuration):
+    def test_fms_version_list(self, fixtures_ver, get_configuration, get_ap_logs):
         PASS = []
         for ap in get_configuration['access_point']:
             # get the latest branch
@@ -135,18 +135,18 @@ class TestFMS(object):
         assert False not in PASS
 
     @pytest.mark.firmware_upgrade
-    def test_firmware_upgrade_request(self, firmware_upgrade):
+    def test_firmware_upgrade_request(self, firmware_upgrade, get_ap_logs):
         for update in firmware_upgrade:
             allure.attach(name='serial: ' + update[0], body="")
         assert True
 
     @pytest.mark.test_firmware_ap
-    def test_firmware_upgrade_status_AP(self, firmware_upgrade):
+    def test_firmware_upgrade_status_AP(self, firmware_upgrade, get_ap_logs):
         allure.attach(name="firmware Upgrade Status:", body="")
         assert True
 
     @pytest.mark.test_firmware_gw
-    def test_firmware_upgrade_status_gateway(self, get_apnos, get_configuration, setup_controller):
+    def test_firmware_upgrade_status_gateway(self, get_apnos, get_configuration, setup_controller, get_ap_logs):
         status = []
         for ap in get_configuration['access_point']:
             ap_ssh = get_apnos(ap, pwd="../libs/apnos/", sdk="2.x")