[py-svn] r31679 - in py/branch/distributed/py/test/rsession: . testing

[fijal at codespeak.net]

Author: fijal
Date: Sat Aug 26 14:53:09 2006
New Revision: 31679

Added:
   py/branch/distributed/py/test/rsession/box.py   (contents, props changed)
   py/branch/distributed/py/test/rsession/testing/test_boxing.py   (contents, props changed)
Modified:
   py/branch/distributed/py/test/rsession/testing/test_report.py
Log:
Added basic boxing stuff.


Added: py/branch/distributed/py/test/rsession/box.py
==============================================================================
--- (empty file)
+++ py/branch/distributed/py/test/rsession/box.py	Sat Aug 26 14:53:09 2006
@@ -0,0 +1,57 @@
+
+""" boxing - wrapping process with another process so we can run
+a process inside and see if it crashes
+"""
+
+import py
+import os
+import sys
+
+PYTESTSTDOUT = "pyteststdout"
+PYTESTSTDERR = "pyteststderr"
+
+class Box(object):
+    def __init__(self, fun):
+        self.fun = fun
+    
+    def run(self):
+        # What to do:
+        # XXX: Creates it in a current dir, and just open it, does not
+        # seem to be sane idea
+        os.mkfifo(PYTESTSTDOUT)
+        os.mkfifo(PYTESTSTDERR)
+        pid = os.fork()
+        if pid:
+            self.parent()
+        else:
+            self.children()
+            os._exit(0)
+    
+    def children(self):
+        # right now we need to call a function, but first we need to
+        # map all IO that might happen
+        fdstdout = os.open(PYTESTSTDOUT, os.O_WRONLY)
+        fdstderr = os.open(PYTESTSTDERR, os.O_WRONLY)
+        os.dup2(fdstdout, 1)
+        os.dup2(fdstderr, 2)
+        os.close(0)
+        self.fun()
+        #sys.exit(0)
+    
+    def parent(self):
+        self.fdstdout = open(PYTESTSTDOUT, "r")
+        self.fdstderr = open(PYTESTSTDERR, "r")
+        pid, exitstat = os.wait()
+        self.signal = exitstat & 0x7f
+        self.exitstat = exitstat & 0xff00
+        self.stdout = self.fdstdout.read()
+        self.stderr = self.fdstderr.read()
+        self.fdstdout.close()
+        self.fdstderr.close()
+    
+    def __del__(self):
+        try:
+            os.unlink(PYTESTSTDOUT)
+            os.unlink(PYTESTSTDERR)
+        except:
+            pass

Added: py/branch/distributed/py/test/rsession/testing/test_boxing.py
==============================================================================
--- (empty file)
+++ py/branch/distributed/py/test/rsession/testing/test_boxing.py	Sat Aug 26 14:53:09 2006
@@ -0,0 +1,22 @@
+
+""" test boxing functionality
+"""
+
+from py.__.test.rsession.box import Box
+from py.__.test.rsession.testing import example2
+
+def test_basic_boxing():
+    b = Box(example2.boxf1)
+    b.run()
+    assert b.stdout == "some out\n"
+    assert b.stderr == "some err\n"
+    assert b.exitstat == 0
+    assert b.signal == 0
+
+def test_boxing_on_fds():
+    b = Box(example2.boxf2)
+    b.run()
+    assert b.stdout == "someout"
+    assert b.stderr == "someerr"
+    assert b.exitstat == 0
+    assert b.signal == 0

Modified: py/branch/distributed/py/test/rsession/testing/test_report.py
==============================================================================
--- py/branch/distributed/py/test/rsession/testing/test_report.py	(original)
+++ py/branch/distributed/py/test/rsession/testing/test_report.py	Sat Aug 26 14:53:09 2006
@@ -34,10 +34,3 @@
         
         if method.startswith("report_") and method != "report_unknown":
             assert method[len('report_'):] in report.__dict__
-
-def f():
-    sad
-
-#def test_fail():
-#    f()
-