def execute(self, args, p, preargs=''):
if runtime.platformType == 'win32':
raise unittest.SkipTest, "can't run cmdline client on win32"
port = self.server.getHost().port
cmd = ('-p %i -l testuser '
'--known-hosts kh_test '
'--user-authentications publickey '
'--host-key-algorithms ssh-rsa '
'-a -I '
'-K direct '
'-i dsa_test '
'-v ') % port + preargs + \
' 127.0.0.1 ' + args
cmds = _makeArgs(cmd.split())
log.msg(str(cmds))
env = os.environ.copy()
env['PYTHONPATH'] = os.pathsep.join(sys.path)
reactor.spawnProcess(p, sys.executable, cmds, env=env)
return p.deferred
python类SkipTest()的实例源码
def testUnquote(self):
try:
from twisted.protocols import _c_urlarg
except ImportError:
raise unittest.SkipTest("_c_urlarg module is not available")
# work exactly like urllib.unquote, including stupid things
# % followed by a non-hexdigit in the middle and in the end
self.failUnlessEqual(urllib.unquote("%notreally%n"),
_c_urlarg.unquote("%notreally%n"))
# % followed by hexdigit, followed by non-hexdigit
self.failUnlessEqual(urllib.unquote("%1quite%1"),
_c_urlarg.unquote("%1quite%1"))
# unquoted text, followed by some quoted chars, ends in a trailing %
self.failUnlessEqual(urllib.unquote("blah%21%40%23blah%"),
_c_urlarg.unquote("blah%21%40%23blah%"))
# Empty string
self.failUnlessEqual(urllib.unquote(""), _c_urlarg.unquote(""))
def testUNIX(self):
# FIXME: This test is far too dense. It needs comments.
# -- spiv, 2004-11-07
if not interfaces.IReactorUNIX(reactor, None):
raise unittest.SkipTest, "This reactor does not support UNIX domain sockets"
s = service.MultiService()
s.startService()
factory = protocol.ServerFactory()
factory.protocol = TestEcho
TestEcho.d = defer.Deferred()
t = internet.UNIXServer('echo.skt', factory)
t.setServiceParent(s)
factory = protocol.ClientFactory()
factory.protocol = Foo
factory.d = defer.Deferred()
factory.line = None
internet.UNIXClient('echo.skt', factory).setServiceParent(s)
factory.d.addCallback(self.assertEqual, 'lalala')
factory.d.addCallback(lambda x : s.stopService())
factory.d.addCallback(lambda x : TestEcho.d)
factory.d.addCallback(self._cbTestUnix, factory, s)
return factory.d
def testVolatile(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()
self.failIfIdentical(t._port, None)
t1 = copy.copy(t)
self.assertIdentical(t1._port, None)
t.stopService()
self.assertIdentical(t._port, None)
self.failIf(t.running)
factory = protocol.ClientFactory()
factory.protocol = wire.Echo
t = internet.UNIXClient('echo.skt', factory)
t.startService()
self.failIfIdentical(t._connection, None)
t1 = copy.copy(t)
self.assertIdentical(t1._connection, None)
t.stopService()
self.assertIdentical(t._connection, None)
self.failIf(t.running)
def _testDetectFilesRemoved(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.close()
# Generate a cache with pluginextra in it.
list(plugin.getPlugins(plugin.ITestPlugin))
finally:
self._unimportPythonModule(
sys.modules['twisted.plugins.pluginextra'],
True)
plgs = list(plugin.getPlugins(plugin.ITestPlugin))
self.assertEquals(1, len(plgs))
def execute(self, args, p, preargs = ''):
cmdline = ('ssh -2 -l testuser -p %i '
'-oUserKnownHostsFile=kh_test '
'-oPasswordAuthentication=no '
# Always use the RSA key, since that's the one in kh_test.
'-oHostKeyAlgorithms=ssh-rsa '
'-a '
'-i dsa_test ') + preargs + \
' 127.0.0.1 ' + args
port = self.server.getHost().port
ssh_path = None
for path in ['/usr', '', '/usr/local']:
if os.path.exists(path+'/bin/ssh'):
ssh_path = path+'/bin/ssh'
break
if not ssh_path:
log.msg('skipping test, cannot find ssh')
raise unittest.SkipTest, 'skipping test, cannot find ssh'
cmds = (cmdline % port).split()
reactor.spawnProcess(p, ssh_path, cmds)
return p.deferred
def execute(self, args, p, preargs=''):
if runtime.platformType == 'win32':
raise unittest.SkipTest, "can't run cmdline client on win32"
port = self.server.getHost().port
cmd = ('-p %i -l testuser '
'--known-hosts kh_test '
'--user-authentications publickey '
'--host-key-algorithms ssh-rsa '
'-a -I '
'-K direct '
'-i dsa_test '
'-v ') % port + preargs + \
' 127.0.0.1 ' + args
cmds = _makeArgs(cmd.split())
log.msg(str(cmds))
env = os.environ.copy()
env['PYTHONPATH'] = os.pathsep.join(sys.path)
reactor.spawnProcess(p, sys.executable, cmds, env=env)
return p.deferred
def testUnquote(self):
try:
from twisted.protocols import _c_urlarg
except ImportError:
raise unittest.SkipTest("_c_urlarg module is not available")
# work exactly like urllib.unquote, including stupid things
# % followed by a non-hexdigit in the middle and in the end
self.failUnlessEqual(urllib.unquote("%notreally%n"),
_c_urlarg.unquote("%notreally%n"))
# % followed by hexdigit, followed by non-hexdigit
self.failUnlessEqual(urllib.unquote("%1quite%1"),
_c_urlarg.unquote("%1quite%1"))
# unquoted text, followed by some quoted chars, ends in a trailing %
self.failUnlessEqual(urllib.unquote("blah%21%40%23blah%"),
_c_urlarg.unquote("blah%21%40%23blah%"))
# Empty string
self.failUnlessEqual(urllib.unquote(""), _c_urlarg.unquote(""))
def testUNIX(self):
# FIXME: This test is far too dense. It needs comments.
# -- spiv, 2004-11-07
if not interfaces.IReactorUNIX(reactor, None):
raise unittest.SkipTest, "This reactor does not support UNIX domain sockets"
s = service.MultiService()
s.startService()
factory = protocol.ServerFactory()
factory.protocol = TestEcho
TestEcho.d = defer.Deferred()
t = internet.UNIXServer('echo.skt', factory)
t.setServiceParent(s)
factory = protocol.ClientFactory()
factory.protocol = Foo
factory.d = defer.Deferred()
factory.line = None
internet.UNIXClient('echo.skt', factory).setServiceParent(s)
factory.d.addCallback(self.assertEqual, 'lalala')
factory.d.addCallback(lambda x : s.stopService())
factory.d.addCallback(lambda x : TestEcho.d)
factory.d.addCallback(self._cbTestUnix, factory, s)
return factory.d
def testVolatile(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()
self.failIfIdentical(t._port, None)
t1 = copy.copy(t)
self.assertIdentical(t1._port, None)
t.stopService()
self.assertIdentical(t._port, None)
self.failIf(t.running)
factory = protocol.ClientFactory()
factory.protocol = wire.Echo
t = internet.UNIXClient('echo.skt', factory)
t.startService()
self.failIfIdentical(t._connection, None)
t1 = copy.copy(t)
self.assertIdentical(t1._connection, None)
t.stopService()
self.assertIdentical(t._connection, None)
self.failIf(t.running)
def _testDetectFilesRemoved(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.close()
# Generate a cache with pluginextra in it.
list(plugin.getPlugins(plugin.ITestPlugin))
finally:
self._unimportPythonModule(
sys.modules['twisted.plugins.pluginextra'],
True)
plgs = list(plugin.getPlugins(plugin.ITestPlugin))
self.assertEquals(1, len(plgs))
def test_drhost_parse(self):
"""
Test that the drhost exports for a project are correctly done.
"""
raise unittest.SkipTest("DRhosts refactoring not yet complete.")
# The list of drhosts for the first host should be 1 in length.
testhost01 = self.project.get_host_byname('testhost01')
self.failUnlessEqual( len(testhost01.get_drhosts()), 1 )
# Grab the test volume
testvol = [x for x in self.project.get_volumes() if x.name == 'testvol01' ][0]
# Check that the target volume qtree is being exported to the dr testhost
targethost = testvol.snapmirrors[0].targetvol.get_qtrees()[0].get_rw_exports()[0]
self.failUnlessEqual(targethost.name, 'dr_testhost01')
def mktime(t9):
"""
Call L{mktime_real}, and if it raises L{OverflowError}, catch it and raise
SkipTest instead.
@param t9: A time as a 9-item tuple.
@type t9: L{tuple}
@return: A timestamp.
@rtype: L{float}
"""
try:
return mktime_real(t9)
except OverflowError:
raise SkipTest(
"Platform cannot construct time zone for {0!r}"
.format(t9)
)
def doNotFailOnNetworkError(func):
"""
A decorator which makes APIBuilder tests not fail because of intermittent
network failures -- mamely, APIBuilder being unable to get the "object
inventory" of other projects.
@param func: The function to decorate.
@return: A decorated function which won't fail if the object inventory
fetching fails.
"""
@functools.wraps(func)
def wrapper(*a, **kw):
try:
func(*a, **kw)
except FailTest as e:
if e.args[0].startswith("'Failed to get object inventory from "):
raise SkipTest(
("This test is prone to intermittent network errors. "
"See ticket 8753. Exception was: {!r}").format(e))
raise
return wrapper
def scriptTest(self, name):
"""
Verify that the given script runs and uses the version of Twisted
currently being tested.
This only works when running tests against a vcs checkout of Twisted,
since it relies on the scripts being in the place they are kept in
version control, and exercises their logic for finding the right version
of Twisted to use in that situation.
@param name: A path fragment, relative to the I{bin} directory of a
Twisted source checkout, identifying a script to test.
@type name: C{str}
@raise SkipTest: if the script is not where it is expected to be.
"""
script = self.bin.preauthChild(name)
if not script.exists():
raise SkipTest(
"Script tests do not apply to installed configuration.")
from twisted.copyright import version
scriptVersion = outputFromPythonScript(script, '--version')
self.assertIn(str(version), scriptVersion)
def test_twistdPathInsert(self):
"""
The twistd script adds the current working directory to sys.path so
that it's able to import modules from it.
"""
script = self.bin.child("twistd")
if not script.exists():
raise SkipTest(
"Script tests do not apply to installed configuration.")
cwd = getcwd()
self.addCleanup(chdir, cwd)
testDir = FilePath(self.mktemp())
testDir.makedirs()
chdir(testDir.path)
testDir.child("bar.tac").setContent(
"import sys\n"
"print sys.path\n")
output = outputFromPythonScript(script, '-ny', 'bar.tac')
self.assertIn(repr(testDir.path), output)
def test_isRunningInit(self):
"""
L{PIDFile.isRunning} returns true for a process that we are not allowed
to kill (errno=EPERM).
@note: This differs from L{PIDFileTests.test_isRunningNotAllowed} in
that it actually invokes the C{kill} system call, which is useful for
testing of our chosen method for probing the existence of a process
that we are not allowed to kill.
@note: In this case, we try killing C{init}, which is process #1 on
POSIX systems, so this test is not portable. C{init} should always be
running and should not be killable by non-root users.
"""
if SYSTEM_NAME != "posix":
raise SkipTest("This test assumes POSIX")
pidFile = PIDFile(DummyFilePath())
pidFile._write(1) # PID 1 is init on POSIX systems
self.assertTrue(pidFile.isRunning())
def test_wsgiErrorsExpectsOnlyNativeStringsInPython2(self):
"""
The C{'wsgi.errors'} file-like object from the C{environ} C{dict}
expects writes of only native strings in Python 2. Some existing WSGI
applications may write non-native (i.e. C{unicode}) strings so, for
compatibility, these elicit only a warning in Python 2.
"""
if _PY3:
raise SkipTest("Not relevant in Python 3")
request, result = self.prepareRequest()
request.requestReceived()
environ, _ = self.successResultOf(result)
errors = environ["wsgi.errors"]
with warnings.catch_warnings(record=True) as caught:
errors.write(u"fred")
self.assertEqual(1, len(caught))
self.assertEqual(UnicodeWarning, caught[0].category)
self.assertEqual(
"write() argument should be str, not u'fred' (unicode)",
str(caught[0].message))
def test_wsgiErrorsAcceptsOnlyNativeStringsInPython3(self):
"""
The C{'wsgi.errors'} file-like object from the C{environ} C{dict}
permits writes of only native strings in Python 3, and raises
C{TypeError} for writes of non-native strings.
"""
if not _PY3:
raise SkipTest("Relevant only in Python 3")
request, result = self.prepareRequest()
request.requestReceived()
environ, _ = self.successResultOf(result)
errors = environ["wsgi.errors"]
error = self.assertRaises(TypeError, errors.write, b"fred")
self.assertEqual(
"write() argument must be str, not b'fred' (bytes)",
str(error))
def test_UDP(self):
"""
Test L{internet.UDPServer} with a random port: starting the service
should give it valid port, and stopService should free it so that we
can start a server on the same port again.
"""
if not interfaces.IReactorUDP(reactor, None):
raise unittest.SkipTest("This reactor does not support UDP sockets")
p = protocol.DatagramProtocol()
t = internet.UDPServer(0, p)
t.startService()
num = t._port.getHost().port
self.assertNotEqual(num, 0)
def onStop(ignored):
t = internet.UDPServer(num, p)
t.startService()
return t.stopService()
return defer.maybeDeferred(t.stopService).addCallback(onStop)
def test_ecSuccessWithRealBindings(self):
"""
Integration test that checks the positive code path to ensure that we
use the API properly.
"""
try:
defaultCurve = sslverify._OpenSSLECCurve(
sslverify._defaultCurveName
)
except NotImplementedError:
raise unittest.SkipTest(
"Underlying pyOpenSSL is not based on cryptography."
)
opts = sslverify.OpenSSLCertificateOptions(
privateKey=self.sKey,
certificate=self.sCert,
)
self.assertEqual(defaultCurve, opts._ecCurve)
# Exercise positive code path. getContext swallows errors so we do it
# explicitly by hand.
opts._ecCurve.addECKeyToContext(opts.getContext())
def test_formatTimeDefault(self):
"""
Time is first field. Default time stamp format is RFC 3339 and offset
respects the timezone as set by the standard C{TZ} environment variable
and L{tzset} API.
"""
if tzset is None:
raise SkipTest(
"Platform cannot change timezone; unable to verify offsets."
)
addTZCleanup(self)
setTZ("UTC+00")
t = mktime((2013, 9, 24, 11, 40, 47, 1, 267, 1))
event = dict(log_format=u"XYZZY", log_time=t)
self.assertEqual(
formatEventAsClassicLogText(event),
u"2013-09-24T11:40:47+0000 [-#-] XYZZY\n",
)
def emccoprhdblockdeviceapi_for_test(test_case):
"""
Create a ``CoprHDBlockDeviceAPI`` instance for use in tests.
:returns: A ``CoprHDBlockDeviceAPI`` instance
"""
user_id = os.getuid()
if user_id != 0:
raise SkipTest(
"``CoprHDBlockDeviceAPI`` queries for iSCSI initiator name which is owned by root, "
"Required UID: 0, Found UID: {!r}".format(user_id)
)
coprhd = tidy_coprhd_client_for_test(test_case)
return coprhd
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 execute(self, args, p, preargs = ''):
cmdline = ('ssh -2 -l testuser -p %i '
'-oUserKnownHostsFile=kh_test '
'-oPasswordAuthentication=no '
# Always use the RSA key, since that's the one in kh_test.
'-oHostKeyAlgorithms=ssh-rsa '
'-a '
'-i dsa_test ') + preargs + \
' 127.0.0.1 ' + args
port = self.server.getHost().port
ssh_path = None
for path in ['/usr', '', '/usr/local']:
if os.path.exists(path+'/bin/ssh'):
ssh_path = path+'/bin/ssh'
break
if not ssh_path:
log.msg('skipping test, cannot find ssh')
raise unittest.SkipTest, 'skipping test, cannot find ssh'
cmds = (cmdline % port).split()
reactor.spawnProcess(p, ssh_path, cmds)
return p.deferred
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
