virtual-deployment/virtualbox/pybox/utils/serial.py

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

"""
This module provides functionality to connect and communicate with a remote host
using local domain socket.
"""

import re
import socket
from sys import stdout
import time
import streamexpect
from utils.install_log import LOG


def connect(hostname, port=10000, prefix=""):
    """
    Connect to local domain socket and return the socket object.

    Arguments:
    - Requires the hostname of target, e.g. controller-0
    - Requires TCP port if using Windows
    """

    if prefix:
        prefix = f"{prefix}_"
    socketname = f"/tmp/{prefix}{hostname}"
    if 'controller-0' in hostname:
        socketname += '_serial'
    LOG.info("Connecting to %s at %s", hostname, socketname)

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP)

    try:
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
        sock.connect(('localhost', port))
    except: # pylint: disable=bare-except
        LOG.info("Connection failed")
        pass  # pylint: disable=unnecessary-pass
        # disconnect(sock)
        sock = None
    # TODO (WEI): double check this # pylint: disable=fixme
    if sock:
        sock.setblocking(False)

    return sock


def disconnect(sock):
    """
    Disconnect a local doamin socket.

    Arguments:
    - Requires socket
    """

    # Shutdown connection and release resources
    LOG.info("Disconnecting from socket")
    sock.shutdown(socket.SHUT_RDWR)
    sock.close()


# pylint: disable=too-many-arguments, too-many-locals, too-many-branches
def get_output(stream, cmd, prompts=None, timeout=5, log=True, as_lines=True, flush=True):
    # pylint: disable=fixme
    # TODO: Not tested, will not work if kernel or other processes throw data on stdout or stderr
    """
    Execute a command and get its output. Make sure no other command is executing.
    And 'dmesg -D' was executed.
    """

    poll_period = 0.1
    max_read_buffer = 1024
    data = ""
    line_buf = ""
    lines = []
    if not prompts:
        prompts = [':~$ ', ':~# ', ':/home/wrsroot# ', '(keystone_.*)]$ ', '(keystone_.*)]# ']
    # Flush buffers
    if flush:
        try:
            trash = stream.poll(1)  # flush input buffers
            if trash:
                try:
                    LOG.info("Buffer has bytes before cmd execution: %s",
                             trash.decode('utf-8'))
                except Exception:  # pylint: disable=W0703
                    pass
        except streamexpect.ExpectTimeout:
            pass

    # Send command
    stream.sendall(f"{cmd}\n".encode('utf-8'))

    # Get response
    patterns = []
    for prompt in prompts:
        patterns.append(re.compile(prompt))

    now = time.time()
    end_time = now + float(timeout)
    prev_timeout = stream.gettimeout()
    stream.settimeout(poll_period)
    incoming = None
    # pylint: disable=too-many-nested-blocks
    try:
        while (end_time - now) >= 0:
            try:
                incoming = stream.recv(max_read_buffer)
            except socket.timeout:
                pass
            if incoming:
                data += incoming
                if log:
                    for char in incoming:
                        if char != '\n':
                            line_buf += char
                        else:
                            LOG.info(line_buf)
                            lines.append(line_buf)
                            line_buf = ""
                for pattern in patterns:
                    if pattern.search(data):
                        if as_lines:
                            return lines
                        return data
                now = time.time()
        raise streamexpect.ExpectTimeout()
    finally:
        stream.settimeout(prev_timeout)


def expect_bytes(stream, text, timeout=180, fail_ok=False, flush=True):
    """
    Wait for user specified text from stream.
    """

    time.sleep(1)
    if timeout < 60:
        LOG.info("Expecting text within %s seconds: %s\n", timeout, text)
    else:
        LOG.info("Expecting text within %s minutes: %s\n", timeout / 60, text)
    try:
        stream.expect_bytes(f"{text}".encode('utf-8'), timeout=timeout)
    except streamexpect.ExpectTimeout:
        if fail_ok:
            return -1

        stdout.write('\n')
        LOG.error("Did not find expected text")
        # disconnect(stream)
        raise
    except Exception as exception:
        LOG.info("Connection failed with %s", exception)
        raise

    stdout.write('\n')
    LOG.info("Found expected text: %s", text)

    time.sleep(1)
    if flush:
        try:
            incoming = stream.poll(1)  # flush input buffers
            if incoming:
                incoming += b'\n'
                try:
                    LOG.info(">>> expect_bytes: Buffer has bytes!")
                    stdout.write(incoming.decode('utf-8'))  # streamexpect hardcodes it
                except Exception:  # pylint: disable=W0703
                    pass
        except streamexpect.ExpectTimeout:
            pass

    return 0


# pylint: disable=inconsistent-return-statements
def send_bytes(stream, text, fail_ok=False, expect_prompt=True,
               prompt=None, timeout=180, send=True, flush=True):
    """
    Send user specified text to stream.
    """

    time.sleep(1)
    if flush:
        try:
            incoming = stream.poll(1)  # flush input buffers
            if incoming:
                incoming += b'\n'
                try:
                    LOG.info(">>> send_bytes: Buffer has bytes!")
                    stdout.write(incoming.decode('utf-8'))  # streamexpect hardcodes it
                except Exception:  # pylint: disable=W0703
                    pass
        except streamexpect.ExpectTimeout:
            pass

    LOG.info("Sending text: %s", text)
    try:
        if send:
            stream.sendall(f"{text}\n".encode('utf-8'))
        else:
            stream.sendall(f"{text}".encode('utf-8'))
        if expect_prompt:
            time.sleep(1)
            if prompt:
                return expect_bytes(stream, prompt, timeout=timeout, fail_ok=fail_ok)

            return_code = expect_bytes(stream, "~$", timeout=timeout, fail_ok=True)
            if return_code != 0:
                send_bytes(stream, '\n', expect_prompt=False)
                expect_bytes(stream, 'keystone', timeout=timeout)
                return
    except streamexpect.ExpectTimeout:
        if fail_ok:
            return -1

        LOG.error("Failed to send text, logging out.")
        stream.sendall("exit".encode('utf-8'))
        raise
    except Exception as exception:
        LOG.info("Connection failed with %s.", exception)
        raise

    return 0