packages/antlion/unit_tests/controllers/iperf_server_test.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 os
import unittest

import mock

from antlion.controllers import iperf_server
from antlion.controllers.iperf_server import (
    IPerfServer,
    IPerfServerOverAdb,
    IPerfServerOverSsh,
)
from antlion.controllers.utils_lib.ssh import settings

# The position in the call tuple that represents the args array.
ARGS = 0

# The position in the call tuple that represents the kwargs dict.
KWARGS = 1

MOCK_LOGFILE_PATH = "/path/to/foo"


class IPerfServerModuleTest(unittest.TestCase):
    """Tests the antlion.controllers.iperf_server module."""

    def test_create_creates_local_iperf_server_with_int(self):
        self.assertIsInstance(
            iperf_server.create([12345])[0],
            IPerfServer,
            "create() failed to create IPerfServer for integer input.",
        )

    def test_create_creates_local_iperf_server_with_str(self):
        self.assertIsInstance(
            iperf_server.create(["12345"])[0],
            IPerfServer,
            "create() failed to create IPerfServer for integer input.",
        )

    def test_create_cannot_create_local_iperf_server_with_bad_str(self):
        with self.assertRaises(ValueError):
            iperf_server.create(["12345BAD_STRING"])

    @mock.patch("antlion.controllers.iperf_server.utils")
    def test_create_creates_server_over_ssh_with_ssh_config_and_port(self, _):
        self.assertIsInstance(
            iperf_server.create(
                [
                    {
                        "ssh_config": {
                            "user": "",
                            "host": "",
                            "identity_file": "/dev/null",
                        },
                        "port": "",
                    }
                ]
            )[0],
            IPerfServerOverSsh,
            "create() failed to create IPerfServerOverSsh for a valid config.",
        )

    def test_create_creates_server_over_adb_with_proper_config(self):
        self.assertIsInstance(
            iperf_server.create([{"AndroidDevice": "53R147", "port": 0}])[0],
            IPerfServerOverAdb,
            "create() failed to create IPerfServerOverAdb for a valid config.",
        )

    def test_create_raises_value_error_on_bad_config_dict(self):
        with self.assertRaises(ValueError):
            iperf_server.create([{"AndroidDevice": "53R147", "ssh_config": {}}])

    def test_get_port_from_ss_output_returns_correct_port_ipv4(self):
        ss_output = (
            "tcp LISTEN  0 5 127.0.0.1:<PORT>  *:*" ' users:(("cmd",pid=<PID>,fd=3))'
        )
        self.assertEqual(
            iperf_server._get_port_from_ss_output(ss_output, "<PID>"), "<PORT>"
        )

    def test_get_port_from_ss_output_returns_correct_port_ipv6(self):
        ss_output = (
            "tcp LISTEN  0 5 ff:ff:ff:ff:ff:ff:<PORT>  *:*"
            ' users:(("cmd",pid=<PID>,fd=3))'
        )
        self.assertEqual(
            iperf_server._get_port_from_ss_output(ss_output, "<PID>"), "<PORT>"
        )


class IPerfServerBaseTest(unittest.TestCase):
    """Tests antlion.controllers.iperf_server.IPerfServerBase."""

    @mock.patch("os.makedirs")
    def test_get_full_file_path_creates_parent_directory(self, mock_makedirs):
        # Will never actually be created/used.
        logging.log_path = "/tmp/unit_test_garbage"

        server = IPerfServer("port")

        full_file_path = server._get_full_file_path()

        self.assertTrue(mock_makedirs.called, "Did not attempt to create a directory.")
        self.assertEqual(
            os.path.dirname(full_file_path),
            mock_makedirs.call_args[ARGS][0],
            "The parent directory of the full file path was not created.",
        )


class IPerfServerTest(unittest.TestCase):
    """Tests antlion.controllers.iperf_server.IPerfServer."""

    PID = 123456

    def setUp(self):
        iperf_server._get_port_from_ss_output = lambda *_: IPerfServerTest.PID

    @mock.patch("builtins.open")
    @mock.patch("antlion.controllers.iperf_server.subprocess")
    @mock.patch("antlion.controllers.iperf_server.job")
    def test_start_makes_started_true(self, mock_job, __, ___):
        """Tests calling start() without calling stop() makes started True."""
        server = IPerfServer("port")
        server._get_full_file_path = lambda _: MOCK_LOGFILE_PATH
        server.start()

        self.assertTrue(server.started)

    @mock.patch("builtins.open")
    @mock.patch("antlion.controllers.iperf_server.subprocess")
    @mock.patch("antlion.controllers.iperf_server.job")
    def test_start_stop_makes_started_false(self, _, __, ___):
        """Tests calling start() without calling stop() makes started True."""
        server = IPerfServer("port")
        server._get_full_file_path = lambda _: MOCK_LOGFILE_PATH

        server.start()
        server.stop()

        self.assertFalse(server.started)

    @mock.patch("builtins.open")
    @mock.patch("antlion.controllers.iperf_server.subprocess")
    @mock.patch("antlion.controllers.iperf_server.job")
    def test_start_sets_current_log_file(self, _, __, ___):
        server = IPerfServer("port")
        server._get_full_file_path = lambda _: MOCK_LOGFILE_PATH

        server.start()

        self.assertEqual(
            server._current_log_file,
            MOCK_LOGFILE_PATH,
            "The _current_log_file was not received from _get_full_file_path.",
        )

    @mock.patch("builtins.open")
    @mock.patch("antlion.controllers.iperf_server.subprocess")
    def test_stop_returns_current_log_file(self, _, __):
        server = IPerfServer("port")
        server._get_full_file_path = lambda _: MOCK_LOGFILE_PATH
        server._current_log_file = MOCK_LOGFILE_PATH
        server._iperf_process = mock.Mock()

        log_file = server.stop()

        self.assertEqual(
            log_file,
            MOCK_LOGFILE_PATH,
            "The _current_log_file was not returned by stop().",
        )

    @mock.patch("builtins.open")
    @mock.patch("antlion.controllers.iperf_server.subprocess")
    @mock.patch("antlion.controllers.iperf_server.job")
    def test_start_does_not_run_two_concurrent_processes(self, start_proc, _, __):
        server = IPerfServer("port")
        server._get_full_file_path = lambda _: MOCK_LOGFILE_PATH
        server._iperf_process = mock.Mock()

        server.start()

        self.assertFalse(
            start_proc.called,
            "start() should not begin a second process if another is running.",
        )

    @mock.patch("antlion.utils.stop_standing_subprocess")
    def test_stop_exits_early_if_no_process_has_started(self, stop_proc):
        server = IPerfServer("port")
        server._get_full_file_path = lambda _: MOCK_LOGFILE_PATH
        server._iperf_process = None

        server.stop()

        self.assertFalse(
            stop_proc.called,
            "stop() should not kill a process if no process is running.",
        )


class IPerfServerOverSshTest(unittest.TestCase):
    """Tests antlion.controllers.iperf_server.IPerfServerOverSsh."""

    INIT_ARGS = [
        settings.from_config(
            {"host": "TEST_HOST", "user": "test", "identity_file": "/dev/null"}
        ),
        "PORT",
    ]

    @mock.patch("antlion.controllers.iperf_server.connection")
    def test_start_makes_started_true(self, _):
        """Tests calling start() without calling stop() makes started True."""
        server = IPerfServerOverSsh(*self.INIT_ARGS)
        server._ssh_session = mock.Mock()
        server._cleanup_iperf_port = mock.Mock()
        server._get_full_file_path = lambda _: MOCK_LOGFILE_PATH

        server.start()

        self.assertTrue(server.started)

    @mock.patch("builtins.open")
    @mock.patch("antlion.controllers.iperf_server.connection")
    def test_start_stop_makes_started_false(self, _, __):
        """Tests calling start() without calling stop() makes started True."""
        server = IPerfServerOverSsh(*self.INIT_ARGS)
        server._ssh_session = mock.Mock()
        server._cleanup_iperf_port = mock.Mock()
        server._get_full_file_path = lambda _: MOCK_LOGFILE_PATH

        server.start()
        server.stop()

        self.assertFalse(server.started)

    @mock.patch("builtins.open")
    @mock.patch("antlion.controllers.iperf_server.connection")
    def test_stop_returns_expected_log_file(self, _, __):
        server = IPerfServerOverSsh(*self.INIT_ARGS)
        server._ssh_session = mock.Mock()
        server._cleanup_iperf_port = mock.Mock()
        server._get_full_file_path = lambda _: MOCK_LOGFILE_PATH
        server._iperf_pid = mock.Mock()

        log_file = server.stop()

        self.assertEqual(
            log_file,
            MOCK_LOGFILE_PATH,
            "The expected log file was not returned by stop().",
        )

    @mock.patch("antlion.controllers.iperf_server.connection")
    def test_start_does_not_run_two_concurrent_processes(self, _):
        server = IPerfServerOverSsh(*self.INIT_ARGS)
        server._ssh_session = mock.Mock()
        server._cleanup_iperf_port = mock.Mock()
        server._get_full_file_path = lambda _: MOCK_LOGFILE_PATH
        server._iperf_pid = mock.Mock()

        server.start()

        self.assertFalse(
            server._ssh_session.run_async.called,
            "start() should not begin a second process if another is running.",
        )

    @mock.patch("antlion.utils.stop_standing_subprocess")
    @mock.patch("antlion.controllers.iperf_server.connection")
    def test_stop_exits_early_if_no_process_has_started(self, _, __):
        server = IPerfServerOverSsh(*self.INIT_ARGS)
        server._ssh_session = mock.Mock()
        server._cleanup_iperf_port = mock.Mock()
        server._get_full_file_path = lambda _: MOCK_LOGFILE_PATH
        server._iperf_pid = None

        server.stop()

        self.assertFalse(
            server._ssh_session.run_async.called,
            "stop() should not kill a process if no process is running.",
        )


class IPerfServerOverAdbTest(unittest.TestCase):
    """Tests antlion.controllers.iperf_server.IPerfServerOverSsh."""

    ANDROID_DEVICE_PROP = (
        "antlion.controllers.iperf_server." "IPerfServerOverAdb._android_device"
    )

    @mock.patch(ANDROID_DEVICE_PROP)
    def test_start_makes_started_true(self, mock_ad):
        """Tests calling start() without calling stop() makes started True."""
        server = IPerfServerOverAdb("53R147", "PORT")
        server._get_full_file_path = lambda _: MOCK_LOGFILE_PATH
        mock_ad.adb.shell.return_value = "<PID>"

        server.start()

        self.assertTrue(server.started)

    @mock.patch("antlion.libs.proc.job.run")
    @mock.patch("builtins.open")
    @mock.patch(ANDROID_DEVICE_PROP)
    def test_start_stop_makes_started_false(self, mock_ad, _, __):
        """Tests calling start() without calling stop() makes started True."""
        server = IPerfServerOverAdb("53R147", "PORT")
        server._get_full_file_path = lambda _: MOCK_LOGFILE_PATH
        mock_ad.adb.shell.side_effect = ["<PID>", "", "", ""]

        server.start()
        server.stop()

        self.assertFalse(server.started)

    @mock.patch("antlion.libs.proc.job.run")
    @mock.patch("builtins.open")
    @mock.patch(ANDROID_DEVICE_PROP)
    def test_stop_returns_expected_log_file(self, mock_ad, _, __):
        server = IPerfServerOverAdb("53R147", "PORT")
        server._get_full_file_path = lambda _: MOCK_LOGFILE_PATH
        server._iperf_process = mock.Mock()
        server._iperf_process_adb_pid = "<PID>"
        mock_ad.adb.shell.side_effect = ["", "", ""]

        log_file = server.stop()

        self.assertEqual(
            log_file,
            MOCK_LOGFILE_PATH,
            "The expected log file was not returned by stop().",
        )

    @mock.patch(ANDROID_DEVICE_PROP)
    def test_start_does_not_run_two_concurrent_processes(self, android_device):
        server = IPerfServerOverAdb("53R147", "PORT")
        server._get_full_file_path = lambda _: MOCK_LOGFILE_PATH
        server._iperf_process = mock.Mock()

        server.start()

        self.assertFalse(
            android_device.adb.shell_nb.called,
            "start() should not begin a second process if another is running.",
        )

    @mock.patch("antlion.libs.proc.job.run")
    @mock.patch("builtins.open")
    @mock.patch(ANDROID_DEVICE_PROP)
    def test_stop_exits_early_if_no_process_has_started(self, android_device, _, __):
        server = IPerfServerOverAdb("53R147", "PORT")
        server._get_full_file_path = lambda _: MOCK_LOGFILE_PATH
        server._iperf_pid = None

        server.stop()

        self.assertFalse(
            android_device.adb.shell_nb.called,
            "stop() should not kill a process if no process is running.",
        )


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