test-tutorial-framework/solvable/src/event_handler_main.py

import logging
from functools import partial

from tornado.ioloop import IOLoop

from tfw.fsm import YamlFSM
from tfw.event_handlers import FSMAwareEventHandler, ControlEventHandler
from tfw.components.ide import IdeHandler
from tfw.components.terminal import TerminalHandler
from tfw.components.frontend import FrontendProxyHandler, ConsoleLogsHandler, MessageQueueHandler
from tfw.components.process_management import ProcessHandler, ProcessLogHandler
from tfw.components.fsm import FSMHandler
from tfw.main import EventHandlerFactory, setup_logger, setup_signal_handlers
from tfw.config import TFWENV, TAOENV
from tfw.internals.networking import Intent

from custom_handlers import CenatorHandler, TestCommandsHandler, messageFSMStepsHandler

LOG = logging.getLogger(__name__)


def main():
    # pylint: disable=unused-variable
    setup_logger(__file__)

    eh_factory = EventHandlerFactory()
    # TFW builtin EventHandlers (required for their respective functionalities)
    # TFW FSM
    fsm_eh = eh_factory.build(FSMHandler(
        fsm_type=partial(
            YamlFSM,
            'test_fsm.yml',
            {}  # jinja2 variables, empty dict enables jinja2 without any variables
        ),
        initial_trigger='step_1'
    ), event_handler_type=ControlEventHandler)
    # Web IDE backend
    ide_eh = eh_factory.build(IdeHandler(
        patterns=['/home/user/workdir/*', '/srv/webservice/user_ops.py']
    ), event_handler_type=ControlEventHandler)
    # Web shell backend
    terminal_eh = eh_factory.build(TerminalHandler(
        port=TFWENV.TERMINADO_PORT,
        user=TAOENV.USER,
        working_directory=TFWENV.TERMINADO_WD,
        histfile=TFWENV.HISTFILE
    ), event_handler_type=ControlEventHandler)
    # Handles 'deploy' button clicks
    processmanager_eh = eh_factory.build(ProcessHandler(
        supervisor_uri=TFWENV.SUPERVISOR_HTTP_URI
    ), event_handler_type=ControlEventHandler)
    # Sends live logs of webservice process to frontend
    logmonitor_eh = eh_factory.build(ProcessLogHandler(
        process_name='webservice',
        supervisor_uri=TFWENV.SUPERVISOR_HTTP_URI,
        log_tail=2000
    ), event_handler_type=ControlEventHandler)
    # Proxies frontend API calls to frontend
    frontend_eh = eh_factory.build(FrontendProxyHandler(), event_handler_type=ControlEventHandler)
    message_queue_eh = eh_factory.build(MessageQueueHandler(25), event_handler_type=ControlEventHandler)
    # Writes live logs to console on frontend
    console_logs_eh = eh_factory.build(ConsoleLogsHandler(stream='stdout'), event_handler_type=ControlEventHandler)

    # Replace these with your custom event handlers
    # Echoes executed commands to messages
    cenator_eh = eh_factory.build(CenatorHandler(), event_handler_type=ControlEventHandler)
    # Echoes FSM steps
    message_fsm_steps_eh = eh_factory.build(
        messageFSMStepsHandler,
        event_handler_type=FSMAwareEventHandler
    )
    # Catches special commands
    commands_eh = eh_factory.build(TestCommandsHandler(
        bashrc=f'/home/{TAOENV.USER}/.bashrc'
    ), event_handler_type=ControlEventHandler)

    setup_signal_handlers()
    IOLoop.current().start()


if __name__ == '__main__':
    main()