Add test for terminating the terminals

.github/workflows/check-release.yml

@@ -1,9 +1,12 @@
 name: Check Release
 on:
   push:
-    branches: ["master"]
+    branches: ["main"]
   pull_request:
-    branches: ["*"]
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.sha }}
+  cancel-in-progress: true
 
 jobs:
   check_release:

.github/workflows/test.yml

@@ -1,11 +1,16 @@
 name: Tests
+
 on:
   push:
-    branches: "*"
+    branches: ["main"]
   pull_request:
-    branches: "*"
   schedule:    
     - cron: "0 8 * * *"
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.sha }}
+  cancel-in-progress: true
+
 jobs:
   test:
     runs-on: ${{ matrix.os }}-latest

terminado/management.py

@@ -25,6 +25,7 @@ def preexec_fn():
 except ImportError:
     try:
         from winpty import PtyProcess as PtyProcessUnicode
+        from winpty.enums import Backend
     except ImportError:
         PtyProcessUnicode = object
     preexec_fn = None
@@ -45,6 +46,8 @@ def __init__(self, argv, env=[], cwd=None):
         kwargs = dict(argv=argv, env=env, cwd=cwd)
         if preexec_fn is not None:
             kwargs["preexec_fn"] = preexec_fn
+        if os.name == 'nt':
+            kwargs.setdefault('backend', Backend.WinPTY)
         self.ptyproc = PtyProcessUnicode.spawn(**kwargs)
         # The output might not be strictly UTF-8 encoded, so
         # we replace the inner decoder of PtyProcessUnicode

terminado/tests/basic_test.py

@@ -90,17 +90,9 @@ async def get_pid(self):
         await self.write_stdin("echo $$\r")
         (stdout, extra) = await self.read_stdout()
         if os.name == 'nt':
-            print(repr(stdout))
-            match = re.search(r'echo \$\$\\x1b\[71X\\x1b\[71C\\r\\n(\d+)', repr(stdout))
-            if match is None:
-                match = re.search(r'echo \$\$ \\r\\n(\d+)', repr(stdout))
-            if match is None:
-                match = re.search(
-                    r'echo \$\$ \\r\\n\\x1b\[\?25h\\x1b\[\?25l(\d+)',
-                    repr(stdout))
+            match = re.search(r'echo \$\$\\.*?\\r\\n(\d+)', repr(stdout))
             pid = int(match.groups()[0])
         else:
-            print('stdout=%r, extra=%r' % (stdout, extra))
             pid = int(stdout.split('\n')[1])
         return pid
 
@@ -216,6 +208,12 @@ async def test_namespace(self):
         self.assertEqual(pids[2], pids[3])
         self.assertNotEqual(pids[0], pids[3])
 
+        terminal = self.named_tm.terminals["1"]
+        killed = await terminal.terminate(True)
+        assert killed
+        assert not terminal.ptyproc.isalive()
+        assert terminal.ptyproc.closed
+
     @tornado.testing.gen_test
     @pytest.mark.skipif('linux' not in platform, reason='It only works on Linux')
     async def test_max_terminals(self):
@@ -235,6 +233,10 @@ async def test_single_process(self):
         pids = await self.get_pids(tms)
         self.assertEqual(pids[0], pids[1])
 
+        killed = await self.single_tm.terminal.terminate(True)
+        assert killed
+        assert self.single_tm.terminal.ptyproc.closed
+
 class UniqueTermTests(TermTestCase):
     @tornado.testing.gen_test
     async def test_unique_processes(self):