baseimage-tutorial-framework/tfw/components/terminal/terminal_handler.py

import logging

from .history_monitor import BashMonitor
from .terminado_mini_server import TerminadoMiniServer

LOG = logging.getLogger(__name__)


class TerminalHandler:
    keys = ['terminal']
    type_id = 'ControlEventHandler'

    def __init__(self, *, port, user, working_directory, histfile):
        self.connector, self._historymonitor = None, None
        self._histfile = histfile
        bash_as_user_cmd = ['sudo', '-u', user, 'bash']

        self.terminado_server = TerminadoMiniServer(
            '/terminal',
            port,
            working_directory,
            bash_as_user_cmd
        )
        self.terminado_server.listen()

        self.commands = {
            'terminal.write': self.handle_write
        }

    def start(self):
        self._historymonitor = BashMonitor(self.connector, self._histfile)
        self._historymonitor.start()

    @property
    def historymonitor(self):
        return self._historymonitor

    def handle_event(self, message, _):
        try:
            self.commands[message['key']](message)
        except KeyError:
            LOG.error('IGNORING MESSAGE: Invalid message received: %s', message)

    def handle_write(self, message):
        concat = message.get('concat', False)
        if not concat:
            self.terminado_server.pty.write('\x15')
        self.terminado_server.pty.write(message['command'])

    def cleanup(self):
        self.terminado_server.stop()
        self.historymonitor.stop()