mirror of
https://github.com/avatao-content/baseimage-tutorial-framework
synced 2024-11-09 02:37:17 +00:00
106 lines
3.0 KiB
Python
106 lines
3.0 KiB
Python
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
|
|
# All Rights Reserved. See LICENSE file for details.
|
|
|
|
from os.path import dirname
|
|
from re import findall
|
|
from re import compile as compileregex
|
|
from abc import ABC, abstractmethod
|
|
|
|
from watchdog.events import PatternMatchingEventHandler
|
|
|
|
from tfw.mixins.callback_mixin import CallbackMixin
|
|
from tfw.mixins.observer_mixin import ObserverMixin
|
|
from tfw.decorators.rate_limiter import RateLimiter
|
|
|
|
|
|
class CallbackEventHandler(PatternMatchingEventHandler, ABC):
|
|
def __init__(self, files, *callbacks):
|
|
super().__init__(files)
|
|
self.callbacks = callbacks
|
|
|
|
@RateLimiter(rate_per_second=2)
|
|
def on_modified(self, event):
|
|
for callback in self.callbacks:
|
|
callback()
|
|
|
|
|
|
class HistoryMonitor(CallbackMixin, ObserverMixin, ABC):
|
|
"""
|
|
Abstract class capable of monitoring and parsing a history file such as
|
|
bash HISTFILEs. Monitoring means detecting when the file was changed and
|
|
notifying subscribers about new content in the file.
|
|
|
|
This is useful for monitoring CLI sessions.
|
|
|
|
To specify a custom HistoryMonitor inherit from this class and override the
|
|
command pattern property and optionally the sanitize_command method.
|
|
See examples below.
|
|
"""
|
|
def __init__(self, histfile):
|
|
self.histfile = histfile
|
|
self._history = []
|
|
self._last_length = len(self._history)
|
|
self.observer.schedule(
|
|
CallbackEventHandler(
|
|
[self.histfile],
|
|
self._fetch_history,
|
|
self._invoke_callbacks
|
|
),
|
|
dirname(self.histfile)
|
|
)
|
|
|
|
@property
|
|
def history(self):
|
|
return self._history
|
|
|
|
def _fetch_history(self):
|
|
self._last_length = len(self._history)
|
|
with open(self.histfile, 'r') as ifile:
|
|
pattern = compileregex(self.command_pattern)
|
|
data = ifile.read()
|
|
self._history = [
|
|
self.sanitize_command(command)
|
|
for command in findall(pattern, data)
|
|
]
|
|
|
|
@property
|
|
@abstractmethod
|
|
def command_pattern(self):
|
|
raise NotImplementedError
|
|
|
|
def sanitize_command(self, command):
|
|
# pylint: disable=no-self-use
|
|
return command
|
|
|
|
def _invoke_callbacks(self):
|
|
if self._last_length < len(self._history):
|
|
self._execute_callbacks(self.history)
|
|
|
|
|
|
class BashMonitor(HistoryMonitor):
|
|
"""
|
|
HistoryMonitor for monitoring bash CLI sessions.
|
|
This requires the following to be set in bash
|
|
(note that this is done automatically by TFW):
|
|
PROMPT_COMMAND="history -a"
|
|
shopt -s cmdhist
|
|
shopt -s histappend
|
|
unset HISTCONTROL
|
|
"""
|
|
@property
|
|
def command_pattern(self):
|
|
return r'.+'
|
|
|
|
def sanitize_command(self, command):
|
|
return command.strip()
|
|
|
|
|
|
class GDBMonitor(HistoryMonitor):
|
|
"""
|
|
HistoryMonitor to monitor GDB sessions.
|
|
For this to work "set trace-commands on" must be set in GDB.
|
|
"""
|
|
@property
|
|
def command_pattern(self):
|
|
return r'(?<=\n)\+(.+)\n'
|