tests/termex.py

   1 #!/usr/bin/env python3
   2
   3 # Copyright (C) 2016-2017 Petr Pavlu <setup@dagobah.cz>
   4 #
   5 # This file is part of CenterIM.
   6 #
   7 # CenterIM is free software; you can redistribute it and/or modify
   8 # it under the terms of the GNU General Public License as published by
   9 # the Free Software Foundation; either version 2 of the License, or
  10 # (at your option) any later version.
  11 #
  12 # CenterIM is distributed in the hope that it will be useful,
  13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 # GNU General Public License for more details.
  16 #
  17 # You should have received a copy of the GNU General Public License
  18 # along with CenterIM.  If not, see <http://www.gnu.org/licenses/>.
  19
  20 """Termex terminal emulator and test framework."""
  21
  22 import argparse
  23 import difflib
  24 import errno
  25 import os
  26 import pty
  27 import re
  28 import selectors
  29 import sys
  30 import time
  31 import xml.etree.ElementTree as ElementTree
  32
  33 # The module relies on selectors.select() to automatically retry the operation
  34 # with a recomputed timeout when it gets interrupted by a signal. This
  35 # behaviour was introduced in Python 3.5.
  36 if sys.hexversion < 0x03050000:
  37     print("This program requires at least Python 3.5.", file=sys.stderr)
  38     sys.exit(1)
  39
  40 ROWS = 24
  41 COLUMNS = 80
  42 CHILD_TIMEOUT = 5
  43
  44 ATTR_NORMAL = 0
  45 ATTR_REVERSE = 1
  46
  47 COLOR_BLACK = 0
  48 COLOR_RED = 1
  49 COLOR_GREEN = 2
  50 COLOR_YELLOW = 3
  51 COLOR_BLUE = 4
  52 COLOR_MAGENTA = 5
  53 COLOR_CYAN = 6
  54 COLOR_WHITE = 7
  55 COLOR_DEFAULT = 9
  56 COLOR_REGISTER = ((COLOR_BLACK, 'black'), (COLOR_RED, 'red'),
  57                   (COLOR_GREEN, 'green'), (COLOR_YELLOW, 'yellow'),
  58                   (COLOR_BLUE, 'blue'), (COLOR_MAGENTA, 'magenta'),
  59                   (COLOR_CYAN, 'cyan'), (COLOR_WHITE, 'white'),
  60                   (COLOR_DEFAULT, 'default'))
  61 COLOR_TO_STRING_MAP = {id_: name for id_, name in COLOR_REGISTER}
  62 STRING_TO_COLOR_MAP = {name: id_ for id_, name in COLOR_REGISTER}
  63 COLORS = {id_ for id_, _ in COLOR_REGISTER}
  64 REAL_COLOR_NAMES = tuple(
  65     name for id_, name in COLOR_REGISTER if id_ != COLOR_DEFAULT)
  66
  67 COLOR_DEFAULT_FOREGROUND = COLOR_BLACK
  68 COLOR_DEFAULT_BACKGROUND = COLOR_WHITE
  69
  70 CODE_ENTER = '\x0d'
  71 CODE_FN = ('\x1bOP', '\x1bOQ', '\x1bOR', '\x1bOS', '\x1b[15~', '\x1b[17~',
  72            '\x1b[18~', '\x1b[19~', '\x1b[20~', '\x1b[21~', '\x1b[23~')
  73 CODE_PAGE_UP = '\x1b[5~'
  74 CODE_PAGE_DOWN = '\x1b[6~'
  75
  76
  77 def attr_to_string(attr):
  78     """Get string representation of given attributes."""
  79
  80     res = []
  81     if (attr & ATTR_REVERSE) != 0:
  82         res.append('reverse')
  83     return '|'.join(res)
  84
  85
  86 def string_to_attr(string):
  87     """
  88     Convert a string to attributes. Exception ValueError is raised if some
  89     attribute is invalid.
  90     """
  91
  92     res = ATTR_NORMAL
  93     for attr in string.split('|'):
  94         if attr == 'normal':
  95             pass
  96         elif attr == 'reverse':
  97             res |= ATTR_REVERSE
  98         else:
  99             raise ValueError("Unrecognized attribute '{}'".format(attr))
 100     return res
 101
 102
 103 def color_to_string(color):
 104     """Get string representation of a given color."""
 105
 106     return COLOR_TO_STRING_MAP[color]
 107
 108
 109 def string_to_color(string):
 110     """
 111     Convert a string to a color. Exception ValueError is raised if the color
 112     name is not recognized.
 113     """
 114
 115     try:
 116         return STRING_TO_COLOR_MAP[string]
 117     except KeyError:
 118         raise ValueError("Unrecognized color '{}'".format(string))
 119
 120
 121 class TermChar:
 122     """On-screen character."""
 123
 124     def __init__(self, char=" ", attr=ATTR_NORMAL, fgcolor=COLOR_DEFAULT,
 125                  bgcolor=COLOR_DEFAULT):
 126         self.char = char
 127         self.attr = attr
 128         self.fgcolor = fgcolor
 129         self.bgcolor = bgcolor
 130
 131     def __eq__(self, other):
 132         return self.__dict__ == other.__dict__
 133
 134     def _get_translated_fgcolor(self):
 135         """
 136         Return the foreground color. If the current color is COLOR_DEFAULT then
 137         COLOR_DEFAULT_FOREGROUND is returned.
 138         """
 139
 140         if self.fgcolor == COLOR_DEFAULT:
 141             return COLOR_DEFAULT_FOREGROUND
 142         return self.fgcolor
 143
 144     def _get_translated_bgcolor(self):
 145         """
 146         Return the background color. If the current color is COLOR_DEFAULT then
 147         COLOR_DEFAULT_BACKGROUND is returned.
 148         """
 149
 150         if self.bgcolor == COLOR_DEFAULT:
 151             return COLOR_DEFAULT_BACKGROUND
 152         return self.bgcolor
 153
 154     def get_tag_foreground(self):
 155         """
 156         Return a name of the final foreground color that should be used to
 157         display the character on the screen.
 158         """
 159
 160         if self.attr & ATTR_REVERSE:
 161             color = self._get_translated_bgcolor()
 162         else:
 163             color = self._get_translated_fgcolor()
 164         return color_to_string(color)
 165
 166     def get_tag_background(self):
 167         """
 168         Return a name of the final background color that should be used to
 169         display the character on the screen.
 170         """
 171
 172         if self.attr & ATTR_REVERSE:
 173             color = self._get_translated_fgcolor()
 174         else:
 175             color = self._get_translated_bgcolor()
 176         return color_to_string(color)
 177
 178
 179 class Term:
 180     """Termex terminal emulator."""
 181
 182     MODE_RUN = 0
 183     MODE_RECORD = 1
 184     MODE_TEST = 2
 185
 186     class _TerminalConnectionException(Exception):
 187         """
 188         Exception reported when communication with the pseudo-terminal fails.
 189         """
 190         pass
 191
 192     def __init__(self, root, program, mode, terminfo=None):
 193         self._root = root
 194         self._program = program
 195         self._mode = mode
 196         self._terminfo = terminfo
 197
 198         # Test mode obtains the program name from the playbook.
 199         if self._mode == self.MODE_TEST:
 200             assert self._program is None
 201
 202         self._child_pid = None
 203         self._fd = None
 204
 205         self._screen = None
 206         self._cur_y = 0
 207         self._cur_x = 0
 208         self._attr = ATTR_NORMAL
 209         self._fgcolor = COLOR_DEFAULT
 210         self._bgcolor = COLOR_DEFAULT
 211         self._charbuf = b''
 212
 213         # Initialize the GUI if requested.
 214         if self._root:
 215             self._root.title("Termex")
 216             self._frame = tkinter.Frame(self._root)
 217
 218             self._text = tkinter.Text(self._root, height=ROWS, width=COLUMNS)
 219             self._text.config(
 220                 foreground=color_to_string(COLOR_DEFAULT_FOREGROUND),
 221                 background=color_to_string(COLOR_DEFAULT_BACKGROUND))
 222             self._text.pack()
 223
 224             # Configure tag values.
 225             for fgcolor_str in REAL_COLOR_NAMES:
 226                 for bgcolor_str in REAL_COLOR_NAMES:
 227                     tag = 'tag_{}-{}'.format(fgcolor_str, bgcolor_str)
 228                     self._text.tag_config(tag, foreground=fgcolor_str,
 229                                           background=bgcolor_str)
 230
 231         self._erase_all()
 232
 233         if self._mode == self.MODE_RECORD:
 234             self._test_e = ElementTree.Element('test')
 235             self._test_e.set('program', self._program)
 236
 237     def _start_program(self):
 238         """
 239         Fork, connect the child's controlling terminal to a pseudo-terminal and
 240         start the selected child program.
 241
 242         Parent behaviour: Returns True when the fork was successful, False
 243         otherwise. Note that the returned value does not provide information
 244         whether the exec call in the child process was successful or not. That
 245         must be determined by attempting communication with the child.
 246
 247         Child behaviour: Execs the selected program and does not return if the
 248         call was successful, returns False otherwise.
 249         """
 250
 251         # Fork and connect the child's controlling terminal to a
 252         # pseudo-terminal.
 253         try:
 254             self._child_pid, self._fd = pty.fork()
 255         except OSError as e:
 256             print("Fork to run '{}' failed: {}".format(self._program, e),
 257                   file=sys.stderr)
 258             return False
 259         if self._child_pid == 0:
 260             try:
 261                 env = {'PATH': '/bin:/usr/bin', 'TERM': 'termex',
 262                        'LC_ALL': 'en_US.UTF-8'}
 263                 if self._terminfo:
 264                     env['TERMINFO'] = self._terminfo
 265                 elif 'TERMINFO' in os.environ:
 266                     env['TERMINFO'] = os.environ['TERMINFO']
 267                 os.execle(self._program, self._program, env)
 268             except OSError as e:
 269                 print("Failed to execute '{}': {}".format(self._program, e),
 270                       file=sys.stderr)
 271                 return False
 272
 273         return True
 274
 275     def _finalize_program(self):
 276         """
 277         Close the connection to the pseudo-terminal and wait for the child
 278         program to complete. Returns True when the connection was successfully
 279         closed and the child completed in the timeout limit, False otherwise.
 280         """
 281
 282         res = True
 283
 284         # Close the file descriptor that is connected to the child's
 285         # controlling terminal.
 286         try:
 287             os.close(self._fd)
 288         except OSError as e:
 289             print("Failed to close file descriptor '{}' that is connected to "
 290                   "the child's controlling terminal: {}.".format(self._fd, e),
 291                   file=sys.stderr)
 292             res = False
 293
 294         # Wait for the child to finish. It should terminate now that its input
 295         # was closed.
 296         for _ in range(CHILD_TIMEOUT):
 297             try:
 298                 pid, _status = os.waitpid(self._child_pid, os.WNOHANG)
 299             except OSError as e:
 300                 print("Failed to wait on child '{}' to complete: "
 301                       "{}.".format(pid, e), file=sys.stderr)
 302                 res = False
 303                 break
 304             if pid != 0:
 305                 break
 306             time.sleep(1)
 307         else:
 308             print("Child '{}' has not completed.".format(self._child_pid),
 309                   file=sys.stderr)
 310             res = False
 311
 312         return res
 313
 314     def run_gui_mainloop(self):
 315         """Start the selected child program and run the tkinter's main loop."""
 316
 317         assert self._mode == self.MODE_RUN or self._mode == self.MODE_RECORD
 318
 319         # Start the specified program.
 320         if not self._start_program():
 321             return
 322
 323         try:
 324             # Prepare for running the main loop.
 325             self._root.createfilehandler(
 326                 self._fd, tkinter.READABLE,
 327                 lambda fd, mask: self._pty_callback())
 328             self._root.bind('<Key>', self._tk_key)
 329             self._root.bind('<<Quit>>', lambda e: self._quit_gui_mainloop())
 330             self._root.protocol('WM_DELETE_WINDOW', self._quit_gui_mainloop)
 331
 332             # Run the main loop.
 333             try:
 334                 self._root.mainloop()
 335             except self._TerminalConnectionException as e:
 336                 print("{}.".format(e), file=sys.stderr)
 337
 338             self._root.deletefilehandler(self._fd)
 339         finally:
 340             # Finalize the run of the child program.
 341             self._finalize_program()
 342
 343     def _quit_gui_mainloop(self):
 344         """Exit the tkinter's main loop."""
 345
 346         assert self._mode == self.MODE_RUN or self._mode == self.MODE_RECORD
 347         self._root.quit()
 348
 349     def _pty_callback(self):
 350         """
 351         Process a data event from the pseudo-terminal. Returns True when the
 352         connection to the pseudo-terminal was closed, False otherwise.
 353         Exception _TerminalConnectionException is raised if the read of the new
 354         data from the pseudo-terminal fails.
 355         """
 356
 357         closed = False
 358         try:
 359             char = os.read(self._fd, 1)
 360         except OSError as e:
 361             if e.errno == errno.EIO:
 362                 closed = True
 363             else:
 364                 raise self._TerminalConnectionException(
 365                     "Error reading from file descriptor '{}' that is "
 366                     "connected to the child's controlling terminal: "
 367                     "{}".format(self._fd, e))
 368
 369         # Check whether the descriptor referring to the pseudo-terminal slave
 370         # has been closed or end of file was reached.
 371         if closed or len(char) == 0:
 372             if self._root:
 373                 self._root.quit()
 374             return True
 375
 376         self._charbuf += char
 377
 378         if self._handle_sequence(self._charbuf):
 379             self._charbuf = b''
 380         else:
 381             # print("Unmatched {}.".format(self._charbuf), file=sys.stderr)
 382             pass
 383         return False
 384
 385     def _send_key(self, chars, name):
 386         """
 387         Write the specified characters that represent one key to the
 388         pseudo-terminal. If the recording mode is enabled then the specified
 389         key name is recorded in the test playbook. Exception
 390         _TerminalConnectionException is raised if the write to the
 391         pseudo-terminal fails.
 392         """
 393
 394         if self._mode == self.MODE_RECORD:
 395             # Record the key.
 396             action_e = ElementTree.SubElement(self._test_e, 'action')
 397             action_e.set('key', name)
 398             print("Recorded key '{}'.".format(name))
 399
 400         # Send the key to the terminal.
 401         try:
 402             os.write(self._fd, str.encode(chars))
 403         except OSError as e:
 404             raise self._TerminalConnectionException(
 405                 "Error writing characters '{}' to file descriptor '{}' that "
 406                 "is connected to the child's controlling terminal: "
 407                 "{}".format(chars, self._fd, e))
 408
 409     def _handle_sequence(self, seq):
 410         """
 411         Process a byte sequence received from the pseudo-terminal. Returns True
 412         when the sequence was recognized and successfully handled, False
 413         otherwise.
 414         """
 415
 416         if re.fullmatch(b'[^\x01-\x1f]+', seq):
 417             try:
 418                 uchar = seq.decode('utf-8')
 419                 self._print_char(
 420                     TermChar(uchar, self._attr, self._fgcolor, self._bgcolor))
 421                 return True
 422             except UnicodeError:
 423                 # Continue on the assumption that it is not yet a complete
 424                 # character. This assumption is wrong if the received text is
 425                 # actually malformed.
 426                 return False
 427
 428         if seq == b'\x07':
 429             # Bell.
 430             if self._root:
 431                 self._root.bell()
 432             return True
 433         if seq == b'\x08':
 434             # Backspace non-destructively.
 435             self._cur_x -= 1
 436             return True
 437         if seq == b'\x0d':
 438             # Go to beginning of line.
 439             self._cur_x = 0
 440             return True
 441         if seq == b'\x0a':
 442             # Move cursor down one line.
 443             self._cursor_down()
 444             return True
 445
 446         # Controls beginning with ESC.
 447
 448         # Control sequences.
 449         match = re.fullmatch(b'\x1b\\[([0-9]+)@', seq)
 450         if match:
 451             # Insert blank characters.
 452             self._insert_blanks(int(match.group(1)))
 453             return True
 454         if seq == b'\x1b[H':
 455             # Set cursor position to the default (top left).
 456             self._cur_y = 0
 457             self._cur_x = 0
 458             return True
 459         match = re.fullmatch(b'\x1b\\[([0-9]+);([0-9]+)H', seq)
 460         if match:
 461             # Set cursor position to (y,x).
 462             self._cur_y = int(match.group(1))
 463             self._cur_x = int(match.group(2))
 464             return True
 465         if self._charbuf == b'\x1b[K':
 466             # Erase in line to right.
 467             for x in range(self._cur_x, COLUMNS):
 468                 self._print_char_at(self._cur_y, x, TermChar())
 469             return True
 470         if seq == b'\x1b[2J':
 471             # Erase display completely.
 472             self._erase_all()
 473             return True
 474         if seq == b'\x1b[m':
 475             # Normal character attribute (all attributes off).
 476             self._attr = ATTR_NORMAL
 477             return True
 478         if seq == b'\x1b[7m':
 479             # Inverse character attribute.
 480             self._attr |= ATTR_REVERSE
 481             return True
 482         match = re.fullmatch(b'\x1b\\[3([0-9]+)m', seq)
 483         if match:
 484             # Set foreground color.
 485             color = int(match.group(1))
 486             if color in COLORS:
 487                 self._fgcolor = color
 488                 return True
 489             return False
 490         match = re.fullmatch(b'\x1b\\[4([0-9]+)m', seq)
 491         if match:
 492             # Set background color.
 493             color = int(match.group(1))
 494             if color in COLORS:
 495                 self._bgcolor = color
 496                 return True
 497             return False
 498         if seq == b'\x1b[?25l':
 499             # Hide cursor.
 500             return True
 501
 502         return False
 503
 504     def _cursor_down(self):
 505         """
 506         Move the screen cursor one line down. The screen is scrolled if the
 507         cursor points to the last line.
 508         """
 509
 510         if self._cur_y < ROWS - 1:
 511             self._cur_y += 1
 512         else:
 513             assert self._cur_y == ROWS - 1
 514
 515             # At the last line of the terminal, scroll up the screen.
 516             del self._screen[0]
 517             self._screen.append([TermChar() for x in range(COLUMNS)])
 518
 519             if self._root:
 520                 self._text.config(state=tkinter.NORMAL)
 521                 self._text.delete('1.0', '2.0')
 522                 self._text.insert(tkinter.END, "\n" + " " * COLUMNS)
 523                 self._text.config(state=tkinter.DISABLED)
 524
 525     def _erase_all(self):
 526         """Completely clear the terminal's screen."""
 527
 528         self._screen = [[TermChar() for x in range(COLUMNS)]
 529                         for y in range(ROWS)]
 530
 531         if self._root:
 532             self._text.config(state=tkinter.NORMAL)
 533             self._text.delete('1.0', tkinter.END)
 534             self._text.insert('1.0', "\n".join([" " * COLUMNS] * ROWS))
 535             self._text.config(state=tkinter.DISABLED)
 536
 537     def _insert_blanks(self, w):
 538         """
 539         Replace the specified number of characters on the current screen line
 540         with blanks.
 541         """
 542
 543         del self._screen[self._cur_y][-w:]
 544         pre = self._screen[self._cur_y][:self._cur_x]
 545         post = self._screen[self._cur_y][self._cur_x:]
 546         self._screen[self._cur_y] = pre + [TermChar() for x in range(w)] + post
 547
 548         if self._root:
 549             self._text.config(state=tkinter.NORMAL)
 550             self._text.delete('{}.end-{}c'.format(self._cur_y + 1, w),
 551                               '{}.end'.format(self._cur_y + 1))
 552             self._text.insert('{}.{}'.format(self._cur_y + 1, self._cur_x),
 553                               " " * w)
 554             self._text.config(state=tkinter.DISABLED)
 555
 556     def _print_char_at(self, y, x, char):
 557         """Output one character on the screen at the specified coordinates."""
 558
 559         # Record the character in the internal screen representation.
 560         self._screen[y][x] = char
 561
 562         if self._root:
 563             # Add the character to the terminal text widget.
 564             self._text.config(state=tkinter.NORMAL)
 565             pos = '{}.{}'.format(y + 1, x)
 566             self._text.delete(pos)
 567
 568             tag = 'tag_{}-{}'.format(char.get_tag_foreground(),
 569                                      char.get_tag_background())
 570             self._text.insert(pos, char.char, tag)
 571             self._text.config(state=tkinter.DISABLED)
 572
 573     def _print_char(self, char):
 574         """Output one character on the screen at the cursor position."""
 575
 576         self._print_char_at(self._cur_y, self._cur_x, char)
 577
 578         # Advance the cursor.
 579         self._cur_x += 1
 580         if self._cur_x == COLUMNS:
 581             self._cur_x = 0
 582             self._cursor_down()
 583
 584     def _tk_key(self, event):
 585         """Process a key pressed by the user."""
 586
 587         if len(event.char) != 0:
 588             if event.char == CODE_ENTER:
 589                 self._send_key(event.char, 'Enter')
 590             else:
 591                 self._send_key(event.char, event.char)
 592             return
 593
 594         # A special key was pressed.
 595         if event.keysym == 'F12':
 596             self._record_expected_screen()
 597             return
 598         if event.keysym == 'Prior':
 599             self._send_key(CODE_PAGE_UP, 'PageUp')
 600             return
 601         if event.keysym == 'Next':
 602             self._send_key(CODE_PAGE_DOWN, 'PageDown')
 603             return
 604         match = re.fullmatch('F([0-9]+)', event.keysym)
 605         if match:
 606             # F1 to F11.
 607             fnum = int(match.group(1))
 608             if fnum >= 1 and fnum <= len(CODE_FN):
 609                 self._send_key(CODE_FN[fnum - 1], event.keysym)
 610                 return
 611
 612         print("Unrecognized key {}.".format(event.keysym), file=sys.stderr)
 613
 614     def _get_screen_xml(self, screen):
 615         """
 616         Return an ElementTree.Element that represents the given screen.
 617         """
 618
 619         expect_e = ElementTree.Element('expect')
 620         data_e = ElementTree.SubElement(expect_e, 'data')
 621
 622         colors = {}
 623         new_key = 'a'
 624
 625         # Print content of the screen.
 626         for y in range(ROWS):
 627             line_e = ElementTree.SubElement(data_e, 'line')
 628             line_e.text = ''
 629
 630             attr = ''
 631
 632             for x in range(COLUMNS):
 633                 term_char = screen[y][x]
 634
 635                 line_e.text += term_char.char
 636
 637                 color = (term_char.attr, term_char.fgcolor, term_char.bgcolor)
 638                 if color == (ATTR_NORMAL, COLOR_DEFAULT, COLOR_DEFAULT):
 639                     key = ' '
 640                 elif color in colors:
 641                     key = colors[color]
 642                 else:
 643                     key = new_key
 644                     colors[color] = key
 645                     assert new_key != 'z'
 646                     new_key = chr(ord(new_key) + 1)
 647                 attr += key
 648
 649             # Record any non-default attributes/colors.
 650             if attr != ' ' * COLUMNS:
 651                 attr_e = ElementTree.SubElement(data_e, 'attr')
 652                 attr_e.text = attr
 653
 654         # Record used color schemes.
 655         if colors:
 656             scheme_e = ElementTree.SubElement(expect_e, 'scheme')
 657             for color, key in sorted(colors.items(), key=lambda x: x[1]):
 658                 attr, fgcolor, bgcolor = color
 659                 color_e = ElementTree.SubElement(scheme_e, 'color')
 660                 color_e.set('key', key)
 661
 662                 attr_str = attr_to_string(attr)
 663                 if attr_str:
 664                     color_e.set('attributes', attr_str)
 665
 666                 fgcolor_str = color_to_string(fgcolor)
 667                 if fgcolor_str:
 668                     color_e.set('foreground', fgcolor_str)
 669
 670                 bgcolor_str = color_to_string(bgcolor)
 671                 if bgcolor_str:
 672                     color_e.set('background', bgcolor_str)
 673
 674         return expect_e
 675
 676     def _record_expected_screen(self):
 677         """
 678         Record the current screen content as an expected screen in the test
 679         playbook that is being created.
 680         """
 681
 682         assert self._mode == self.MODE_RUN or self._mode == self.MODE_RECORD
 683
 684         if self._mode != self.MODE_RECORD:
 685             print("Recording is not enabled.", file=sys.stderr)
 686             return
 687
 688         expect_e = self._get_screen_xml(self._screen)
 689         self._test_e.append(expect_e)
 690         print("Recorded expected screen.")
 691
 692     # Method _indent_xml() is based on a code from
 693     # http://effbot.org/zone/element-lib.htm#prettyprint.
 694     def _indent_xml(self, elem, level=0):
 695         """
 696         Indent elements of a given ElementTree so it can be pretty-printed.
 697         """
 698
 699         i = '\n' + '\t' * level
 700         if len(elem):
 701             if not elem.text or not elem.text.strip():
 702                 elem.text = i + '\t'
 703             for e in elem:
 704                 self._indent_xml(e, level+1)
 705             if not e.tail or not e.tail.strip():
 706                 e.tail = i
 707         if not elem.tail or not elem.tail.strip():
 708             elem.tail = i
 709
 710     def output_test(self, filename):
 711         """
 712         Output a recorded playbook to a file with the given name. Returns True
 713         when the writing of the test data succeeded, False otherwise.
 714         """
 715
 716         assert self._mode == self.MODE_RECORD
 717
 718         # Pretty-format the XML tree.
 719         self._indent_xml(self._test_e)
 720
 721         # Output the test.
 722         tree = ElementTree.ElementTree(self._test_e)
 723         try:
 724             tree.write(filename, 'unicode', True)
 725         except Exception as e:
 726             print("Failed to write playbook file '{}': {}.".format(
 727                 filename, e), file=sys.stderr)
 728             return False
 729
 730         return True
 731
 732     class _TestFailure(Exception):
 733         """Exception reported when a test failed."""
 734         pass
 735
 736     def _playbook_key(self, cmd_e):
 737         """
 738         Parse a description of one key action and send the key to the terminal.
 739         Exception _TestFailure is raised if the description is malformed or
 740         incomplete, exception _TerminalConnectionException can be thrown when
 741         communication with the pseudo-terminal fails.
 742         """
 743
 744         assert self._mode == self.MODE_TEST
 745
 746         try:
 747             key = cmd_e.attrib['key']
 748         except KeyError:
 749             raise self._TestFailure("Element 'action' is missing required "
 750                                     "attribute 'key'")
 751
 752         # Handle simple characters.
 753         if len(key) == 1:
 754             self._send_key(key, key)
 755             return
 756
 757         # Handle special keys.
 758         if key == 'Enter':
 759             self._send_key(CODE_ENTER, key)
 760             return
 761         if key == 'PageUp':
 762             self._send_key(CODE_PAGE_UP, key)
 763             return
 764         if key == 'PageDown':
 765             self._send_key(CODE_PAGE_DOWN, key)
 766         match = re.fullmatch('F([0-9]+)', key)
 767         if match:
 768             # F1 to F11.
 769             fnum = int(match.group(1))
 770             if fnum >= 1 and fnum <= len(CODE_FN):
 771                 self._send_key(CODE_FN[fnum - 1], key)
 772                 return
 773
 774         raise self._TestFailure(
 775             "Element 'action' specifies unrecognized key '{}'".format(key))
 776
 777     def _parse_color_scheme(self, scheme_e):
 778         """
 779         Parse color scheme of one expected screen. Dictionary with
 780         {'key': (attr, fgcolor, bgcolor), ...} is returned on success,
 781         exception _TestFailure is raised if the description is malformed or
 782         incomplete.
 783         """
 784
 785         assert self._mode == self.MODE_TEST
 786
 787         colors = {}
 788         for color_e in scheme_e:
 789             try:
 790                 key = color_e.attrib['key']
 791             except KeyError:
 792                 raise self._TestFailure(
 793                     "Element 'color' is missing required attribute 'key'")
 794
 795             attr = ATTR_NORMAL
 796             if 'attributes' in color_e.attrib:
 797                 try:
 798                     attr = string_to_attr(color_e.attrib['attributes'])
 799                 except ValueError as e:
 800                     raise self._TestFailure(
 801                         "Value of attribute 'attributes' is invalid: "
 802                         "{}".format(e))
 803
 804             fgcolor = COLOR_DEFAULT
 805             if 'foreground' in color_e.attrib:
 806                 try:
 807                     fgcolor = string_to_color(color_e.attrib['foreground'])
 808                 except ValueError as e:
 809                     raise self._TestFailure(
 810                         "Value of attribute 'foreground' is invalid: "
 811                         "{}".format(e))
 812
 813             bgcolor = COLOR_DEFAULT
 814             if 'background' in color_e.attrib:
 815                 try:
 816                     bgcolor = string_to_color(color_e.attrib['background'])
 817                 except ValueError as e:
 818                     raise self._TestFailure(
 819                         "Value of attribute 'background' is invalid: "
 820                         "{}".format(e))
 821
 822             colors[key] = (attr, fgcolor, bgcolor)
 823
 824         return colors
 825
 826     def _parse_screen_data(self, data_e, colors):
 827         """
 828         Parse screen lines of one expected screen. Internal screen
 829         representation is returned on success, exception _TestFailure is raised
 830         if the description is malformed or incomplete.
 831         """
 832
 833         assert self._mode == self.MODE_TEST
 834
 835         NEW_LINE = 0
 836         NEW_LINE_OR_ATTR = 1
 837         state = NEW_LINE
 838         line = None
 839         expected_screen = []
 840
 841         for data_sub_e in data_e:
 842             # Do common processing for both states.
 843             if data_sub_e.tag == 'line':
 844                 # Append the previous line.
 845                 if line:
 846                     expected_screen.append(line)
 847                 # Parse the new line.
 848                 line = [TermChar(char) for char in data_sub_e.text]
 849                 state = NEW_LINE_OR_ATTR
 850
 851             if state == NEW_LINE and data_sub_e.tag != 'line':
 852                 raise self._TestFailure("Element '{}' is invalid, expected "
 853                                         "'line'".format(data_sub_e.tag))
 854
 855             elif state == NEW_LINE_OR_ATTR:
 856                 if data_sub_e.tag == 'attr':
 857                     if len(data_sub_e.text) != len(line):
 858                         raise self._TestFailure(
 859                             "Element 'attr' does not match the previous line, "
 860                             "expected '{}' attribute characters but got "
 861                             "'{}'".format(len(line), len(data_sub_e.text)))
 862
 863                     for i, key in enumerate(data_sub_e.text):
 864                         if key == ' ':
 865                             continue
 866
 867                         try:
 868                             attr, fgcolor, bgcolor = colors[key]
 869                         except KeyError:
 870                             raise self._TestFailure("Color attribute '{}' is "
 871                                                     "not defined".format(key))
 872                         line[i].attr = attr
 873                         line[i].fgcolor = fgcolor
 874                         line[i].bgcolor = bgcolor
 875
 876                     state = NEW_LINE
 877
 878                 elif data_sub_e.tag != 'line':
 879                     raise self._TestFailure(
 880                         "Element '{}' is invalid, expected 'line' or "
 881                         "'attr'".format(data_sub_e.tag))
 882
 883         # Append the final line.
 884         if line:
 885             expected_screen.append(line)
 886
 887         return expected_screen
 888
 889     def _parse_expected_screen(self, expect_e):
 890         """
 891         Parse a description of one expected screen. Internal screen
 892         representation is returned on success, exception _TestFailure is raised
 893         if the description is malformed or incomplete.
 894         """
 895
 896         assert self._mode == self.MODE_TEST
 897
 898         data_e = None
 899         scheme_e = None
 900         for sub_e in expect_e:
 901             if sub_e.tag == 'data':
 902                 if data_e:
 903                     raise self._TestFailure("Element 'expect' contains "
 904                                             "multiple 'data' sub-elements")
 905                 data_e = sub_e
 906             elif sub_e.tag == 'scheme':
 907                 if scheme_e:
 908                     raise self._TestFailure("Element 'expect' contains "
 909                                             "multiple 'scheme' sub-elements")
 910                 scheme_e = sub_e
 911
 912         if not data_e:
 913             raise self._TestFailure(
 914                 "Element 'expect' is missing required sub-element 'data'")
 915
 916         # Parse the color scheme.
 917         if scheme_e:
 918             colors = self._parse_color_scheme(scheme_e)
 919         else:
 920             colors = {}
 921
 922         # Parse the screen data.
 923         return self._parse_screen_data(data_e, colors)
 924
 925     def _report_failed_expectation(self, expected_screen):
 926         """
 927         Report that the expected screen state has not been reached. The output
 928         consists of the expected screen, the current screen content, followed
 929         by differences between the two screens.
 930         """
 931
 932         assert self._mode == self.MODE_TEST
 933
 934         # Print the expected screen. The output is not verbatim as it was
 935         # specified in the input file, but instead the screen is printed in the
 936         # same way the current screen gets output. This allows to properly show
 937         # differences between the two screens.
 938
 939         expected_screen_e = self._get_screen_xml(expected_screen)
 940         self._indent_xml(expected_screen_e)
 941         expected_screen_str = ElementTree.tostring(
 942             expected_screen_e, 'unicode')
 943         print("Expected (normalized) screen:", file=sys.stderr)
 944         print(expected_screen_str, file=sys.stderr)
 945
 946         # Print the current screen.
 947         current_screen_e = self._get_screen_xml(self._screen)
 948         self._indent_xml(current_screen_e)
 949         current_screen_str = ElementTree.tostring(current_screen_e, 'unicode')
 950         print("Current screen:", file=sys.stderr)
 951         print(current_screen_str, file=sys.stderr)
 952
 953         # Print the delta.
 954         print("Differences:", file=sys.stderr)
 955         sys.stderr.writelines(difflib.unified_diff(
 956             expected_screen_str.splitlines(keepends=True),
 957             current_screen_str.splitlines(keepends=True),
 958             fromfile="Expected screen", tofile="Current screen"))
 959
 960     def _execute_playbook(self, test_e):
 961         """
 962         Run the main loop and execute the given test playbook. Normal return
 963         from the method indicates that the test succeeded. Exception
 964         _TestFailure is raised when the test fails and exception
 965         _TerminalConnectionException can be thrown when communication with the
 966         pseudo-terminal fails.
 967         """
 968
 969         assert self._mode == self.MODE_TEST
 970
 971         cmd_iter = iter(test_e)
 972
 973         # Start the main loop.
 974         with selectors.DefaultSelector() as sel:
 975             sel.register(self._fd, selectors.EVENT_READ)
 976
 977             expected_screen = None
 978             more_commands = True
 979             while True:
 980                 # Process any actions and find an expected screen.
 981                 while not expected_screen and more_commands:
 982                     try:
 983                         cmd_e = next(cmd_iter)
 984                         if cmd_e.tag == 'action':
 985                             self._playbook_key(cmd_e)
 986                         elif cmd_e.tag == 'expect':
 987                             expected_screen = self._parse_expected_screen(
 988                                 cmd_e)
 989                             # Stop processing more commands for now and wait
 990                             # for the expected screen to appear.
 991                             break
 992                         else:
 993                             raise self._TestFailure(
 994                                 "Element '{}' is invalid, expected 'action' "
 995                                 "or 'expect'".format(cmd_e.tag))
 996                     except StopIteration:
 997                         # No more commands.
 998                         more_commands = False
 999
1000                 # Wait for the expected screen.
1001                 events = sel.select(CHILD_TIMEOUT)
1002                 if not events:
1003                     if expected_screen:
1004                         self._report_failed_expectation(expected_screen)
1005                     raise self._TestFailure(
1006                         "Timeout reached. No event received in the last {} "
1007                         "second(s)".format(CHILD_TIMEOUT))
1008
1009                 # Expect only an event on self._fd.
1010                 assert len(events) == 1
1011                 event = events[0]
1012                 key, _mask = event
1013                 assert key.fd == self._fd
1014
1015                 closed = self._pty_callback()
1016                 if closed:
1017                     if more_commands:
1018                         raise self._TestFailure(
1019                             "Connection to the terminal was closed but the "
1020                             "playbook contains more commands")
1021                     break
1022
1023                 # Check if the expected screen is present.
1024                 if self._screen == expected_screen:
1025                     expected_screen = None
1026
1027     def execute_test(self, filename):
1028         """
1029         Load test data from a given file, start the program under the test and
1030         execute the test playbook. Returns True when the test succeeded, False
1031         otherwise.
1032         """
1033
1034         assert self._mode == self.MODE_TEST
1035
1036         # Read the test data.
1037         try:
1038             tree = ElementTree.ElementTree(file=filename)
1039         except Exception as e:
1040             print("Failed to read playbook file '{}': {}.".format(filename, e),
1041                   file=sys.stderr)
1042             return False
1043
1044         # Read what program to execute.
1045         test_e = tree.getroot()
1046         if test_e.tag != 'test':
1047             print("Root element '{}' is invalid, expected 'test'.".format(
1048                   test_e.tag), file=sys.stderr)
1049             return False
1050         try:
1051             self._program = test_e.attrib['program']
1052         except KeyError:
1053             print("Element 'test' is missing required attribute 'program'.",
1054                   file=sys.stderr)
1055             return False
1056
1057         # Start the specified program.
1058         if not self._start_program():
1059             return False
1060
1061         # Execute the test playbook.
1062         res = True
1063         try:
1064             self._execute_playbook(tree.getroot())
1065         except (self._TerminalConnectionException, self._TestFailure) as e:
1066             print("{}.".format(e), file=sys.stderr)
1067             res = False
1068         finally:
1069             # Finalize the run of the child program.
1070             if not self._finalize_program():
1071                 res = False
1072
1073         # Return whether the test passed.
1074         return res
1075
1076
1077 def main():
1078     """
1079     Parse command line arguments and execute the operation that the user
1080     selected. Returns 0 if the operation was successful and a non-zero value
1081     otherwise.
1082     """
1083
1084     # Parse command line arguments.
1085     parser = argparse.ArgumentParser()
1086     parser.add_argument(
1087         '-t', '--terminfo', metavar='PATH', help="path to terminfo directory")
1088
1089     subparsers = parser.add_subparsers(dest='program')
1090     subparsers.required = True
1091
1092     program_parser = argparse.ArgumentParser(add_help=False)
1093     program_parser.add_argument('program', help="executable to run")
1094
1095     # Create the parser for the 'run' command.
1096     parser_run = subparsers.add_parser(
1097         'run', parents=[program_parser], help="run a program")
1098     parser_run.set_defaults(mode=Term.MODE_RUN)
1099
1100     # Create the parser for the 'record' command.
1101     parser_record = subparsers.add_parser(
1102         'record', parents=[program_parser], help="record a test")
1103     parser_record.set_defaults(mode=Term.MODE_RECORD)
1104     parser_record.add_argument(
1105         '-o', '--playbook', metavar='FILE', required=True,
1106         help="output playbook file")
1107
1108     # Create the parser for the 'test' command.
1109     parser_test = subparsers.add_parser('test', help="perform a test")
1110     parser_test.set_defaults(program=None)
1111     parser_test.set_defaults(mode=Term.MODE_TEST)
1112     parser_test.add_argument('playbook', help="input playbook file")
1113
1114     args = parser.parse_args()
1115
1116     tk_root = None
1117     if args.mode in (Term.MODE_RUN, Term.MODE_RECORD):
1118         # Start the terminal GUI.
1119         global tkinter
1120         import tkinter
1121
1122         try:
1123             tk_root = tkinter.Tk()
1124         except tkinter.TclError as e:
1125             print("Failed to initialize GUI: {}.".format(e), file=sys.stderr)
1126             return 1
1127
1128     term = Term(tk_root, args.program, args.mode, args.terminfo)
1129     if tk_root:
1130         # Start the GUI main loop.
1131         term.run_gui_mainloop()
1132     else:
1133         # Execute and check the playbook, without running GUI.
1134         ok = term.execute_test(args.playbook)
1135         if ok:
1136             msg = "succeeded"
1137             res = 0
1138         else:
1139             msg = "failed"
1140             res = 1
1141
1142         print("Checking of playbook '{}' {}.".format(args.playbook, msg))
1143         return res
1144
1145     if args.mode == Term.MODE_RECORD:
1146         # Get the recorded test data and write them to a file.
1147         if not term.output_test(args.playbook):
1148             return 1
1149
1150     return 0
1151
1152
1153 if __name__ == '__main__':
1154     sys.exit(main())
1155
1156 # vim: set tabstop=4 shiftwidth=4 textwidth=79 expandtab :