tests/netstack/ToggleWlanInterfaceStressTest.py

#!/usr/bin/env python3
#
# Copyright 2022 The Fuchsia Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import logging
import time

from mobly import asserts, signals, test_runner

from antlion import base_test
from antlion.controllers import fuchsia_device
from antlion.controllers.fuchsia_device import FuchsiaDevice
from antlion.test_utils.abstract_devices.wlan_device import (
    AssociationMode,
    create_wlan_device,
)


class ToggleWlanInterfaceStressTest(base_test.AntlionBaseTest):
    def setup_class(self):
        self.log = logging.getLogger()
        self.fuchsia_devices: list[FuchsiaDevice] = self.register_controller(
            fuchsia_device
        )

        asserts.abort_class_if(
            len(self.fuchsia_devices) == 0, "Requires at least one Fuchsia device"
        )

        self.dut = create_wlan_device(self.fuchsia_devices[0], AssociationMode.POLICY)

    def test_iface_toggle_and_ping(self):
        """Test that we don't error out when toggling WLAN interfaces.

        Steps:
        1. Find a WLAN interface
        2. Destroy it
        3. Create a new WLAN interface
        4. Ping after association
        5. Repeat 1-4 1,000 times

        Expected Result:
        Verify there are no errors in destroying the wlan interface.

        Returns:
          signals.TestPass if no errors
          signals.TestFailure if there are any errors during the test.

        TAGS: WLAN, Stability
        Priority: 1
        """

        # Test assumes you've already connected to some AP.

        for i in range(1000):
            wlan_interfaces = self.dut.get_wlan_interface_id_list()
            self.log.info(wlan_interfaces)

            if len(wlan_interfaces) < 1:
                raise signals.TestFailure("Not enough wlan interfaces for test")

            self.dut.destroy_wlan_interface(wlan_interfaces[0])
            # Really make sure it is dead
            self.fuchsia_devices[0].ssh.run(f"wlan iface del {wlan_interfaces[0]}")

            # Grace period
            time.sleep(2)

            self.fuchsia_devices[0].ssh.run("wlan iface new --phy 0 --role Client")
            end_time = time.time() + 300

            while time.time() < end_time:
                time.sleep(1)
                if self.dut.is_connected():
                    try:
                        ping_result = self.dut.ping("8.8.8.8", 10, 1000, 1000, 25)
                        print(ping_result)
                    except Exception as err:
                        # TODO: Once we gain more stability, fail test when pinging fails
                        print(f"some err {err}")
                    time.sleep(2)  # give time for some traffic
                    break
            if not self.dut.is_connected():
                raise signals.TestFailure(f"Failed at iteration {i + 1}")
            self.log.info(f"Iteration {i + 1} successful")
        raise signals.TestPass("Success")


if __name__ == "__main__":
    test_runner.main()