Telnet.py

Go to the documentation of this file.

#  Copyright 2008-2015 Nokia Networks
#  Copyright 2016-     Robot Framework Foundation
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
 
from contextlib import contextmanager
import inspect
import re
import socket
import struct
import telnetlib
import time
 
try:
    import pyte
except ImportError:
    pyte = None
 
from robot.api import logger
from robot.api.deco import keyword
from robot.utils import (ConnectionCache, is_bytes, is_string, is_truthy,
                         secs_to_timestr, seq2str, timestr_to_secs)
from robot.version import get_version
 
 
 
class Telnet:
    ROBOT_LIBRARY_SCOPE = 'SUITE'
    ROBOT_LIBRARY_VERSION = get_version()
 
    
    def __init__(self, timeout='3 seconds', newline='CRLF',
                 prompt=None, prompt_is_regexp=False,
                 encoding='UTF-8', encoding_errors='ignore',
                 default_log_level='INFO', window_size=None,
                 environ_user=None, terminal_emulation=False,
                 terminal_type=None, telnetlib_log_level='TRACE',
                 connection_timeout=None):
        self._timeout_timeout = timeout or 3.0
        self._set_connection_timeout_set_connection_timeout(connection_timeout)
        self._newline_newline = newline or 'CRLF'
        self._prompt_prompt = (prompt, prompt_is_regexp)
        self._encoding_encoding = encoding
        self._encoding_errors_encoding_errors = encoding_errors
        self._default_log_level_default_log_level = default_log_level
        self._window_size_window_size = window_size
        self._environ_user_environ_user = environ_user
        self._terminal_emulation_terminal_emulation = terminal_emulation
        self._terminal_type_terminal_type = terminal_type
        self._telnetlib_log_level_telnetlib_log_level = telnetlib_log_level
        self._cache_cache = ConnectionCache()
        self._conn_conn = None
        self._conn_kws_conn_kws = self._lib_kws_lib_kws = None
 
    def get_keyword_names(self):
        return self._get_library_keywords_get_library_keywords() + self._get_connection_keywords_get_connection_keywords()
 
    def _get_library_keywords(self):
        if self._lib_kws_lib_kws is None:
            self._lib_kws_lib_kws = self._get_keywords_get_keywords(self, ['get_keyword_names'])
        return self._lib_kws_lib_kws
 
    def _get_keywords(self, source, excluded):
        return [name for name in dir(source)
                if self._is_keyword_is_keyword(name, source, excluded)]
 
    def _is_keyword(self, name, source, excluded):
        return (name not in excluded and
                not name.startswith('_') and
                name != 'get_keyword_names' and
                inspect.ismethod(getattr(source, name)))
 
    def _get_connection_keywords(self):
        if self._conn_kws_conn_kws is None:
            conn = self._get_connection_get_connection()
            excluded = [name for name in dir(telnetlib.Telnet())
                        if name not in ['write', 'read', 'read_until']]
            self._conn_kws_conn_kws = self._get_keywords_get_keywords(conn, excluded)
        return self._conn_kws_conn_kws
 
    def __getattr__(self, name):
        if name not in self._get_connection_keywords_get_connection_keywords():
            raise AttributeError(name)
        # If no connection is initialized, get attributes from a non-active
        # connection. This makes it possible for Robot to create keyword
        # handlers when it imports the library.
        return getattr(self._conn_conn or self._get_connection_get_connection(), name)
 
    @keyword(types=None)
    
    def open_connection(self, host, alias=None, port=23, timeout=None,
                        newline=None, prompt=None, prompt_is_regexp=False,
                        encoding=None, encoding_errors=None,
                        default_log_level=None, window_size=None,
                        environ_user=None, terminal_emulation=None,
                        terminal_type=None, telnetlib_log_level=None,
                        connection_timeout=None):
        timeout = timeout or self._timeout_timeout
        connection_timeout = (timestr_to_secs(connection_timeout)
                              if connection_timeout
                              else self._connection_timeout_connection_timeout)
        newline = newline or self._newline_newline
        encoding = encoding or self._encoding_encoding
        encoding_errors = encoding_errors or self._encoding_errors_encoding_errors
        default_log_level = default_log_level or self._default_log_level_default_log_level
        window_size = self._parse_window_size_parse_window_size(window_size or self._window_size_window_size)
        environ_user = environ_user or self._environ_user_environ_user
        if terminal_emulation is None:
            terminal_emulation = self._terminal_emulation_terminal_emulation
        terminal_type = terminal_type or self._terminal_type_terminal_type
        telnetlib_log_level = telnetlib_log_level or self._telnetlib_log_level_telnetlib_log_level
        if not prompt:
            prompt, prompt_is_regexp = self._prompt_prompt
        logger.info('Opening connection to %s:%s with prompt: %s%s'
                    % (host, port, prompt, ' (regexp)' if prompt_is_regexp else ''))
        self._conn_conn = self._get_connection_get_connection(host, port, timeout, newline,
                                          prompt, is_truthy(prompt_is_regexp),
                                          encoding, encoding_errors,
                                          default_log_level,
                                          window_size,
                                          environ_user,
                                          is_truthy(terminal_emulation),
                                          terminal_type,
                                          telnetlib_log_level,
                                          connection_timeout)
        return self._cache_cache.register(self._conn_conn, alias)
 
    def _parse_window_size(self, window_size):
        if not window_size:
            return None
        try:
            cols, rows = window_size.split('x', 1)
            return int(cols), int(rows)
        except ValueError:
            raise ValueError("Invalid window size '%s'. Should be "
                             "<rows>x<columns>." % window_size)
 
    
    def _get_connection(self, *args):
        return TelnetConnection(*args)
 
    def _set_connection_timeout(self, connection_timeout):
        self._connection_timeout_connection_timeout = connection_timeout
        if self._connection_timeout_connection_timeout:
            self._connection_timeout_connection_timeout = timestr_to_secs(connection_timeout)
 
    
    def switch_connection(self, index_or_alias):
        old_index = self._cache_cache.current_index
        self._conn_conn = self._cache_cache.switch(index_or_alias)
        return old_index
 
    
    def close_all_connections(self):
        self._conn_conn = self._cache_cache.close_all()
 
 
class TelnetConnection(telnetlib.Telnet):
    NEW_ENVIRON_IS = b'\x00'
    NEW_ENVIRON_VAR = b'\x00'
    NEW_ENVIRON_VALUE = b'\x01'
    INTERNAL_UPDATE_FREQUENCY = 0.03
 
    def __init__(self, host=None, port=23, timeout=3.0, newline='CRLF',
                 prompt=None, prompt_is_regexp=False,
                 encoding='UTF-8', encoding_errors='ignore',
                 default_log_level='INFO', window_size=None, environ_user=None,
                 terminal_emulation=False, terminal_type=None,
                 telnetlib_log_level='TRACE', connection_timeout=None):
        if connection_timeout is None:
            telnetlib.Telnet.__init__(self, host, int(port) if port else 23)
        else:
            telnetlib.Telnet.__init__(self, host, int(port) if port else 23,
                                      connection_timeout)
        self._set_timeout_set_timeout(timeout)
        self._set_newline_set_newline(newline)
        self._set_prompt_set_prompt(prompt, prompt_is_regexp)
        self._set_encoding_set_encoding(encoding, encoding_errors)
        self._set_default_log_level_set_default_log_level(default_log_level)
        self._window_size_window_size = window_size
        self._environ_user_environ_user = self._encode_encode(environ_user) if environ_user else None
        self._terminal_emulator_terminal_emulator = self._check_terminal_emulation_check_terminal_emulation(terminal_emulation)
        self._terminal_type_terminal_type = self._encode_encode(terminal_type) if terminal_type else None
        self.set_option_negotiation_callback(self._negotiate_options_negotiate_options)
        self._set_telnetlib_log_level_set_telnetlib_log_level(telnetlib_log_level)
        self._opt_responses_opt_responses = list()
 
    
    def set_timeout(self, timeout):
        self._verify_connection_verify_connection()
        old = self._timeout_timeout
        self._set_timeout_set_timeout(timeout)
        return secs_to_timestr(old)
 
    def _set_timeout(self, timeout):
        self._timeout_timeout = timestr_to_secs(timeout)
 
    
    def set_newline(self, newline):
        self._verify_connection_verify_connection()
        if self._terminal_emulator_terminal_emulator:
            raise AssertionError("Newline can not be changed when terminal emulation is used.")
        old = self._newline_newline
        self._set_newline_set_newline(newline)
        return old
 
    def _set_newline(self, newline):
        newline = str(newline).upper()
        self._newline_newline = newline.replace('LF', '\n').replace('CR', '\r')
 
    
    def set_prompt(self, prompt, prompt_is_regexp=False):
        self._verify_connection_verify_connection()
        old = self._prompt_prompt
        self._set_prompt_set_prompt(prompt, prompt_is_regexp)
        if old[1]:
            return old[0].pattern, True
        return old
 
    def _set_prompt(self, prompt, prompt_is_regexp):
        if is_truthy(prompt_is_regexp):
            self._prompt_prompt = (re.compile(prompt), True)
        else:
            self._prompt_prompt = (prompt, False)
 
    def _prompt_is_set(self):
        return self._prompt_prompt[0] is not None
 
    @keyword(types=None)
    
    def set_encoding(self, encoding=None, errors=None):
        self._verify_connection_verify_connection()
        if self._terminal_emulator_terminal_emulator:
            raise AssertionError("Encoding can not be changed when terminal emulation is used.")
        old = self._encoding_encoding
        self._set_encoding_set_encoding(encoding or old[0], errors or old[1])
        return old
 
    def _set_encoding(self, encoding, errors):
        self._encoding_encoding = (encoding.upper(), errors)
 
    def _encode(self, text):
        if is_bytes(text):
            return text
        if self._encoding_encoding[0] == 'NONE':
            return text.encode('ASCII')
        return text.encode(*self._encoding_encoding)
 
    def _decode(self, bytes):
        if self._encoding_encoding[0] == 'NONE':
            return bytes
        return bytes.decode(*self._encoding_encoding)
 
    
    def set_telnetlib_log_level(self, level):
        self._verify_connection_verify_connection()
        old = self._telnetlib_log_level_telnetlib_log_level
        self._set_telnetlib_log_level_set_telnetlib_log_level(level)
        return old
 
    def _set_telnetlib_log_level(self, level):
        if level.upper() == 'NONE':
            self._telnetlib_log_level_telnetlib_log_level = 'NONE'
        elif self._is_valid_log_level_is_valid_log_level(level) is False:
            raise AssertionError("Invalid log level '%s'" % level)
        self._telnetlib_log_level_telnetlib_log_level = level.upper()
 
    
    def set_default_log_level(self, level):
        self._verify_connection_verify_connection()
        old = self._default_log_level_default_log_level
        self._set_default_log_level_set_default_log_level(level)
        return old
 
    def _set_default_log_level(self, level):
        if level is None or not self._is_valid_log_level_is_valid_log_level(level):
            raise AssertionError("Invalid log level '%s'" % level)
        self._default_log_level_default_log_level = level.upper()
 
    def _is_valid_log_level(self, level):
        if level is None:
            return True
        if not is_string(level):
            return False
        return level.upper() in ('TRACE', 'DEBUG', 'INFO', 'WARN')
 
    
    def close_connection(self, loglevel=None):
        if self.sock:
            self.sock.shutdown(socket.SHUT_RDWR)
        self.close()
        output = self._decode_decode(self.read_all())
        self._log_log(output, loglevel)
        return output
 
    
    def login(self, username, password, login_prompt='login: ',
              password_prompt='Password: ', login_timeout='1 second',
              login_incorrect='Login incorrect'):
        output = self._submit_credentials_submit_credentials(username, password, login_prompt,
                                          password_prompt)
        if self._prompt_is_set_prompt_is_set():
            success, output2 = self._read_until_prompt_read_until_prompt()
        else:
            success, output2 = self._verify_login_without_prompt_verify_login_without_prompt(
                    login_timeout, login_incorrect)
        output += output2
        self._log_log(output)
        if not success:
            raise AssertionError('Login incorrect')
        return output
 
    def _submit_credentials(self, username, password, login_prompt, password_prompt):
        # Using write_bare here instead of write because don't want to wait for
        # newline: https://github.com/robotframework/robotframework/issues/1371
        output = self.read_untilread_until(login_prompt, 'TRACE')
        self.write_barewrite_bare(username + self._newline_newline)
        output += self.read_untilread_until(password_prompt, 'TRACE')
        self.write_barewrite_bare(password + self._newline_newline)
        return output
 
    def _verify_login_without_prompt(self, delay, incorrect):
        time.sleep(timestr_to_secs(delay))
        output = self.readread('TRACE')
        success = incorrect not in output
        return success, output
 
    
    def write(self, text, loglevel=None):
        newline = self._get_newline_for_get_newline_for(text)
        if newline in text:
            raise RuntimeError("'Write' keyword cannot be used with strings "
                               "containing newlines. Use 'Write Bare' instead.")
        self.write_barewrite_bare(text + newline)
        # Can't read until 'text' because long lines are cut strangely in the output
        return self.read_untilread_until(self._newline_newline, loglevel)
 
    def _get_newline_for(self, text):
        if is_bytes(text):
            return self._encode_encode(self._newline_newline)
        return self._newline_newline
 
    
    def write_bare(self, text):
        self._verify_connection_verify_connection()
        telnetlib.Telnet.write(self, self._encode_encode(text))
 
    
    def write_until_expected_output(self, text, expected, timeout,
                                    retry_interval, loglevel=None):
        timeout = timestr_to_secs(timeout)
        retry_interval = timestr_to_secs(retry_interval)
        maxtime = time.time() + timeout
        while time.time() < maxtime:
            self.write_barewrite_bare(text)
            self.read_untilread_until(text, loglevel)
            try:
                with self._custom_timeout_custom_timeout(retry_interval):
                    return self.read_untilread_until(expected, loglevel)
            except AssertionError:
                pass
        raise NoMatchError(expected, timeout)
 
    
    def write_control_character(self, character):
        self._verify_connection_verify_connection()
        self.sock.sendall(telnetlib.IAC + self._get_control_character_get_control_character(character))
 
    def _get_control_character(self, int_or_name):
        try:
            ordinal = int(int_or_name)
            return bytes(bytearray([ordinal]))
        except ValueError:
            return self._convert_control_code_name_to_character_convert_control_code_name_to_character(int_or_name)
 
    def _convert_control_code_name_to_character(self, name):
        code_names = {
                'BRK' : telnetlib.BRK,
                'IP' : telnetlib.IP,
                'AO' : telnetlib.AO,
                'AYT' : telnetlib.AYT,
                'EC' : telnetlib.EC,
                'EL' : telnetlib.EL,
                'NOP' : telnetlib.NOP
        }
        try:
            return code_names[name]
        except KeyError:
            raise RuntimeError("Unsupported control character '%s'." % name)
 
    
    def read(self, loglevel=None):
        self._verify_connection_verify_connection()
        output = self._decode_decode(self.read_very_eager())
        if self._terminal_emulator_terminal_emulator:
            self._terminal_emulator_terminal_emulator.feed(output)
            output = self._terminal_emulator_terminal_emulator.read()
        self._log_log(output, loglevel)
        return output
 
    
    def read_until(self, expected, loglevel=None):
        success, output = self._read_until_read_until(expected)
        self._log_log(output, loglevel)
        if not success:
            raise NoMatchError(expected, self._timeout_timeout, output)
        return output
 
    def _read_until(self, expected):
        self._verify_connection_verify_connection()
        if self._terminal_emulator_terminal_emulator:
            return self._terminal_read_until_terminal_read_until(expected)
        expected = self._encode_encode(expected)
        output = telnetlib.Telnet.read_until(self, expected, self._timeout_timeout)
        return output.endswith(expected), self._decode_decode(output)
 
    @property
    _terminal_frequency = property
    
    def _terminal_frequency(self):
        return min(self.INTERNAL_UPDATE_FREQUENCYINTERNAL_UPDATE_FREQUENCY, self._timeout_timeout)
 
    def _terminal_read_until(self, expected):
        max_time = time.time() + self._timeout_timeout
        output = self._terminal_emulator_terminal_emulator.read_until(expected)
        if output:
            return True, output
        while time.time() < max_time:
            output = telnetlib.Telnet.read_until(self, self._encode_encode(expected),
                                                 self._terminal_frequency_terminal_frequency_terminal_frequency)
            self._terminal_emulator_terminal_emulator.feed(self._decode_decode(output))
            output = self._terminal_emulator_terminal_emulator.read_until(expected)
            if output:
                return True, output
        return False, self._terminal_emulator_terminal_emulator.read()
 
    def _read_until_regexp(self, *expected):
        self._verify_connection_verify_connection()
        if self._terminal_emulator_terminal_emulator:
            return self._terminal_read_until_regexp_terminal_read_until_regexp(expected)
        expected = [self._encode_encode(exp) if is_string(exp) else exp
                    for exp in expected]
        return self._telnet_read_until_regexp_telnet_read_until_regexp(expected)
 
    def _terminal_read_until_regexp(self, expected_list):
        max_time = time.time() + self._timeout_timeout
        regexps_bytes = [self._to_byte_regexp_to_byte_regexp(rgx) for rgx in expected_list]
        regexps_unicode = [re.compile(self._decode_decode(rgx.pattern))
                               for rgx in regexps_bytes]
        out = self._terminal_emulator_terminal_emulator.read_until_regexp(regexps_unicode)
        if out:
            return True, out
        while time.time() < max_time:
            output = self.expect(regexps_bytes, self._terminal_frequency_terminal_frequency_terminal_frequency)[-1]
            self._terminal_emulator_terminal_emulator.feed(self._decode_decode(output))
            out = self._terminal_emulator_terminal_emulator.read_until_regexp(regexps_unicode)
            if out:
                return True, out
        return False, self._terminal_emulator_terminal_emulator.read()
 
    def _telnet_read_until_regexp(self, expected_list):
        expected = [self._to_byte_regexp_to_byte_regexp(exp) for exp in expected_list]
        try:
            index, _, output = self.expect(expected, self._timeout_timeout)
        except TypeError:
            index, output = -1, b''
        return index != -1, self._decode_decode(output)
 
    def _to_byte_regexp(self, exp):
        if is_bytes(exp):
            return re.compile(exp)
        if is_string(exp):
            return re.compile(self._encode_encode(exp))
        pattern = exp.pattern
        if is_bytes(pattern):
            return exp
        return re.compile(self._encode_encode(pattern))
 
    
    def read_until_regexp(self, *expected):
        if not expected:
            raise RuntimeError('At least one pattern required')
        if self._is_valid_log_level_is_valid_log_level(expected[-1]):
            loglevel = expected[-1]
            expected = expected[:-1]
        else:
            loglevel = None
        success, output = self._read_until_regexp_read_until_regexp(*expected)
        self._log_log(output, loglevel)
        if not success:
            expected = [exp if is_string(exp) else exp.pattern
                        for exp in expected]
            raise NoMatchError(expected, self._timeout_timeout, output)
        return output
 
    
    def read_until_prompt(self, loglevel=None, strip_prompt=False):
        if not self._prompt_is_set_prompt_is_set():
            raise RuntimeError('Prompt is not set.')
        success, output = self._read_until_prompt_read_until_prompt()
        self._log_log(output, loglevel)
        if not success:
            prompt, regexp = self._prompt_prompt
            raise AssertionError("Prompt '%s' not found in %s."
                                 % (prompt if not regexp else prompt.pattern,
                                    secs_to_timestr(self._timeout_timeout)))
        if is_truthy(strip_prompt):
            output = self._strip_prompt_strip_prompt(output)
        return output
 
    def _read_until_prompt(self):
        prompt, regexp = self._prompt_prompt
        read_until = self._read_until_regexp_read_until_regexp if regexp else self._read_until_read_until
        return read_until(prompt)
 
    def _strip_prompt(self, output):
        prompt, regexp = self._prompt_prompt
        if not regexp:
            length = len(prompt)
        else:
            match = prompt.search(output)
            length = match.end() - match.start()
        return output[:-length]
 
    
    def execute_command(self, command, loglevel=None, strip_prompt=False):
        self.writewrite(command, loglevel)
        return self.read_until_promptread_until_prompt(loglevel, strip_prompt)
 
    @contextmanager
    def _custom_timeout(self, timeout):
        old = self.set_timeoutset_timeout(timeout)
        try:
            yield
        finally:
            self.set_timeoutset_timeout(old)
 
    def _verify_connection(self):
        if not self.sock:
            raise RuntimeError('No connection open')
 
    def _log(self, msg, level=None):
        msg = msg.strip()
        if msg:
            logger.write(msg, level or self._default_log_level_default_log_level)
 
    def _negotiate_options(self, sock, cmd, opt):
        # We don't have state changes in our accepted telnet options.
        # Therefore, we just track if we've already responded to an option. If
        # this is the case, we must not send any response.
        if cmd in (telnetlib.DO, telnetlib.DONT, telnetlib.WILL, telnetlib.WONT):
            if (cmd, opt) in self._opt_responses_opt_responses:
                return
            else:
                self._opt_responses_opt_responses.append((cmd, opt))
 
        # This is supposed to turn server side echoing on and turn other options off.
        if opt == telnetlib.ECHO and cmd in (telnetlib.WILL, telnetlib.WONT):
            self._opt_echo_on_opt_echo_on(opt)
        elif cmd == telnetlib.DO and opt == telnetlib.TTYPE and self._terminal_type_terminal_type:
            self._opt_terminal_type_opt_terminal_type(opt, self._terminal_type_terminal_type)
        elif cmd == telnetlib.DO and opt == telnetlib.NEW_ENVIRON and self._environ_user_environ_user:
            self._opt_environ_user_opt_environ_user(opt, self._environ_user_environ_user)
        elif cmd == telnetlib.DO and opt == telnetlib.NAWS and self._window_size_window_size:
            self._opt_window_size_opt_window_size(opt, *self._window_size_window_size)
        elif opt != telnetlib.NOOPT:
            self._opt_dont_and_wont_opt_dont_and_wont(cmd, opt)
 
    def _opt_echo_on(self, opt):
        return self.sock.sendall(telnetlib.IAC + telnetlib.DO + opt)
 
    def _opt_terminal_type(self, opt, terminal_type):
        self.sock.sendall(telnetlib.IAC + telnetlib.WILL + opt)
        self.sock.sendall(telnetlib.IAC + telnetlib.SB + telnetlib.TTYPE
                          + self.NEW_ENVIRON_ISNEW_ENVIRON_IS + terminal_type
                          + telnetlib.IAC + telnetlib.SE)
 
    def _opt_environ_user(self, opt, environ_user):
        self.sock.sendall(telnetlib.IAC + telnetlib.WILL + opt)
        self.sock.sendall(telnetlib.IAC + telnetlib.SB + telnetlib.NEW_ENVIRON
                          + self.NEW_ENVIRON_ISNEW_ENVIRON_IS + self.NEW_ENVIRON_VARNEW_ENVIRON_VAR
                          + b"USER" + self.NEW_ENVIRON_VALUENEW_ENVIRON_VALUE + environ_user
                          + telnetlib.IAC + telnetlib.SE)
 
    def _opt_window_size(self, opt, window_x, window_y):
        self.sock.sendall(telnetlib.IAC + telnetlib.WILL + opt)
        self.sock.sendall(telnetlib.IAC + telnetlib.SB + telnetlib.NAWS
                          + struct.pack('!HH', window_x, window_y)
                          + telnetlib.IAC + telnetlib.SE)
 
    def _opt_dont_and_wont(self, cmd, opt):
        if cmd in (telnetlib.DO, telnetlib.DONT):
            self.sock.sendall(telnetlib.IAC + telnetlib.WONT + opt)
        elif cmd in (telnetlib.WILL, telnetlib.WONT):
            self.sock.sendall(telnetlib.IAC + telnetlib.DONT + opt)
 
    def msg(self, msg, *args):
        # Forward telnetlib's debug messages to log
        if self._telnetlib_log_level_telnetlib_log_level != 'NONE':
            logger.write(msg % args, self._telnetlib_log_level_telnetlib_log_level)
 
    def _check_terminal_emulation(self, terminal_emulation):
        if not terminal_emulation:
            return False
        if not pyte:
            raise RuntimeError("Terminal emulation requires pyte module!\n"
                               "http://pypi.python.org/pypi/pyte/")
        return TerminalEmulator(window_size=self._window_size_window_size,
                                newline=self._newline_newline)
 
 
class TerminalEmulator:
 
    def __init__(self, window_size=None, newline="\r\n"):
        self._rows, self._columns_columns = window_size or (200, 200)
        self._newline_newline = newline
        self._stream_stream = pyte.Stream()
        self._screen_screen = pyte.HistoryScreen(self._rows,
                                          self._columns_columns,
                                          history=100000)
        self._stream_stream.attach(self._screen_screen)
        self._buffer_buffer = ''
        self._whitespace_after_last_feed_whitespace_after_last_feed = ''
 
    @property
    current_output = property
    
    def current_output(self):
        return self._buffer_buffer + self._dump_screen_dump_screen()
 
    def _dump_screen(self):
        return self._get_history_get_history(self._screen_screen) + \
               self._get_screen_get_screen(self._screen_screen) + \
               self._whitespace_after_last_feed_whitespace_after_last_feed
 
    def _get_history(self, screen):
        if not screen.history.top:
            return ''
        rows = []
        for row in screen.history.top:
            # Newer pyte versions store row data in mappings
            data = (char.data for _, char in sorted(row.items()))
            rows.append(''.join(data).rstrip())
        return self._newline_newline.join(rows).rstrip(self._newline_newline) + self._newline_newline
 
    def _get_screen(self, screen):
        rows = (row.rstrip() for row in screen.display)
        return self._newline_newline.join(rows).rstrip(self._newline_newline)
 
    def feed(self, text):
        self._stream_stream.feed(text)
        self._whitespace_after_last_feed_whitespace_after_last_feed = text[len(text.rstrip()):]
 
    def read(self):
        current_out = self.current_outputcurrent_outputcurrent_output
        self._update_buffer_update_buffer('')
        return current_out
 
    def read_until(self, expected):
        current_out = self.current_outputcurrent_outputcurrent_output
        exp_index = current_out.find(expected)
        if exp_index != -1:
            self._update_buffer_update_buffer(current_out[exp_index+len(expected):])
            return current_out[:exp_index+len(expected)]
        return None
 
    def read_until_regexp(self, regexp_list):
        current_out = self.current_outputcurrent_outputcurrent_output
        for rgx in regexp_list:
            match = rgx.search(current_out)
            if match:
                self._update_buffer_update_buffer(current_out[match.end():])
                return current_out[:match.end()]
        return None
 
    def _update_buffer(self, terminal_buffer):
        self._buffer_buffer = terminal_buffer
        self._whitespace_after_last_feed_whitespace_after_last_feed = ''
        self._screen_screen.reset()
 
 
class NoMatchError(AssertionError):
    ROBOT_SUPPRESS_NAME = True
 
    def __init__(self, expected, timeout, output=None):
        self.expectedexpected = expected
        self.timeouttimeout = secs_to_timestr(timeout)
        self.outputoutput = output
        AssertionError.__init__(self, self._get_message_get_message())
 
    def _get_message(self):
        expected = "'%s'" % self.expectedexpected \
                   if is_string(self.expectedexpected) \
                   else seq2str(self.expectedexpected, lastsep=' or ')
        msg = "No match found for %s in %s." % (expected, self.timeouttimeout)
        if self.outputoutput is not None:
            msg += ' Output:\n%s' % self.outputoutput
        return msg