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