Source code for mirakuru.output

# Copyright (C) 2014 by Clearcode <http://clearcode.cc>
# and associates (see AUTHORS).

# This file is part of mirakuru.

# mirakuru is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# mirakuru is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Lesser General Public License for more details.

# You should have received a copy of the GNU Lesser General Public License
# along with mirakuru.  If not, see <http://www.gnu.org/licenses/>.
"""This executor awaits for appearance of a predefined banner in output."""

import re
import select

from mirakuru.base import Executor


[docs]class OutputExecutor(Executor):
"""Executor that awaits for string output being present in output.""" def __init__(self, command, banner, shell=False, timeout=None, sleep=0.1): """ Initialize OutputExecutor executor. :param str command: command to run to start service :param str banner: string that has to appear in process output - should compile to regular expression. :param bool shell: see `subprocess.Popen` :param int timeout: time to wait for process to start or stop. if None, wait indefinitely. :param float sleep: how often to check for start/stop condition """ Executor.__init__(self, command, shell, timeout, sleep) self._banner = re.compile(banner) def start(self): """ Start process. .. note:: Process will be considered started, when defined banner will appear in process output. """ Executor.start(self) # get a polling object self.poll_obj = select.poll() # register a file descriptor # POLLIN because we will wait for data to read self.poll_obj.register(self.output(), select.POLLIN) try: self.wait_for(self._wait_for_output) # unregister the file descriptor and delete the polling object self.poll_obj.unregister(self.output()) finally: del self.poll_obj def _wait_for_output(self): """ Check if output matches banner. .. warning:: Waiting for I/O completion. It does not work on Windows. Sorry. """ # Here we should get an empty list or list with a tuple [(fd, event)] # When we get list with a tuple we can use readline method on # the file descriptor. poll_result = self.poll_obj.poll(0) if poll_result: line = self.output().readline() if self._banner.match(line): return True return False