def testStoppingServer(self):
if not interfaces.IReactorUNIX(reactor, None):
raise unittest.SkipTest, "This reactor does not support UNIX domain sockets"
factory = protocol.ServerFactory()
factory.protocol = wire.Echo
t = internet.UNIXServer('echo.skt', factory)
t.startService()
t.stopService()
self.failIf(t.running)
factory = protocol.ClientFactory()
d = defer.Deferred()
factory.clientConnectionFailed = lambda *args: d.callback(None)
reactor.connectUNIX('echo.skt', factory)
return d
python类SkipTest()的实例源码
def testEncryptedStyles(self):
try:
import Crypto
except ImportError:
raise unittest.SkipTest()
for o in objects:
phrase='once I was the king of spain'
p = sob.Persistent(o, '')
for style in 'xml source pickle'.split():
if style == 'xml' and not gotMicrodom:
continue
p.setStyle(style)
p.save(filename='epersisttest.'+style, passphrase=phrase)
o1 = sob.load('epersisttest.'+style, style, phrase)
self.failUnlessEqual(o, o1)
def testEncryptedPython(self):
try:
import Crypto
except ImportError:
raise unittest.SkipTest()
phrase='once I was the king of spain'
open("epersisttest.python", 'w').write(
sob._encrypt(phrase, 'foo=[1,2,3]'))
o = sob.loadValueFromFile('epersisttest.python', 'foo', phrase)
self.failUnlessEqual(o, [1,2,3])
def _testDetectNewFiles(self):
writeFileName = sibpath(plugins.__file__, 'pluginextra.py')
try:
wf = file(writeFileName, 'w')
except IOError, ioe:
if ioe.errno == errno.EACCES:
raise unittest.SkipTest(
"No permission to add things to twisted.plugins")
else:
raise
else:
try:
wf.write(begintest)
wf.flush()
self.failIfIn('twisted.plugins.pluginextra', sys.modules)
self.failIf(hasattr(sys.modules['twisted.plugins'], 'pluginextra'),
"plugins package still has pluginextra module")
plgs = list(plugin.getPlugins(plugin.ITestPlugin))
self.assertEquals(
len(plgs), 2,
"Unexpected plugins found: %r" % (
[p.__name__ for p in plgs]))
names = ['TestPlugin', 'FourthTestPlugin']
for p in plgs:
names.remove(p.__name__)
p.test1()
finally:
wf.close()
self._unimportPythonModule(
sys.modules['twisted.plugins.pluginextra'],
True)
def _testDetectFilesChanged(self):
writeFileName = sibpath(plugins.__file__, 'pluginextra.py')
try:
writeFile = file(writeFileName, 'w')
except IOError, ioe:
if ioe.errno == errno.EACCES:
raise unittest.SkipTest(
"No permission to add things to twisted.plugins")
else:
raise
try:
writeFile.write(begintest)
writeFile.flush()
plgs = list(plugin.getPlugins(plugin.ITestPlugin))
# Sanity check
self.assertEquals(
len(plgs), 2,
"Unexpected plugins found: %r" % (
[p.__name__ for p in plgs]))
writeFile.write(extratest)
writeFile.flush()
# Fake out Python.
self._unimportPythonModule(sys.modules['twisted.plugins.pluginextra'])
# Make sure additions are noticed
plgs = list(plugin.getPlugins(plugin.ITestPlugin))
self.assertEquals(len(plgs), 3, "Unexpected plugins found: %r" % (
[p.__name__ for p in plgs]))
names = ['TestPlugin', 'FourthTestPlugin', 'FifthTestPlugin']
for p in plgs:
names.remove(p.__name__)
p.test1()
finally:
writeFile.close()
self._unimportPythonModule(
sys.modules['twisted.plugins.pluginextra'],
True)
def setUpClass(self):
raise unittest.SkipTest, "thi stest is fukct"
def test_skip1(self):
raise unittest.SkipTest('skip1')
def test_skip1(self):
raise unittest.SkipTest('skip1')
def setUpClass(self):
raise unittest.SkipTest("class")
def test_skip1(self):
raise unittest.SkipTest('skip1')
def _cb1(self, ignored):
raise unittest.SkipTest("skip me")
def _cb_skip(self, reason):
raise unittest.SkipTest(reason)
def setUpClass(self):
try:
import pyexpat
except ImportError:
raise unittest.SkipTest, "Skipping ExpatElementStream test, since no expat wrapper is available."
self.streamClass = domish.ExpatElementStream
def setUpClass(self):
if domish.SuxElementStream is None:
raise unittest.SkipTest, "Skipping SuxElementStream test, since twisted.web is not available."
self.streamClass = domish.SuxElementStream
def setUp(self):
if not ssh:
raise unittest.SkipTest("Crypto requirements missing, can't run historic recvline tests over ssh")
u, p = 'testuser', 'testpass'
rlm = TerminalRealm()
rlm.userFactory = TestUser
rlm.chainedProtocolFactory = lambda: insultsServer
ptl = portal.Portal(
rlm,
[checkers.InMemoryUsernamePasswordDatabaseDontUse(**{u: p})])
sshFactory = ConchFactory(ptl)
sshFactory.serverProtocol = self.serverProtocol
sshFactory.startFactory()
recvlineServer = self.serverProtocol()
insultsServer = insults.ServerProtocol(lambda: recvlineServer)
sshServer = sshFactory.buildProtocol(None)
clientTransport = LoopbackRelay(sshServer)
recvlineClient = NotifyingExpectableBuffer()
insultsClient = insults.ClientProtocol(lambda: recvlineClient)
sshClient = TestTransport(lambda: insultsClient, (), {}, u, p, self.WIDTH, self.HEIGHT)
serverTransport = LoopbackRelay(sshClient)
sshClient.makeConnection(clientTransport)
sshServer.makeConnection(serverTransport)
self.recvlineClient = recvlineClient
self.sshClient = sshClient
self.sshServer = sshServer
self.clientTransport = clientTransport
self.serverTransport = serverTransport
return recvlineClient.onConnection
def testCheckGoogle(self):
raise unittest.SkipTest("no violation of google ToS")
d = google.checkGoogle('site:www.twistedmatrix.com twisted')
d.addCallback(self.assertEquals, 'http://twistedmatrix.com/')
return d
def setUp(self):
if self.good_sql is None:
raise unittest.SkipTest('no good sql for reconnect test')
self.startDB()
self.dbpool = self.makePool(cp_max=1, cp_reconnect=True,
cp_good_sql=self.good_sql)
self.dbpool.start()
return self.dbpool.runOperation(simple_table_schema)
def setUpClass(self):
if self.needs_dbdir:
self.DB_DIR = tempfile.mktemp()
os.mkdir(self.DB_DIR)
if not self.can_connect():
raise unittest.SkipTest, '%s: Cannot access db' % self.TEST_PREFIX