[spec/stateful] Implemented initial test param-passing for stateful tests needing term dimensions

Author: Andy C

spec/stateful/bind.py

@@ -15,7 +15,7 @@
 import pyte
 
 import harness
-from harness import expect_prompt, register
+from harness import expect_prompt, register, TerminalDimensionEnvVars
 
 from test.spec_lib import log
 
@@ -181,8 +181,8 @@ def bind_x_readline_point(sh):
     sh.expect("READLINE_POINT is unset")
 
 
-@register(not_impl_shells=['dash', 'mksh'])
-def bind_x_set_readline_point_to_insert(sh):
+@register(not_impl_shells=['dash', 'mksh'], needs_dimensions=True)
+def bind_x_set_readline_point_to_insert(sh, test_params):
     "test bind -x for correctly using $READLINE_POINT to overwrite the cmd"
 
     failing = "failing"
@@ -193,47 +193,48 @@ def bind_x_set_readline_point_to_insert(sh):
     bind_cmd = r"""bind -x '"\C-y": READLINE_POINT=%d' """ % new_rl_point
 
     try:
-        num_lines = int(os.environ['LINES'])
-        num_columns = int(os.environ['COLUMNS'])
+        num_lines = test_params['num_lines']
+        num_columns = test_params['num_columns']
     except KeyError:
-        raise RuntimeError("LINES and COLUMNS must be set")
+        raise RuntimeError("num_lines and num_columns must be passed in")
 
-    screen = pyte.Screen(num_columns, num_lines)
-    stream = pyte.Stream(screen)
+    with TerminalDimensionEnvVars(num_lines, num_columns):
+        screen = pyte.Screen(num_columns, num_lines)
+        stream = pyte.Stream(screen)
 
-    # Need to echo, because we don't want just output
-    sh.setecho(True)
+        # Need to echo, because we don't want just output
+        sh.setecho(True)
 
-    def _emulate_ansi_terminal(raw_output):
-        stream.feed(raw_output)
+        def _emulate_ansi_terminal(raw_output):
+            stream.feed(raw_output)
 
-        lines = screen.display
-        screen.reset()
+            lines = screen.display
+            screen.reset()
 
-        return '\n'.join(lines)
+            return '\n'.join(lines)
 
-    # sh.sendline('stty -icanon')
-    # time.sleep(0.1)
+        # sh.sendline('stty -icanon')
+        # time.sleep(0.1)
 
-    sh.sendline(bind_cmd)
-    time.sleep(0.1)
+        sh.sendline(bind_cmd)
+        time.sleep(0.1)
 
-    sh.send(echo_cmd)
-    time.sleep(0.1)
+        sh.send(echo_cmd)
+        time.sleep(0.1)
 
-    sh.sendcontrol('y')
-    time.sleep(0.2)
+        sh.sendcontrol('y')
+        time.sleep(0.2)
 
-    sh.send("no longer ")
-    time.sleep(0.1)
+        sh.send("no longer ")
+        time.sleep(0.1)
 
-    sh.expect(pexpect.TIMEOUT, timeout=2)
+        sh.expect(pexpect.TIMEOUT, timeout=2)
 
-    screen_contents = _emulate_ansi_terminal(sh.before)
-    if expected_cmd not in screen_contents:
-        raise Exception(
-            f"Expected command '{expected_cmd}' not found in screen contents:\n{screen_contents}"
-        )
+        screen_contents = _emulate_ansi_terminal(sh.before)
+        if expected_cmd not in screen_contents:
+            raise Exception(
+                f"Expected command '{expected_cmd}' not found in screen contents:\n{screen_contents}"
+            )
 
 
 @register(not_impl_shells=['dash', 'mksh'])

spec/stateful/harness.py

@@ -11,6 +11,7 @@
 import optparse
 import os
 import pexpect
+import re
 import signal
 import sys
 
@@ -49,12 +50,13 @@ def stop_process__hack(name, sig_num=signal.SIGSTOP):
 CASES = []
 
 
-def register(skip_shells=None, not_impl_shells=None):
+def register(skip_shells=None, not_impl_shells=None, needs_dimensions=False):
     skip_shells = skip_shells or []
     not_impl_shells = not_impl_shells or []
 
     def decorator(func):
-        CASES.append((func.__doc__, func, skip_shells, not_impl_shells))
+        CASES.append((func.__doc__, func, skip_shells, not_impl_shells,
+                      needs_dimensions))
         return func
 
     return decorator
@@ -67,6 +69,40 @@ class Result(object):
     FAIL = 4
 
 
+class TerminalDimensionEnvVars:
+    """
+    Context manager for setting and unsetting LINES and COLUMNS environment variables.
+    """
+
+    def __init__(self, lines: int, columns: int):
+        self.lines = lines
+        self.columns = columns
+        self.original_lines = None
+        self.original_columns = None
+
+    def __enter__(self):
+        # Save original values
+        self.original_lines = os.environ.get('LINES')
+        self.original_columns = os.environ.get('COLUMNS')
+
+        # Set new values
+        os.environ['LINES'] = str(self.lines)
+        os.environ['COLUMNS'] = str(self.columns)
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        # Restore original values
+        if self.original_lines is None:
+            del os.environ['LINES']
+        else:
+            os.environ['LINES'] = self.original_lines
+
+        if self.original_columns is None:
+            del os.environ['COLUMNS']
+        else:
+            os.environ['COLUMNS'] = self.original_columns
+
+
 class TestRunner(object):
 
     def __init__(self, num_retries, pexpect_timeout, verbose, num_lines,
@@ -77,7 +113,7 @@ def __init__(self, num_retries, pexpect_timeout, verbose, num_lines,
         self.num_lines = num_lines
         self.num_columns = num_columns
 
-    def RunOnce(self, shell_path, shell_label, func):
+    def RunOnce(self, shell_path, shell_label, func, test_params={}):
         sh_argv = []
         if shell_label in ('bash', 'osh'):
             sh_argv.extend(['--rcfile', '/dev/null'])
@@ -87,57 +123,44 @@ def RunOnce(self, shell_path, shell_label, func):
             sh_argv.append('--norc')
         #print(sh_argv)
 
-        # Set LINES and COLUMNS in case a program needs them (like pyte tests)
-        # Setting the dimensions kw param to spawn() is not sufficient
-        original_lines = os.environ.get('LINES')
-        original_columns = os.environ.get('COLUMNS')
-        os.environ['LINES'] = str(self.num_lines)
-        os.environ['COLUMNS'] = str(self.num_columns)
+        # Python 3: encoding required
+        sh = pexpect.spawn(
+            shell_path,
+            sh_argv,
+            encoding="utf-8",
+            dimensions=(self.num_lines, self.num_columns),
+            # Generally don't want local echo of input, it gets confusing fast.
+            echo=False,
+            timeout=self.pexpect_timeout,
+        )
 
-        try:
-            # Python 3: encoding required
-            sh = pexpect.spawn(
-                shell_path,
-                sh_argv,
-                encoding="utf-8",
-                dimensions=(self.num_lines, self.num_columns),
-                # Generally don't want local echo of input, it gets confusing fast.
-                echo=False,
-                timeout=self.pexpect_timeout,
-            )
-
-            sh.shell_label = shell_label  # for tests to use
-
-            if self.verbose:
-                sh.logfile = sys.stdout
-
-            ok = True
-            try:
-                func(sh)
-            except Exception as e:
-                import traceback
-                traceback.print_exc(file=sys.stderr)
-                return Result.FAIL
-                ok = False
+        sh.shell_label = shell_label  # for tests to use
 
-            finally:
-                sh.close()
+        if self.verbose:
+            sh.logfile = sys.stdout
 
-            if ok:
-                return Result.OK
+        ok = True
+        try:
+            # Support tests that need extra params (like dimensions), without
+            # impacting existing tests
+            if len(test_params) > 0:
+                func(sh, test_params)
+            else:
+                func(sh)
+        except Exception:
+            import traceback
+            traceback.print_exc(file=sys.stderr)
+            return Result.FAIL
+            ok = False
 
         finally:
-            if original_lines is None:
-                del os.environ['LINES']
-            else:
-                os.environ['LINES'] = original_lines
-            if original_columns is None:
-                del os.environ['COLUMNS']
-            else:
-                os.environ['COLUMNS'] = original_columns
+            sh.close()
 
-    def RunCase(self, shell_path, shell_label, func):
-        result = self.RunOnce(shell_path, shell_label, func)
+        if ok:
+            return Result.OK
+
+    def RunCase(self, shell_path, shell_label, func, test_params={}):
+        result = self.RunOnce(shell_path, shell_label, func, test_params)
 
         if result == Result.OK:
             return result, -1  # short circuit for speed
@@ -148,7 +171,8 @@ def RunCase(self, shell_path, shell_label, func):
                 log('\tFAILED first time: Retrying 4 times')
                 for i in range(self.num_retries):
                     log('\tRetry %d of %d', i + 1, self.num_retries)
-                    result = self.RunOnce(shell_path, shell_label, func)
+                    result = self.RunOnce(shell_path, shell_label, func,
+                                          test_params)
                     if result == Result.OK:
                         num_success += 1
             else:
@@ -164,13 +188,18 @@ def RunCase(self, shell_path, shell_label, func):
 
     def RunCases(self, cases, case_predicate, shell_pairs, result_table,
                  flaky):
-        for case_num, (desc, func, skip_shells,
-                       not_impl_shells) in enumerate(cases):
+        for case_num, (desc, func, skip_shells, not_impl_shells,
+                       needs_dimensions) in enumerate(cases):
             if not case_predicate(case_num, desc):
                 continue
 
             result_row = [case_num]
 
+            test_params = {}
+            if needs_dimensions:
+                test_params['num_lines'] = self.num_lines
+                test_params['num_columns'] = self.num_columns
+
             for shell_label, shell_path in shell_pairs:
                 skip_str = ''
                 if shell_label in skip_shells:
@@ -195,7 +224,8 @@ def RunCases(self, cases, case_predicate, shell_pairs, result_table,
                     flaky[case_num, shell_label] = -1
                     continue
 
-                result, retries = self.RunCase(shell_path, shell_label, func)
+                result, retries = self.RunCase(shell_path, shell_label, func,
+                                               test_params)
                 flaky[case_num, shell_label] = retries
 
                 result_row.append(result)
@@ -360,7 +390,7 @@ def main(argv):
 
     # List test cases and return
     if opts.do_list:
-        for i, (desc, _, _, _) in enumerate(CASES):
+        for i, (desc, *_) in enumerate(CASES):
             print('%d\t%s' % (i, desc))
         return
 

test/stateful.sh

@@ -55,7 +55,7 @@ job-control() {
 }
 
 bind() {
-  spec/stateful/bind.py $FIRST --oils-failures-allowed 4 "$@"
+  spec/stateful/bind.py $FIRST --oils-failures-allowed 4 --num-lines 24 --num-columns 80 "$@"
 }
 
 # Run on just 2 shells