我们从Python开源项目中，提取了以下48个代码示例，用于说明如何使用twisted.trial.unittest.SkipTest()。
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 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 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 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
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
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 setUpClass(self): raise unittest.SkipTest("class")
def _cb1(self, ignored): raise unittest.SkipTest("skip me")
def _cb_skip(self, reason): raise unittest.SkipTest(reason)
