# Copyright 2019 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Unit tests for VM."""

import fcntl
import logging
import multiprocessing
import os
import socket
import sys
from unittest import mock

from chromite.lib import constants
from chromite.lib import cros_test_lib
from chromite.lib import device
from chromite.lib import osutils
from chromite.lib import partial_mock
from chromite.lib import qemu
from chromite.lib import remote_access
from chromite.lib import vm


pytestmark = cros_test_lib.pytestmark_inside_only


# pylint: disable=protected-access
class VMTester(cros_test_lib.RunCommandTempDirTestCase):
    """Test vm.VM."""

    def setUp(self) -> None:
        """Common set up method for all tests."""
        # Pick a port that is valid, but we can't bind normally, and is unlikely
        # to be used in general.
        opts = vm.VM.GetParser().parse_args(["--ssh-port=1"])
        opts.enable_kvm = True
        with mock.patch.object(multiprocessing, "cpu_count", return_value=8):
            self._vm = vm.VM(opts)
        self._vm.use_sudo = False
        self._vm.board = "amd64-generic"
        self._vm.cache_dir = self.tempdir
        self._vm.image_path = self.TempFilePath(constants.TEST_IMAGE_BIN)
        osutils.Touch(self._vm.image_path)

        # This function is unittested in qemu_unittest, so mocking the basic API
        # should be safe for us.
        self.mock_qemu_install = self.PatchObject(
            qemu, "InstallFromCipd", side_effect=self._InstallQemu
        )

        self.nested_kvm_file = self.TempFilePath("kvm_intel_nested")
        osutils.WriteFile(self.nested_kvm_file, "N")
        # Make the glob match the path we just created for nested_kvm_file.
        self._vm.NESTED_KVM_GLOB = os.path.join(
            os.path.dirname(self.nested_kvm_file), "kvm_*_nested"
        )

        # Satisfy QEMU version check.
        version_str = (
            "QEMU emulator version 2.6.0, Copyright (c) "
            "2003-2008 Fabrice Bellard"
        )
        self.rc.AddCmdResult(partial_mock.In("--version"), stdout=version_str)

        self.ssh_port = self._vm.ssh_port

    def _InstallQemu(self, cache_dir):
        """Mock for installing qemu into the cache from cipd."""
        # Creates a fake SDK cache with the QEMU binary.
        path = cache_dir / "qemu-cipd" / "bin"
        path.mkdir(parents=True, exist_ok=True)

        for exe in ("qemu-system-x86_64", "qemu-img"):
            (path / exe).touch()

        return path

    def TempFilePath(self, file_path):
        return os.path.join(self.tempdir, file_path)

    def TempVMPath(self, kvm_file):
        return self.TempFilePath(
            os.path.join("cros_vm_%d" % self.ssh_port, kvm_file)
        )

    def FindPathInArgs(self, args, path):
        """Checks the called commands to see if the path is present.

        Args:
            args: List of called commands.
            path: Path to check if present in the called commands.

        Returns:
            Whether the path is found in the called commands.
        """
        for call in args:
            # A typical call looks like:
            # call(['.../chroot/usr/bin/qemu-system-x86_64', '--version'],
            #      capture_output=True)
            if any(path in a for a in call[0][0]):
                return True
        return False

    def testStart(self) -> None:
        self._vm.Start()
        self.assertCommandContains([self._vm.qemu_path])
        self.assertCommandContains(
            [
                "-m",
                "8G",
                "-smp",
                "8",
                "-daemonize",
                "-device",
                "usb-tablet",
            ]
        )
        self.assertCommandContains(
            [
                "-pidfile",
                self.TempVMPath("kvm.pid"),
                "-chardev",
                "pipe,id=control_pipe,path=%s" % self.TempVMPath("kvm.monitor"),
                "-serial",
                "file:%s" % self.TempVMPath("kvm.monitor.serial"),
                "-mon",
                "chardev=control_pipe",
            ]
        )
        self.assertCommandContains(
            [
                "-cpu",
                "Haswell-noTSX,vendor=GenuineIntel"
                ",-invpcid,-tsc-deadline,check",
            ]
        )
        self.assertCommandContains(
            [
                "-device",
                "virtio-net,netdev=eth0",
                "-netdev",
                "user,id=eth0,net=10.0.2.0/27,hostfwd=tcp:127.0.0.1:"
                f"{self.ssh_port}-:22",
            ]
        )
        self.assertCommandContains(
            [
                "-device",
                "virtio-scsi-pci,id=scsi",
                "-device",
                "scsi-hd,drive=hd,rotation_rate=1",
                "-drive",
                "if=none,id=hd,file=%s,cache=unsafe,format=raw"
                % self.TempFilePath(constants.TEST_IMAGE_BIN),
            ]
        )
        self.assertCommandContains(["-enable-kvm"])
        self.assertCommandContains(
            [
                "-vga",
                "virtio",
            ]
        )

    @mock.patch("chromite.lib.device.Device.WaitForBoot")
    def testStartRetriesSuccess(self, mock_wait) -> None:
        """Start() returns normally if WaitForBoot fails transiently once."""
        mock_wait.side_effect = (
            device.DeviceError("error"),
            True,
        )
        self._vm.Start()

    @mock.patch("chromite.lib.device.Device.WaitForBoot")
    def testStartRetriesFailure(self, mock_wait) -> None:
        """Start() raises a DeviceError if WaitForBoot fails all attempts."""
        mock_wait.side_effect = (
            device.DeviceError("error"),
            device.DeviceError("error"),
        )
        self.assertRaises(device.DeviceError, self._vm.Start)

    @mock.patch("chromite.lib.device.Device.WaitForBoot")
    def testNoWaitForBoot(self, mock_wait) -> None:
        """Start() does not wait for boot if wait_for_boot is False."""
        mock_wait.assert_not_called()
        self._vm.wait_for_boot = False
        self._vm.Start()

    def testStartWithVMX(self) -> None:
        """Verify vmx is enabled if the host supports nested virtualization."""
        osutils.WriteFile(self.nested_kvm_file, "1")
        self._vm.Start()
        self.assertCommandContains(
            [
                "-cpu",
                "Haswell-noTSX,vendor=GenuineIntel"
                ",-invpcid,-tsc-deadline,check,vmx=on,svm=on",
            ]
        )

    def testStop(self) -> None:
        pid = "12345"
        self.assertEqual(self._vm.pidfile, self.TempVMPath("kvm.pid"))
        osutils.WriteFile(self._vm.pidfile, pid)
        self._vm.Stop()
        self.assertCommandContains(["kill", "-9", pid])

    def testBuiltVMImagePath(self) -> None:
        """Verify locally built VM image path is picked up by vm.VM."""
        self._vm.image_path = None
        expected_vm_image_path = os.path.join(
            constants.SOURCE_ROOT,
            "src",
            "build",
            "images",
            self._vm.board,
            "latest",
            constants.TEST_IMAGE_BIN,
        )
        osutils.Touch(expected_vm_image_path, makedirs=True)
        self._vm.Start()
        self.assertTrue(
            self.FindPathInArgs(self.rc.call_args_list, expected_vm_image_path)
        )

    def testSDKVMImagePath(self) -> None:
        """Verify vm.VM picks up the downloaded VM in the SDK."""
        self._vm.image_path = None
        vm_image_dir = cros_test_lib.FakeSDKCache(
            self._vm.cache_dir
        ).CreateCacheReference(self._vm.board, constants.TEST_IMAGE_TAR)
        vm_image_path = os.path.join(vm_image_dir, constants.TEST_IMAGE_BIN)
        osutils.Touch(vm_image_path, makedirs=True)
        self._vm.Start()
        expected_vm_image_path = os.path.join(
            self._vm.cache_dir,
            "chrome-sdk/symlinks/%s+12225.0.0+%s/%s"
            % (
                self._vm.board,
                constants.TEST_IMAGE_TAR,
                constants.TEST_IMAGE_BIN,
            ),
        )
        self.assertTrue(
            self.FindPathInArgs(self.rc.call_args_list, expected_vm_image_path)
        )

    def testVMImageNotFound(self) -> None:
        """Verify VMError is raised when a fake board image cannot be found."""
        self._vm.image_path = None
        self._vm.board = "fake_board_name"
        self.assertRaises(vm.VMError, self._vm.Start)

    def testVMImageDoesNotExist(self) -> None:
        """Verify that VMError is raised when image path is not real."""
        self._vm.image_path = "/fake/path/to/the/vm/image"
        self.assertRaises(vm.VMError, self._vm.Start)

    def testAppendBinFile(self) -> None:
        """Verify bin file appended when image-path points to a directory."""
        self._vm.image_path = self.tempdir
        self._vm.Start()
        self.assertEqual(
            self._vm.image_path, self.TempFilePath(constants.TEST_IMAGE_BIN)
        )

    def testQemuCipdPath(self) -> None:
        """Verify vm.VM picks up the downloaded QEMU in the SDK."""
        qemu_exe = self._InstallQemu(self._vm.cache_dir) / "qemu-system-x86_64"
        self._vm.qemu_path = None
        self._vm._SetQemuPath()
        self.assertEqual(self._vm.qemu_path, str(qemu_exe))

    def testInvalidQemuBiosPath(self) -> None:
        """Verify that VMError is raised for nonexistent qemu bios path."""
        self._vm.qemu_bios_path = "/invalid/qemu/bios/path/"
        self.assertRaises(vm.VMError, self._vm.Start)

    def testCreateQcow2Image(self) -> None:
        """Tests that a qcow2 image is created with --copy-on-write."""
        self._vm.copy_on_write = True
        initial_img_path = self._vm.image_path
        self._vm.Start()

        # The command that creates the Qcow2 image.
        self.assertCommandContains(
            [
                self._vm.qemu_img_path,
                "create",
                "-f",
                "qcow2",
                "-o",
                "backing_file=%s,backing_fmt=raw" % initial_img_path,
                os.path.join(self._vm.vm_dir, "qcow2.img"),
            ]
        )
        # The command that launches a VM with the new Qcow2 image.
        self.assertCommandContains(
            [
                "-drive",
                "if=none,id=hd,file=%s,cache=unsafe,format=qcow2"
                % os.path.join(self._vm.vm_dir, "qcow2.img"),
            ]
        )

    def testQemuImageNotFound(self) -> None:
        """Veryify that VMError is raised for nonexistent qemu image path."""
        self._vm.copy_on_write = True
        self._vm.qemu_img_path = "/invalid/qemu/img/path/"
        self.assertRaises(vm.VMError, self._vm._SetQemuPath)

    def testRmVMDir(self) -> None:
        """Verify that the vm directory is removed after calling RmVMDir."""
        self.assertExists(self._vm.vm_dir)
        self._vm.use_sudo = False
        self._vm.Stop()
        self.assertNotExists(self._vm.vm_dir)

    @mock.patch("chromite.lib.osutils.SafeMakedirs", return_value=False)
    def testCreateVMDirError(self, make_dir_mock) -> None:
        """Verify an error is raised when vm_dir is not a valid directory."""
        self._vm.vm_dir = "/not/a/valid/dir"
        self.assertRaises(AssertionError, self._vm._CreateVMDir)
        make_dir_mock.assert_called()

    @mock.patch("chromite.lib.osutils.SafeMakedirs", return_value=False)
    def testCreateVMDirLinkError(self, make_dir_mock) -> None:
        """Verify that an error is raised when vm_dir is a symbolic link."""
        # Create the symlink.
        symlink = self.TempFilePath("symlink")
        os.symlink(self.TempFilePath("fakepath"), symlink)
        self._vm.vm_dir = symlink

        self.assertRaises(AssertionError, self._vm._CreateVMDir)
        make_dir_mock.assert_called()

    @mock.patch("chromite.lib.osutils.SafeMakedirs", return_value=False)
    @mock.patch("os.getuid")
    def testCreateVMDirStatError(self, getuid_mock, make_dir_mock) -> None:
        """Verify an error is raised  when user does not own the vm dir."""
        self.assertRaises(AssertionError, self._vm._CreateVMDir)
        getuid_mock.assert_called()
        make_dir_mock.assert_called()

    def testQemuVersionError(self) -> None:
        """Verify VMError is raised without an expected QEMU version number."""
        version_str = "Fake Version String"
        self.rc.AddCmdResult(partial_mock.In("--version"), stdout=version_str)
        self.assertRaises(vm.VMError, self._vm._SetQemuPath)

    def testQemuVersion(self) -> None:
        """Verify that the correct QEMU version is identified."""
        version_str = (
            "QEMU emulator version 2.8.0, Copyright (c) "
            "2003-2008 Fabrice Bellard"
        )
        self.rc.AddCmdResult(partial_mock.In("--version"), stdout=version_str)
        self._vm._SetQemuPath()
        self.assertEqual("2.8.0", self._vm.QemuVersion())
        self.assertCommandContains([self._vm.qemu_path, "--version"])

    def testCheckQemuError(self) -> None:
        """Verify that VMError is raised when the QEMU version is too old."""
        version_str = (
            "QEMU emulator version 2.5.0, Copyright (c) "
            "2003-2008 Fabrice Bellard"
        )
        self.rc.AddCmdResult(partial_mock.In("--version"), stdout=version_str)
        self.assertRaises(vm.VMError, self._vm._SetQemuPath)

    def testRunError(self) -> None:
        """Verify that VMError is raised when no action is specified."""
        self._vm.start = False
        self._vm.stop = False
        self._vm.cmd = None
        self.assertRaises(vm.VMError, self._vm.Run)

    def testIsRunningError(self) -> None:
        """Verify that VMError is raised when VM is not running."""
        self._vm.cmd = ["fake_command", "--test_cmd"]
        self.assertRaises(vm.VMError, self._vm.Run)

    @mock.patch("chromite.lib.vm.VM.IsRunning", return_value=True)
    def testRunRemoteCmd(self, is_running_mock) -> None:
        """Tests that the VM runs with a specific command."""
        self._vm.cmd = ["fake_command", "--test_cmd"]
        self._vm.Run()
        self.assertCommandContains(
            [
                "ssh",
                "-p",
                str(self.ssh_port),
                "root@localhost",
                "--",
                "fake_command",
                "--test_cmd",
            ]
        )
        is_running_mock.assert_called()

    def testGetVMPidDir(self) -> None:
        """Verify that isRunning is False with a nonexistent directory."""
        self._vm.vm_dir = "fake/directory"
        self.assertFalse(self._vm.IsRunning())

    def testGetVMPidFile(self) -> None:
        """Verify that isRunning is False with a nonexistent pid file."""
        self._vm.pidfile = "fake/pid/file"
        self.assertFalse(self._vm.IsRunning())

    def testPidString(self) -> None:
        """Verify that isRunning is False if the pid is not an integer."""
        osutils.WriteFile(self._vm.pidfile, "fake_pid")
        self.assertFalse(self._vm.IsRunning())

    def testGetVMPid(self) -> None:
        """Verify that a proper pid number kills the VM process."""
        # Using this process's pid to fake the VM's pid.
        pid = str(os.getpid())
        osutils.WriteFile(self._vm.pidfile, pid)
        self.assertTrue(self._vm.IsRunning())
        self._vm.Stop()
        self.assertCommandContains(["kill", "-9", pid])

    def testBiosPath(self) -> None:
        """Verify QEMU bios path."""
        self._vm.qemu_bios_path = self.TempFilePath("qemu/bios/path")
        osutils.SafeMakedirs(self._vm.qemu_bios_path)
        self._vm.Start()
        self.assertCommandContains(["-L", self._vm.qemu_bios_path])

    def testQemuHost(self) -> None:
        """Verify QEMU host forwarding."""
        self._vm.ssh_port = 1028
        self._vm.qemu_hostfwd = ["tcp:127.0.0.1:1024-:22"]
        self._vm.Start()
        self.assertCommandContains(
            [
                "-netdev",
                "user,id=eth0,net=10.0.2.0/27,"
                "hostfwd=tcp:127.0.0.1:1028-:22,hostfwd=tcp:127.0.0.1:1024-:22",
            ]
        )

    def testQemuArgs(self) -> None:
        """Verify QEMU arguments."""
        self._vm.qemu_args = ["-portrait", "-full-screen", "-no-reboot"]
        self._vm.Start()
        self.assertCommandContains(["-portrait", "-full-screen", "-no-reboot"])

    def testNoDisplay(self) -> None:
        """Check the command call's arguments when there is no display."""
        self._vm.display = False
        self._vm.Start()
        self.assertCommandContains(["-display", "none"])

    def testWaitForSSHPort(self) -> None:
        """Verify VM correctly waits on the SSH port if it is busy."""
        # Assigning an unused port to the VM SSH Port.
        self._vm.ssh_port = remote_access.GetUnusedPort()
        sock = socket.socket()
        sock.bind((remote_access.LOCALHOST_IP, self._vm.ssh_port))

        # Look for retry messages in output.
        with cros_test_lib.LoggingCapturer(log_level=logging.INFO) as logger:
            with self.assertRaises(vm.VMError) as ctx:
                self._vm._WaitForSSHPort(sleep=0)
            e = ctx.exception
        in_use_message = "SSH port %d in use" % self._vm.ssh_port
        self.assertEqual(in_use_message, str(e))
        self.assertTrue(logger.LogsMatch((in_use_message + "...\n") * 11))

        # Verify the VM works correctly when the port is not in use.
        # There should be no retries after the port is released.
        # Another process could grab the unused port between closing it
        # and calling _WaitForSSHPort but this is extremely unlikely.
        sock.close()
        with cros_test_lib.LoggingCapturer(log_level=logging.INFO) as logger:
            self._vm._WaitForSSHPort()
        self.assertEqual(logger.messages, "")

    @mock.patch(
        "chromite.lib.remote_access.RemoteDevice.GetRunningPids",
        return_value=[],
    )
    def testWaitForProcsError(self, pid_mocker) -> None:
        """Verify an error is raised when no chrome processes are running."""
        # Look for retry messages in output.
        with cros_test_lib.LoggingCapturer(log_level=logging.INFO) as logger:
            with self.assertRaises(vm.VMError) as ctx:
                self._vm._WaitForProcs(sleep=0)
            e = ctx.exception

        pid_message = "chrome pids: []\n"
        self.assertTrue(logger.LogsContain(pid_message * 6))
        self.assertIn(
            "_WaitForProcs failed: timed out while waiting for 8 chrome "
            "processes to start.",
            str(e),
        )
        pid_mocker.assert_called()

    @mock.patch(
        "chromite.lib.remote_access.RemoteDevice.GetRunningPids",
        return_value=[756, 905, 1065, 1092, 1096, 1171, 1180, 1181],
    )
    def testWaitForProcs(self, pid_mocker) -> None:
        """Verify VM waits for chrome processes to launch."""
        # Check the log output for expected chrome pids.
        with cros_test_lib.LoggingCapturer(log_level=logging.INFO) as logger:
            self._vm._WaitForProcs(sleep=0)
        self.assertEqual(
            logger.messages,
            "chrome pids: " "[756, 905, 1065, 1092, 1096, 1171, 1180, 1181]\n",
        )
        pid_mocker.assert_called()

    @mock.patch("chromite.lib.vm.VM._WaitForProcs")
    @mock.patch("chromite.lib.device.Device.WaitForBoot")
    @mock.patch("chromite.lib.vm.VM.Start")
    def testWaitForBoot(self, start_mock, boot_mock, procs_mock) -> None:
        """Verify we wait for the VM to boot up under different conditions."""
        # Testing with an existing VM directory and hardware emulation.
        self._vm.vm_dir = self.TempFilePath("vm_dir")
        osutils.SafeMakedirs(self._vm.vm_dir)
        self._vm.enable_kvm = True
        self._vm.WaitForBoot()
        start_mock.assert_not_called()
        boot_mock.assert_called()
        procs_mock.assert_not_called()

        start_mock.reset_mock()
        boot_mock.reset_mock()
        procs_mock.reset_mock()

        # Testing with a non-existent VM directory and software emulation.
        self._vm._RmVMDir()
        self._vm.enable_kvm = False
        self._vm.WaitForBoot()
        start_mock.assert_called()
        boot_mock.assert_called()
        procs_mock.assert_called()

    @mock.patch("fcntl.fcntl")
    def testSaveVMImageOnShutdownBasic(self, fcntl_mock) -> None:
        # mock.mock_open only seems to properly mock out read, not readline, so
        # do it ourselves.
        def readline_impl():
            readline_impl.count += 1
            if readline_impl.count == 1:
                return "some_output\n"
            return "thisisafakecommand\n"

        readline_impl.count = 0

        fcntl_mock.return_value = 0

        builtin = "__builtin__" if sys.version_info[0] == 2 else "builtins"
        m = mock.mock_open()
        filehandle = m()
        filehandle.readline.side_effect = readline_impl
        with mock.patch("%s.open" % builtin, m, create=True):
            self._vm.SaveVMImageOnShutdown("/some/dir/")

        self.assertTrue(self._vm.copy_image_on_shutdown)
        self.assertEqual(self._vm.image_copy_dir, "/some/dir/")

        write_calls = [
            mock.call("savevm chromite_lib_vm_snapshot\n"),
            mock.call("thisisafakecommand\n"),
        ]
        filehandle.write.assert_has_calls(write_calls)
        self.assertEqual(filehandle.write.call_count, 2)

        fcntl_calls = [
            mock.call(mock.ANY, fcntl.F_GETFL),
            mock.call(mock.ANY, fcntl.F_SETFL, os.O_NONBLOCK),
        ]
        fcntl_mock.assert_has_calls(fcntl_calls)
        self.assertEqual(fcntl_mock.call_count, 2)

    @mock.patch("fcntl.fcntl")
    @mock.patch("time.time")
    @mock.patch("time.sleep")
    def testSaveVMImageOnShutdownTimeout(
        self, sleep_mock, time_mock, fcntl_mock
    ) -> None:
        def time_impl():
            time_impl.count += 1
            if time_impl.count <= 2:
                return 0
            if time_impl.count == 3:
                return 30
            if time_impl.count == 4:
                return 31
            return 100

        time_impl.count = 0
        time_mock.side_effect = time_impl

        fcntl_mock.return_value = 0

        self._vm.copy_on_write = True
        builtin = "__builtin__" if sys.version_info[0] == 2 else "builtins"
        m = mock.mock_open()
        filehandle = m()
        filehandle.readline.side_effect = IOError(
            "Resource temporarily unavailable"
        )
        with mock.patch("%s.open" % builtin, m, create=True):
            self._vm.SaveVMImageOnShutdown("/some/dir/")

        self.assertEqual(sleep_mock.call_count, 1)

        self.assertTrue(self._vm.copy_image_on_shutdown)
        self.assertEqual(self._vm.image_copy_dir, "/some/dir/")

        write_calls = [
            mock.call("savevm chromite_lib_vm_snapshot\n"),
            mock.call("thisisafakecommand\n"),
        ]
        filehandle.write.assert_has_calls(write_calls)
        self.assertEqual(filehandle.write.call_count, 2)

        fcntl_calls = [
            mock.call(mock.ANY, fcntl.F_GETFL),
            mock.call(mock.ANY, fcntl.F_SETFL, os.O_NONBLOCK),
        ]
        fcntl_mock.assert_has_calls(fcntl_calls)
        self.assertEqual(fcntl_mock.call_count, 2)