Initial commit.

test/.cvsignore

+*.pyc
+*.pyo

test/test_popen.py

+#!/usr/bin/env python
+import unittest
+
+import time
+import signal
+
+from pcl_expect import Controller
+from pcl_expect.popen import Popen
+
+class Timeout(Exception): pass
+
+def handler(signo, stk):
+    raise Timeout()
+
+class TestPopen(unittest.TestCase):
+    def setUp(self):
+        signal.signal(signal.SIGALRM, handler)
+        signal.alarm(20)
+
+    def tearDown(self):
+        signal.alarm(0)
+
+    def test_popen_sleep_1(self):
+        t0 = time.time()
+        sleeper = Popen('sleep 1')
+        x = Controller()
+        while x.loop():
+            if sleeper.re(x, '.'):
+                self.fail('got output from sleep')
+            elif x.timeout():
+                self.fail('timeout waiting for sleep to terminate')
+            elif sleeper.eof(x):
+                break
+        status = sleeper.close()
+        self.assertEqual(status, None)
+        t1 = time.time()
+        self.assertAlmostEqual(t0 + 1.0, t1, 1,
+                               'sleep 1 slept too long or too short')
+    def test_popen_sleep_1_compat(self):
+        t0 = time.time()
+        sleeper = Popen('sleep 1')
+        x = Controller()
+        while x.loop():
+            if x.re(sleeper, '.'):
+                self.fail('got output from sleep')
+            elif x.timeout():
+                self.fail('timeout waiting for sleep to terminate')
+            elif x.eof(sleeper):
+                break
+        status = sleeper.close()
+        self.assertEqual(status, None)
+        t1 = time.time()
+        self.assertAlmostEqual(t0 + 1.0, t1, 1,
+                               'sleep 1 slept too long or too short')
+
+    def test_read_hello(self):
+        hello = Popen('echo hello, world')
+        x = Controller()
+        while x.loop():
+            if hello.re(x, 'hello, world\n'):
+                break
+        x = Controller()
+        while x.loop():
+            if hello.eof(x):
+                break
+        status = hello.close()
+        self.assertEqual(status, None)
+
+    def test_read_hello_partial(self):
+        hello = Popen('echo hello, world')
+        x = Controller()
+        while x.loop():
+            if hello.re(x, 'hello'):
+                break
+        x = Controller()
+        while x.loop():
+            if hello.re(x, ', world'):
+                break
+        x = Controller()
+        while x.loop():
+            if hello.re(x, '\n'):
+                break
+        x = Controller()
+        while x.loop():
+            if hello.eof(x):
+                break
+        status = hello.close()
+        self.assertEqual(status, None)
+
+    def test_read_hello_compat(self):
+        hello = Popen('echo hello, world')
+        x = Controller()
+        while x.loop():
+            if x.re(hello, 'hello, world\n'):
+                break
+        x = Controller()
+        while x.loop():
+            if x.eof(hello):
+                break
+        status = hello.close()
+        self.assertEqual(status, None)
+
+if __name__ == '__main__':
+    unittest.main()

test/test_spawn.py

+#!/usr/bin/env python
+import unittest
+
+import time
+
+import pcl_expect
+import pcl_expect.spawn
+
+class TestSpawn(unittest.TestCase):
+    def test_spawn_sleep_1(self):
+        t0 = time.time()
+        sleeper = pcl_expect.spawn.Spawn(['sleep', '1'])
+        x = pcl_expect.Controller()
+        while x.loop():
+            if sleeper.re(x, '.'):
+                self.fail('got output from sleep')
+            elif x.timeout():
+                self.fail('timeout waiting for sleep to terminate')
+            elif sleeper.eof(x):
+                break
+        pid, status = sleeper.close()
+        self.assertEqual(status, 0)
+        t1 = time.time()
+        self.assertAlmostEqual(t0 + 1.0, t1, 1,
+                               'sleep 1 slept too long or too short')
+
+    def test_spawn_sleep_1_compat(self):
+        t0 = time.time()
+        sleeper = pcl_expect.spawn.Spawn(['sleep', '1'])
+        x = pcl_expect.Controller()
+        while x.loop():
+            if x.re(sleeper, '.'):
+                self.fail('got output from sleep')
+            elif x.timeout():
+                self.fail('timeout waiting for sleep to terminate')
+            elif x.eof(sleeper):
+                break
+        pid, status = sleeper.close()
+        self.assertEqual(status, 0)
+        t1 = time.time()
+        self.assertAlmostEqual(t0 + 1.0, t1, 1,
+                               'sleep 1 slept too long or too short')
+
+
+if __name__ == '__main__':
+    unittest.main()