Merge pull request #18 from rekter0/winProcessFix

Fix WinProcess

ptrlib/connection/tube.py

@@ -304,7 +304,6 @@ def __enter__(self):
     def __exit__(self, e_type, e_value, traceback):
         self.close()
 
-    @abstractmethod
     def is_alive(self) -> bool:
         pass
 

tests/connection/test_windows_proc.py

@@ -0,0 +1,52 @@
+import unittest
+import os
+import random
+from ptrlib import Process, is_scanf_safe
+from logging import getLogger, FATAL
+
+
+class TestProcess(unittest.TestCase):
+    def setUp(self):
+        getLogger("ptrlib").setLevel(FATAL)
+
+    def test_basic(self):
+        while True:
+            msg = os.urandom(16)
+            if is_scanf_safe(msg):
+                break
+
+        p = Process("./tests/test.bin/test_echo.pe.exe")
+
+        # send / recv
+        p.sendline(b"Message : " + msg)
+        self.assertEqual(p.recvlineafter(" : "), msg)
+
+        # send / recvregex
+        a, b = random.randrange(1<<32), random.randrange(1<<32)
+        p.sendline("Hello 0x{:08x}, 0x{:08x}".format(a, b))
+        r = p.recvregex("0x([0-9a-f]+), 0x([0-9a-f]+)")
+        p.recvline()
+        self.assertEqual(int(r[0], 16), a)
+        self.assertEqual(int(r[1], 16), b)
+
+        # shutdown
+        p.send(msg[::-1])
+        p.shutdown('write')
+        self.assertEqual(p.recvonce(len(msg)), msg[::-1])
+
+
+        p.close()
+
+    def test_timeout(self):
+        p = Process("./tests/test.bin/test_echo.pe.exe")
+        try:
+            p.recvuntil("*** never expected ***", timeout=1)
+            result = False
+        except TimeoutError:
+            result = True
+        except:
+            result = False
+        finally:
+            p.close()
+
+        self.assertEqual(result, True)