virtual-deployment/virtualbox/pybox/helper/host_helper.py

#!/usr/bin/python3
#
# SPDX-License-Identifier: Apache-2.0
#

"""
This module provides functions to interact with a StarlingX controller-0 server via a
serial connection. The functions can be used to perform operations such as unlocking,
locking, rebooting, and installing a host. The module uses streamexpect library to
facilitate stream parsing.
"""

import streamexpect
from consts.timeout import HostTimeout
from utils import serial
from utils.install_log import LOG


def unlock_host(stream, hostname):
    """
    Unlocks given host
    Args:
        stream(stream): Stream to active controller
        hostname(str): Name of host to unlock
    Steps:
        - Check that host is locked
        - Unlock host
    """

    cmd = f"system host-list | grep {hostname}"
    serial.send_bytes(stream, cmd, expect_prompt=False)
    try:
        serial.expect_bytes(stream, "locked")
    except streamexpect.ExpectTimeout:
        LOG.error("Host %s not locked", hostname)
        return 1

    LOG.info("#### Unlock %s", hostname)
    cmd = f"system host-unlock {hostname}"
    serial.send_bytes(stream, cmd, expect_prompt=False)
    LOG.info("Unlocking %s", hostname)
    return None


def lock_host(stream, hostname):
    """
    Locks the specified host.
    Args:
        stream(stream): Stream to controller-0
        hostname(str): Name of host to lock
    Steps:
        - Check that host is unlocked
        - Lock host
    """

    cmd = f"system host-list |grep {hostname}"
    serial.send_bytes(stream, cmd, expect_prompt=False)
    try:
        serial.expect_bytes(stream, "unlocked")
    except streamexpect.ExpectTimeout:
        LOG.error("Host %s not unlocked", hostname)
        return 1

    LOG.info("Lock %s", hostname)
    cmd = f"system host-lock {hostname}"
    serial.send_bytes(stream, cmd, expect_prompt="keystone")
    LOG.info("Locking %s", hostname)
    return None


def reboot_host(stream, hostname):
    """
    Reboots host specified
    Args:
        stream():
        hostname(str): Host to reboot
    """

    LOG.info("Rebooting %s", hostname)
    cmd = f"system host-reboot {hostname}"
    serial.send_bytes(stream, cmd, expect_prompt=False)
    serial.expect_bytes(stream, "rebooting", HostTimeout.REBOOT)


def install_host(stream, hostname, host_type, host_id):
    """
    Initiates install of specified host. Requires controller-0 to be installed already.
    Args:
        stream(stream): Stream to cont0
        hostname(str): Name of host
        host_type(str): Type of host being installed e.g. 'storage' or 'worker'
        host_id(int): id to identify host
    """

    LOG.info("Installing %s with id %s", hostname, host_id)
    if host_type == 'controller':
        cmd = f"system host-update {host_id} personality=controller"
        serial.send_bytes(stream, cmd, expect_prompt=False)
    elif host_type == 'storage':
        cmd = f"system host-update {host_id} personality=storage"
        serial.send_bytes(stream, cmd, expect_prompt=False)
    else:
        cmd = f"system host-update {host_id} personality=worker hostname={hostname}"
        serial.send_bytes(stream, cmd, expect_prompt=False)


def disable_logout(stream):
    """
    Disables automatic logout of users.
    Args:
        stream(stream): stream to cont0
    """

    LOG.info('Disabling automatic logout')
    cmd = "export TMOUT=0"
    serial.send_bytes(stream, cmd)


def change_password(stream, username, password):
    """
    changes the default password (username) on initial login.
    Args:
        stream(stream): stream to cont0
        username: the current username (which should also be the password)
        password: the new password
    """

    LOG.info('Changing password to %s', password)
    serial.send_bytes(stream, username, expect_prompt=False)
    serial.expect_bytes(stream, "Password:")
    serial.send_bytes(stream, username, expect_prompt=False)
    serial.expect_bytes(stream, "Current password:")
    serial.send_bytes(stream, username, expect_prompt=False)
    serial.expect_bytes(stream, "New password:")
    serial.send_bytes(stream, password, expect_prompt=False)
    serial.expect_bytes(stream, "Retype new password")
    serial.send_bytes(stream, password)


def check_password(stream, password):
    """
    Checks the password.
    Args:
        stream(stream): Stream to cont0
        password(str): password to check.
    """
    ret = serial.expect_bytes(stream, 'assword', fail_ok=True, timeout=5)
    if ret == 0:
        serial.send_bytes(stream, password, expect_prompt=False)