baseimage-tutorial-framework/lib/tfw/components/terminado_mini_server.py

import logging

from tornado.ioloop import IOLoop
from tornado.web import Application
from terminado import TermSocket, SingleTermManager

from tfw.config import TFWENV

LOG = logging.getLogger(__name__)


class TerminadoMiniServer:
    def __init__(self, url, port, workdir, shellcmd):
        self.port = port
        self._term_manager = SingleTermManager(
            shell_command=shellcmd,
            term_settings={'cwd': workdir}
        )
        self.application = Application([(
            url,
            TerminadoMiniServer.ResetterTermSocket,
            {'term_manager': self._term_manager}
        )])

    @property
    def term_manager(self):
        return self._term_manager

    @property
    def pty(self):
        if self.term_manager.terminal is None:
            self.term_manager.get_terminal()
        return self.term_manager.terminal.ptyproc

    class ResetterTermSocket(TermSocket): # pylint: disable=abstract-method
        def check_origin(self, origin):
            return True

        def on_close(self):
            self.term_manager.terminal = None
            self.term_manager.get_terminal()

    def listen(self):
        self.application.listen(self.port)

    def stop(self):
        self.term_manager.shutdown()


if __name__ == '__main__':
    LOG.info('Terminado Mini Server listening on %s', TFWENV.TERMINADO_PORT)
    TerminadoMiniServer(
        '/terminal',
        TFWENV.TERMINADO_PORT,
        TFWENV.TERMINADO_WD,
        ['bash']
    ).listen()
    IOLoop.instance().start()