test/check-machines-consoles

#!/usr/bin/python3 -cimport os, sys; os.execv(os.path.dirname(sys.argv[1]) + "/common/pywrap", sys.argv)

# This file is part of Cockpit.
#
# Copyright (C) 2021 Red Hat, Inc.
#
# Cockpit is free software; you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation; either version 2.1 of the License, or
# (at your option) any later version.
#
# Cockpit is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Cockpit; If not, see <http://www.gnu.org/licenses/>.

import time

import machineslib
import testlib


@testlib.nondestructive
class TestMachinesConsoles(machineslib.VirtualMachinesCase):

    def waitDownloadFile(self, filename: str, expected_size: int | None = None, content: str | None = None) -> None:
        b = self.browser
        filepath = b.driver.download_dir / filename

        # Big downloads can take a while
        testlib.wait(filepath.exists, tries=120)
        if expected_size is not None:
            testlib.wait(lambda: filepath.stat().st_size == expected_size)

        if content is not None:
            self.assertEqual(filepath.read_text(), content)

    @testlib.skipImage('SPICE not supported on RHEL', "rhel-*", "centos-*")
    def testExternalConsole(self):
        b = self.browser

        self.createVm("subVmTest1", graphics="spice")

        self.login_and_go("/machines")
        self.waitPageInit()
        self.waitVmRow("subVmTest1")

        b.wait_in_text("#vm-subVmTest1-system-state", "Running")  # running or paused
        self.goToVmPage("subVmTest1")

        # since VNC is not defined for this VM, the view for "Desktop Viewer" is rendered by default
        b.wait_in_text(".pf-v5-c-console__manual-connection dl > div:first-child dd", "127.0.0.1")
        b.wait_in_text(".pf-v5-c-console__manual-connection dl > div:nth-child(2) dd", "5900")

        b.allow_download()
        b.click(".pf-v5-c-console__remote-viewer-launch-vv")  # "Launch Remote Viewer" button
        content = """[virt-viewer]
type=spice
host=127.0.0.1
port=5900
delete-this-file=1
fullscreen=0

[...............................GraphicsConsole]
"""
        # HACK: Due to a bug in cockpit-machines, Firefox downloads the desktop
        # viewer file as a randomly generated file while chromium as
        # "download". As we want to download this as "$vmName.vv" in the end
        # work around the issue for now.
        if b.browser == "chromium":
            self.waitDownloadFile("download", content=content)
        else:
            b.wait_visible("#dynamically-generated-file")  # is .vv file generated for download?
            self.assertEqual(b.attr("#dynamically-generated-file", "href"),
                            u"data:application/x-virt-viewer,%5Bvirt-viewer%5D%0Atype%3Dspice%0Ahost%3D127.0.0.1%0Aport%3D5900%0Adelete-this-file%3D1%0Afullscreen%3D0%0A")

        # Go to the expanded console view
        b.click("button:contains(Expand)")

        # Check "More information"
        b.click('.pf-v5-c-console__remote-viewer .pf-v5-c-expandable-section__toggle')
        b.wait_in_text('.pf-v5-c-expandable-section__content',
                       'Clicking "Launch remote viewer" will download')

        b.assert_pixels("#vm-subVmTest1-consoles-page", "vm-details-console-external", skip_layouts=["rtl"])

    def testInlineConsole(self, urlroot=""):
        b = self.browser

        args = self.createVm("subVmTest1", "vnc")

        if urlroot != "":
            self.machine.write("/etc/cockpit/cockpit.conf", f"[WebService]\nUrlRoot={urlroot}")

        self.login_and_go("/machines", urlroot=urlroot)
        self.waitPageInit()
        self.waitVmRow("subVmTest1")

        b.wait_in_text("#vm-subVmTest1-system-state", "Running")  # running or paused
        self.goToVmPage("subVmTest1")

        # since VNC is defined for this VM, the view for "In-Browser Viewer" is rendered by default
        b.wait_visible(".pf-v5-c-console__vnc canvas")

        # make sure the log file is full - then empty it and reboot the VM - the log file should fill up again
        self.waitGuestBooted(args['logfile'])

        self.machine.execute(f"echo '' > {args['logfile']}")
        b.click("#subVmTest1-system-vnc-sendkey")
        b.click("#ctrl-alt-Delete")
        self.waitLogFile(args['logfile'], "reboot: Restarting system")

    def testInlineConsoleWithUrlRoot(self, urlroot=""):
        self.testInlineConsole(urlroot="/webcon")

    def testSerialConsole(self):
        b = self.browser
        m = self.machine
        name = "vmWithSerialConsole"

        self.createVm(name, graphics='vnc', ptyconsole=True)

        self.login_and_go("/machines")
        self.waitPageInit()
        self.waitVmRow(name)

        self.goToVmPage(name)
        b.wait_in_text(f"#vm-{name}-system-state", "Running")

        b.click("#pf-v5-c-console__type-selector")
        b.wait_visible("#pf-v5-c-console__type-selector + .pf-v5-c-select__menu")
        b.click("#SerialConsole button")
        b.wait_not_present("#pf-v5-c-console__type-selector + .pf-v5-c-select__menu")

        # In case the OS already finished booting, press Enter into the console to re-trigger the login prompt
        # Sometimes, pressing Enter one time doesn't take effect, so loop to press Enter to make sure
        # the console has accepted it.
        for _ in range(0, 60):
            b.focus(f"#{name}-terminal .xterm-accessibility-tree")
            b.key("Enter")
            if "Welcome to Alpine Linux" in b.text(f"#{name}-terminal .xterm-accessibility-tree"):
                break
            time.sleep(1)
        # Make sure the content of console is expected
        testlib.wait(lambda: "Welcome to Alpine Linux" in b.text(f"#{name}-terminal .xterm-accessibility-tree"))

        b.click(f"#{name}-serialconsole-disconnect")
        b.wait_text(f"#{name}-terminal", "Disconnected from serial console. Click the connect button.")

        b.click(f"#{name}-serialconsole-connect")
        b.wait_in_text(f"#{name}-terminal .xterm-accessibility-tree > div:nth-child(1)",
                       f"Connected to domain '{name}'")

        b.click("button:contains(Expand)")
        b.assert_pixels("#vm-vmWithSerialConsole-consoles-page", "vm-details-console-serial",
                        ignore=[".pf-v5-c-console__vnc"], skip_layouts=["rtl"])

        # Add a second serial console
        m.execute("""
            virsh destroy vmWithSerialConsole;
            virt-xml --add-device vmWithSerialConsole --console pty,target_type=virtio;
            virsh start vmWithSerialConsole""")
        b.click("#pf-v5-c-console__type-selector")
        b.wait_visible("#pf-v5-c-console__type-selector + .pf-v5-c-select__menu")
        b.click("li:contains('Serial console (serial0)') button")
        b.wait(lambda: m.execute("ps aux | grep 'virsh -c qemu:///system console vmWithSerialConsole serial0'"))
        b.click("#pf-v5-c-console__type-selector")
        b.click("li:contains('Serial console (console1)') button")
        b.wait(lambda: m.execute("ps aux | grep 'virsh -c qemu:///system console vmWithSerialConsole console1'"))

        # Add multiple serial consoles
        # Remove all console firstly
        m.execute("virsh destroy vmWithSerialConsole")
        m.execute("virt-xml --remove-device vmWithSerialConsole --console all")
        # Add console1 ~ console5
        m.execute("""
                  for i in {1..5}; do
                    virt-xml vmWithSerialConsole --add-device --console pty,target.type=virtio;
                  done
                  virsh start vmWithSerialConsole
                  """)

        for i in range(0, 6):
            b.click("#pf-v5-c-console__type-selector")
            b.wait_visible("#pf-v5-c-console__type-selector + .pf-v5-c-select__menu")
            b.click(f'li:contains(\'Serial console ({"serial" if i == 0 else "console"}{i})\') button')
            b.wait(lambda: m.execute(
                f'ps aux | grep \'virsh -c qemu:///system console vmWithSerialConsole {"serial" if i == 0 else "console"}{i}\''))  # noqa: B023, E501

        # disconnecting the serial console closes the pty channel
        self.allow_journal_messages("connection unexpectedly closed by peer",
                                    ".*Connection reset by peer")
        self.allow_browser_errors("Disconnection timed out.",
                                  "Failed when connecting: Connection closed")
        self.allow_journal_messages(".* couldn't shutdown fd: Transport endpoint is not connected")
        self.allow_journal_messages("127.0.0.1:5900: couldn't read: Connection refused")

    def testBasic(self):
        b = self.browser
        name = "subVmTest1"

        self.createVm(name, graphics="vnc", ptyconsole=True)

        self.login_and_go("/machines")
        self.waitPageInit()

        self.waitVmRow(name)
        self.goToVmPage(name)
        b.wait_in_text(f"#vm-{name}-system-state", "Running")

        # test switching console from serial to graphical
        b.wait_visible(f"#vm-{name}-consoles")
        b.wait_visible(".pf-v5-c-console__vnc canvas")

        b.click("#pf-v5-c-console__type-selector")
        b.wait_visible("#pf-v5-c-console__type-selector + .pf-v5-c-select__menu")
        b.click("#SerialConsole button")
        b.wait_not_present("#pf-v5-c-console__type-selector + .pf-v5-c-select__menu")

        b.wait_not_present(".pf-v5-c-console__vnc canvas")
        b.wait_visible(f"#{name}-terminal")

        # Go back to Vnc console
        b.click("#pf-v5-c-console__type-selector")
        b.wait_visible("#pf-v5-c-console__type-selector + .pf-v5-c-select__menu")
        b.click("#VncConsole button")
        b.wait_not_present("#pf-v5-c-console__type-selector + .pf-v5-c-select__menu")
        b.wait_visible(".pf-v5-c-console__vnc canvas")

        # Go to the expanded console view
        b.click("button:contains(Expand)")

        # Test message is present if VM is not running
        self.performAction(name, "forceOff", checkExpectedState=False)

        b.wait_in_text("#vm-not-running-message", "start the virtual machine")

        # Test deleting VM from console page will not trigger any error
        self.performAction(name, "delete")
        b.wait_visible(f"#vm-{name}-delete-modal-dialog")
        b.click(f"#vm-{name}-delete-modal-dialog button:contains(Delete)")
        self.waitPageInit()
        self.waitVmRow(name, present=False)

        b.wait_not_present("#navbar-oops")

        self.allow_journal_messages("connection unexpectedly closed by peer")
        self.allow_browser_errors("Disconnection timed out.",
                                  "Failed when connecting: Connection closed")


if __name__ == '__main__':
    testlib.test_main()