summaryrefslogtreecommitdiff
path: root/tests/processes_test.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/processes_test.py')
-rw-r--r--tests/processes_test.py110
1 files changed, 0 insertions, 110 deletions
diff --git a/tests/processes_test.py b/tests/processes_test.py
deleted file mode 100644
index 889c824..0000000
--- a/tests/processes_test.py
+++ /dev/null
@@ -1,110 +0,0 @@
-import sys
-import warnings
-from tests import LimitedTestCase, main, skip_on_windows
-
-warnings.simplefilter('ignore', DeprecationWarning)
-from eventlet import processes, api
-warnings.simplefilter('default', DeprecationWarning)
-
-
-class TestEchoPool(LimitedTestCase):
- def setUp(self):
- super(TestEchoPool, self).setUp()
- self.pool = processes.ProcessPool('echo', ["hello"])
-
- @skip_on_windows
- def test_echo(self):
- result = None
-
- proc = self.pool.get()
- try:
- result = proc.read()
- finally:
- self.pool.put(proc)
- self.assertEqual(result, 'hello\n')
-
- @skip_on_windows
- def test_read_eof(self):
- proc = self.pool.get()
- try:
- proc.read()
- self.assertRaises(processes.DeadProcess, proc.read)
- finally:
- self.pool.put(proc)
-
- @skip_on_windows
- def test_empty_echo(self):
- p = processes.Process('echo', ['-n'])
- self.assertEqual('', p.read())
- self.assertRaises(processes.DeadProcess, p.read)
-
-
-class TestCatPool(LimitedTestCase):
- def setUp(self):
- super(TestCatPool, self).setUp()
- api.sleep(0)
- self.pool = processes.ProcessPool('cat')
-
- @skip_on_windows
- def test_cat(self):
- result = None
-
- proc = self.pool.get()
- try:
- proc.write('goodbye')
- proc.close_stdin()
- result = proc.read()
- finally:
- self.pool.put(proc)
-
- self.assertEqual(result, 'goodbye')
-
- @skip_on_windows
- def test_write_to_dead(self):
- result = None
-
- proc = self.pool.get()
- try:
- proc.write('goodbye')
- proc.close_stdin()
- result = proc.read()
- self.assertRaises(processes.DeadProcess, proc.write, 'foo')
- finally:
- self.pool.put(proc)
-
- @skip_on_windows
- def test_close(self):
- result = None
-
- proc = self.pool.get()
- try:
- proc.write('hello')
- proc.close()
- self.assertRaises(processes.DeadProcess, proc.write, 'goodbye')
- finally:
- self.pool.put(proc)
-
-
-class TestDyingProcessesLeavePool(LimitedTestCase):
- def setUp(self):
- super(TestDyingProcessesLeavePool, self).setUp()
- self.pool = processes.ProcessPool('echo', ['hello'], max_size=1)
-
- @skip_on_windows
- def test_dead_process_not_inserted_into_pool(self):
- proc = self.pool.get()
- try:
- try:
- result = proc.read()
- self.assertEqual(result, 'hello\n')
- result = proc.read()
- except processes.DeadProcess:
- pass
- finally:
- self.pool.put(proc)
- proc2 = self.pool.get()
- assert proc is not proc2
-
-
-if __name__ == '__main__':
- main()