You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
mitogen/tests/econtext_test.py

128 lines
3.0 KiB
Python

#!/usr/bin/env python2.5
"""
def DoStuff():
import time
file('/tmp/foobar', 'w').write(time.ctime())
localhost = pyrpc.SSHConnection('localhost')
localhost.Connect()
try:
ret = localhost.Evaluate(DoStuff)
except OSError, e:
Tests
- Test Channel objects to destruction.
- External contexts sometimes don't appear to die during a crash. This needs
tested to destruction.
- Test reconnecting to previously idle-killed contexts.
- Test remote context longevity to destruction. They should never stay
around after parent dies or disconnects.
"""
import sys
import unittest
import econtext
#
# Helper functions.
#
class GetModuleImportsTestCase(unittest.TestCase):
# This must be kept in sync with our actual imports.
IMPORTS = [
('econtext', 'econtext'),
('sys', 'PythonSystemModule'),
('sys', 'sys'),
('unittest', 'unittest')
]
def setUp(self):
global PythonSystemModule
import sys as PythonSystemModule
def tearDown(Self):
global PythonSystemModule
del PythonSystemModule
def testImports(self):
self.assertEqual(set(self.IMPORTS),
set(econtext.GetModuleImports(sys.modules[__name__])))
class BuildPartialModuleTestCase(unittest.TestCase):
def testNullModule(self):
"""Pass empty sequences; result should contain nothing but a hash bang line
and whitespace."""
lines = econtext.BuildPartialModule([], []).strip().split('\n')
self.assert_(lines[0].startswith('#!'))
self.assert_('import' not in lines[1:])
def testPassingMethodTypeFails(self):
"""Pass an instance method and ensure we refuse it."""
self.assertRaises(TypeError, econtext.BuildPartialModule,
[self.testPassingMethodTypeFails], [])
@staticmethod
def exampleStaticMethod():
pass
def testStaticMethodGetsUnwrapped(self):
"Ensure that @staticmethod decorators are stripped."
dct = {}
exec econtext.BuildPartialModule([self.exampleStaticMethod], []) in dct
self.assertFalse(isinstance(dct['exampleStaticMethod'], staticmethod))
#
# Streams
#
class StreamTestBase:
"""This defines rules that should remain true for all Stream subclasses. We
test in this manner to guard against a subclass breaking Stream's
polymorphism (e.g. overriding a method with the wrong prototype).
def testCommandLine(self):
print self.driver.command_line
"""
class SSHStreamTestCase(unittest.TestCase, StreamTestBase):
DRIVER_CLASS = econtext.SSHStream
def setUp(self):
# Stubs.
# Instance initialization.
self.stream = econtext.SSHStream('localhost', 'test-agent')
def tearDown(self):
pass
def testConstructor(self):
pass
class TCPStreamTestCase(unittest.TestCase, StreamTestBase):
pass
#
# Run the tests.
#
if __name__ == '__main__':
unittest.main()