[Zope3-checkins] SVN: Zope3/branches/mkerrin-remove_trial_tests/
Integrate the trial ftp tests with the Zope testrunner and
Michael Kerrin
michael.kerrin at openapp.biz
Tue Feb 21 09:46:50 EST 2006
Log message for revision 41729:
Integrate the trial ftp tests with the Zope testrunner and
remove the trial tests from Zope
Changed:
D Zope3/branches/mkerrin-remove_trial_tests/src/zope/app/twisted/ftp/test/
A Zope3/branches/mkerrin-remove_trial_tests/src/zope/app/twisted/ftp/tests/test_zope_ftp.py
A Zope3/branches/mkerrin-remove_trial_tests/src/zope/app/twisted/ftp/tests/test_zopetrial.py
U Zope3/branches/mkerrin-remove_trial_tests/test.py
-=-
Copied: Zope3/branches/mkerrin-remove_trial_tests/src/zope/app/twisted/ftp/tests/test_zope_ftp.py (from rev 41495, Zope3/branches/mkerrin-remove_trial_tests/src/zope/app/twisted/ftp/test/test_zope_ftp.py)
===================================================================
--- Zope3/branches/mkerrin-remove_trial_tests/src/zope/app/twisted/ftp/test/test_zope_ftp.py 2006-01-30 14:54:33 UTC (rev 41495)
+++ Zope3/branches/mkerrin-remove_trial_tests/src/zope/app/twisted/ftp/tests/test_zope_ftp.py 2006-02-21 14:46:49 UTC (rev 41729)
@@ -0,0 +1,336 @@
+##############################################################################
+#
+# Copyright (c) 2005 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""This file basically contains FTP functional tests.
+
+$Id:$
+"""
+__docformat__="restructuredtext"
+from cStringIO import StringIO
+import posixpath
+import unittest
+
+from twisted.test import test_ftp
+from twisted.internet import reactor, protocol, defer
+from twisted.protocols import ftp
+
+from zope.app.twisted.ftp.server import FTPRealm, FTPFactory
+from zope.app.twisted.ftp.tests.test_publisher import RequestFactory
+from zope.app.twisted.ftp.tests import demofs
+
+from twisted.trial.util import wait
+from twisted.trial.unittest import TestCase
+
+import test_zopetrial
+
+class DemoFileSystem(demofs.DemoFileSystem):
+ def mkdir_nocheck(self, path):
+ path, name = posixpath.split(path)
+ d = self.getdir(path)
+ if name in d.files:
+ raise OSError("Already exists:", name)
+ newdir = self.Directory()
+ newdir.grant(self.user, demofs.read | demofs.write)
+ d.files[name] = newdir
+
+ def writefile_nocheck(self, path, instream, start = None,
+ end = None, append = False):
+ path, name = posixpath.split(path)
+ d = self.getdir(path)
+ f = d.files.get(name)
+ if f is None:
+ d = self.getdir(path)
+ f = d.files[name] = self.File()
+ f.grant(self.user, demofs.read | demofs.write)
+ elif f.type != 'f':
+ raise OSError("Can't overwrite a directory")
+
+ if append:
+ f.data += instream.read()
+ else:
+ f.data = instream.read()
+
+class FTPServerTestCase(test_ftp.FTPServerTestCase):
+ def tearDown(self):
+ test_zopetrial.tearDown()
+
+ # Clean up sockets
+ self.client.transport.loseConnection()
+ d = self.port.stopListening()
+ if d is not None:
+ wait(d)
+
+ del self.serverProtocol
+
+ def setUp(self):
+ test_zopetrial.setUp()
+
+ root = demofs.Directory()
+ # the tuple has a user name is used by ZopeSimpleAuthentication to
+ # authenticate users.
+ root.grant('root', demofs.write)
+ self.rootfs = rootfs = DemoFileSystem(root, ('root', 'root'))
+
+ # Start the server
+ self.factory = FTPFactory(request_factory = RequestFactory(rootfs))
+ self.port = reactor.listenTCP(0, self.factory, interface="127.0.0.1")
+
+ # Hook the server's buildProtocol to make the protocol instance
+ # accessible to tests.
+ buildProtocol = self.factory.buildProtocol
+ def _rememberProtocolInstance(addr):
+ protocol = buildProtocol(addr)
+ self.serverProtocol = protocol.wrappedProtocol
+ return protocol
+ self.factory.buildProtocol = _rememberProtocolInstance
+
+ # Connect a client to it
+ portNum = self.port.getHost().port
+ clientCreator = protocol.ClientCreator(reactor, ftp.FTPClientBasic)
+ self.client = wait(clientCreator.connectTCP("127.0.0.1", portNum))
+
+ def _anonymousLogin(self):
+ responseLines = wait(self.client.queueStringCommand('USER anonymous'))
+ self.assertEquals(
+ ['331 Password required for anonymous.'],
+ responseLines
+ )
+
+ responseLines = wait(self.client.queueStringCommand(
+ 'PASS test at twistedmatrix.com')
+ )
+ self.assertEquals(
+ ['230 User logged in, proceed'],
+ responseLines
+ )
+
+class BasicFTPServerTestCase(FTPServerTestCase,
+ test_ftp.BasicFTPServerTestCase):
+ def _authLogin(self):
+ responseLines = wait(self.client.queueStringCommand('USER root'))
+ self.assertEquals(
+ ['331 Password required for root.'],
+ responseLines
+ )
+
+ responseLines = wait(self.client.queueStringCommand(
+ 'PASS root')
+ )
+ self.assertEquals(
+ ['230 User logged in, proceed'],
+ responseLines
+ )
+
+ def testQuit(self):
+ # this test is causing we problems. Works on 2.2.X but times out
+ # on 2.1.X
+ pass
+
+ def test_MKD(self):
+ self._authLogin()
+ responseLines = wait(self.client.queueStringCommand('MKD /newdir'))
+ self.assertEqual(['257 "/newdir" created'], responseLines)
+
+ def test_RMD(self):
+ self.rootfs.mkdir_nocheck('/newdir')
+
+ self._authLogin()
+ responseLines = wait(self.client.queueStringCommand('RMD /newdir'))
+ self.assertEqual(
+ ['250 Requested File Action Completed OK'], responseLines)
+
+ def test_DELE(self):
+ self.rootfs.writefile_nocheck('/file.txt', StringIO('x' * 20))
+
+ self._authLogin()
+ responseLines = wait(self.client.queueStringCommand('DELE /file.txt'))
+ self.assertEqual(
+ ['250 Requested File Action Completed OK'], responseLines)
+
+ def test_SIZE(self):
+ self.rootfs.writefile_nocheck('/file.txt', StringIO('x' * 20))
+
+ self._anonymousLogin()
+ responseLines = wait(self.client.queueStringCommand('SIZE /file.txt'))
+ self.assertEqual(['213 20'], responseLines)
+
+ def test_SIZE_on_dir(self):
+ self._anonymousLogin()
+ responseLines = wait(self.client.queueStringCommand('SIZE /'))
+ self.assertEqual(['213 0'] , responseLines)
+
+
+class FTPServerPasvDataConnectionTestCase(FTPServerTestCase,
+ test_ftp.FTPServerPasvDataConnectionTestCase):
+
+ def testLIST(self):
+ # Login
+ self._anonymousLogin()
+
+ # Download a listing
+ downloader = self._makeDataConnection()
+ d = self.client.queueStringCommand('LIST')
+ wait(defer.gatherResults([d, downloader.d]))
+
+ # No files, so the file listing should be empty
+ self.assertEqual('', downloader.buffer)
+
+ # Make some directories
+ self.rootfs.mkdir_nocheck('/foo')
+ self.rootfs.mkdir_nocheck('/bar')
+
+ # Download a listing again
+ downloader = self._makeDataConnection()
+ d = self.client.queueStringCommand('LIST')
+ wait(defer.gatherResults([d, downloader.d]))
+
+ # Now we expect 2 lines because there are two files.
+ self.assertEqual(2, len(downloader.buffer[:-2].split('\r\n')))
+
+ # Download a names-only listing
+ downloader = self._makeDataConnection()
+ d = self.client.queueStringCommand('NLST ')
+ wait(defer.gatherResults([d, downloader.d]))
+ filenames = downloader.buffer[:-2].split('\r\n')
+ filenames.sort()
+ self.assertEqual(['bar', 'foo'], filenames)
+
+ # Download a listing of the 'foo' subdirectory
+ downloader = self._makeDataConnection()
+ d = self.client.queueStringCommand('LIST foo')
+ wait(defer.gatherResults([d, downloader.d]))
+
+ # 'foo' has no files, so the file listing should be empty
+ self.assertEqual('', downloader.buffer)
+
+ # Change the current working directory to 'foo'
+ wait(self.client.queueStringCommand('CWD foo'))
+
+ # Download a listing from within 'foo', and again it should be empty
+ downloader = self._makeDataConnection()
+ d = self.client.queueStringCommand('LIST')
+ wait(defer.gatherResults([d, downloader.d]))
+ self.assertEqual('', downloader.buffer)
+
+ def testManyLargeDownloads(self):
+ # Login
+ self._anonymousLogin()
+
+
+ # Download a range of different size files
+ for size in range(100000, 110000, 500):
+ self.rootfs.writefile_nocheck('/%d.txt' % (size,),
+ StringIO('x' * size))
+
+ downloader = self._makeDataConnection()
+ d = self.client.queueStringCommand('RETR %d.txt' % (size,))
+ wait(defer.gatherResults([d, downloader.d]))
+ self.assertEqual('x' * size, downloader.buffer)
+
+
+class FTPServerPortDataConnectionTestCaes(FTPServerPasvDataConnectionTestCase,
+ test_ftp.FTPServerPortDataConnectionTestCase):
+ def setUp(self):
+ FTPServerPasvDataConnectionTestCase.setUp(self)
+ self.dataPorts = []
+
+ def tearDown(self):
+ l = [defer.maybeDeferred(port.stopListening) for port in self.dataPorts]
+ wait(defer.DeferredList(l, fireOnOneErrback=True))
+ return FTPServerPasvDataConnectionTestCase.tearDown(self)
+
+from twisted.test.test_ftp import _BufferingProtocol
+
+class ZopeFTPPermissionTestCases(FTPServerTestCase):
+ def setUp(self):
+ FTPServerTestCase.setUp(self)
+ self.filename = 'nopermissionfolder'
+ self.rootfs.writefile('/%s' % self.filename, StringIO('x' * 100))
+ file = self.rootfs.get(self.filename)
+ file.grant('michael', 0)
+ del file.access['anonymous']
+
+ def _makeDataConnection(self):
+ # Establish a passive data connection (i.e. client connecting to
+ # server).
+ responseLines = wait(self.client.queueStringCommand('PASV'))
+ host, port = ftp.decodeHostPort(responseLines[-1][4:])
+ downloader = wait(
+ protocol.ClientCreator(reactor,
+ _BufferingProtocol).connectTCP('127.0.0.1',
+ port)
+ )
+ return downloader
+
+ def _michaelLogin(self):
+ responseLines = wait(self.client.queueStringCommand('USER michael'))
+ self.assertEquals(
+ ['331 Password required for michael.'],
+ responseLines
+ )
+
+ responseLines = wait(self.client.queueStringCommand(
+ 'PASS michael')
+ )
+ self.assertEquals(
+ ['230 User logged in, proceed'],
+ responseLines
+ )
+
+ def testNoSuchDirectory(self):
+ self._michaelLogin()
+ deferred = self.client.queueStringCommand('CWD /nosuchdir')
+ failureResponseLines = self._waitForCommandFailure(deferred)
+ self.failUnless(failureResponseLines[-1].startswith('550'),
+ "Response didn't start with 550: %r" %
+ failureResponseLines[-1])
+
+ def testListNonPermission(self):
+ self._michaelLogin()
+
+ # Download a listing
+ downloader = self._makeDataConnection()
+ d = self.client.queueStringCommand('NLST ')
+ wait(defer.gatherResults([d, downloader.d]))
+
+ # No files, so the file listing should be empty
+ filenames = downloader.buffer[:-2].split('\r\n')
+ filenames.sort()
+ self.assertEqual([self.filename], filenames)
+
+ def testRETR_wo_Permission(self):
+ self._michaelLogin()
+
+ downloader = self._makeDataConnection()
+ d = self.client.queueStringCommand('RETR %s' % self.filename)
+ failureResponseLines = self._waitForCommandFailure(d)
+ self.failUnless(failureResponseLines[-1].startswith('550'),
+ "Response didn't start with 550: %r" %
+ failureResponseLines[-1])
+ if downloader.transport.connected:
+ downloader.transport.loseConnection()
+
+
+def test_suite():
+ suite = unittest.TestSuite()
+
+ suite.addTest(unittest.makeSuite(FTPServerTestCase))
+ suite.addTest(unittest.makeSuite(BasicFTPServerTestCase))
+ suite.addTest(unittest.makeSuite(FTPServerPasvDataConnectionTestCase))
+ suite.addTest(unittest.makeSuite(FTPServerPortDataConnectionTestCaes))
+ suite.addTest(unittest.makeSuite(ZopeFTPPermissionTestCases))
+
+ return suite
+
+if __name__ == '__main__':
+ test_suite()
Property changes on: Zope3/branches/mkerrin-remove_trial_tests/src/zope/app/twisted/ftp/tests/test_zope_ftp.py
___________________________________________________________________
Name: svn:keywords
+ Id
Name: svn:eol-style
+ native
Added: Zope3/branches/mkerrin-remove_trial_tests/src/zope/app/twisted/ftp/tests/test_zopetrial.py
===================================================================
--- Zope3/branches/mkerrin-remove_trial_tests/src/zope/app/twisted/ftp/tests/test_zopetrial.py 2006-02-21 14:39:58 UTC (rev 41728)
+++ Zope3/branches/mkerrin-remove_trial_tests/src/zope/app/twisted/ftp/tests/test_zopetrial.py 2006-02-21 14:46:49 UTC (rev 41729)
@@ -0,0 +1,228 @@
+##############################################################################
+#
+# Copyright (c) 2006 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""Contains helper functions and monkey patches to integrate twisted trial
+tests with the zope testrunner.
+
+WARNING - use with care - actaully just DO NOT USE THIS CODE. I only wrote
+his to test the ftp server and the only the ftp server. It should NOT be used
+for anything else. If we need to integrate the twisted trial framework and the
+zope.testing testrunner for anything else we need to think about this and come
+up with something better. Seriously.
+
+Twisted wraps all failures inside twisted.python.failure.Failure objects, and
+it also catches every exception thrown so I need this code to make my tests
+fail when they are broken. Otherwise they will always pass with the exception
+of syntax errors.
+
+Michael Kerrin <michael.kerrin at openapp.biz>
+
+$Id:$
+"""
+__docformat__="restructuredtext"
+
+import unittest
+
+import twisted.trial.unittest
+from twisted.python.failure import Failure
+
+import zope.testing.testrunner
+from zope.testing.testrunner import TestResult
+
+from twisted.trial import util
+
+old_addError = zope.testing.testrunner.TestResult.addError
+old_addFailure = zope.testing.testrunner.TestResult.addFailure
+
+RUNNING_TESTS = False
+
+def print_traceback(msg, exc_info):
+ print
+ print msg
+
+ tb = "".join(exc_info.getTraceback())
+
+ print tb
+
+
+def new_addError(self, test, exc_info):
+ if isinstance(exc_info, Failure):
+ if self.options.verbose > 2:
+ print " (%.3f ms)" % (time.time() - self._start_time)
+ self.errors.append((test, exc_info.getTraceback()))
+
+ if not RUNNING_TESTS:
+ print
+ print_traceback("Error in test %s" % test, exc_info)
+
+ if self.options.post_mortem:
+ if self.options.resume_layer:
+ print
+ print '*'*70
+ print ("Can't post-mortem debug when running a layer"
+ " as a subprocess!")
+ print '*'*70
+ print
+ else:
+ zope.testing.testrunner.post_mortem(exc_info)
+
+ self.test_width = self.last_width = 0
+ else:
+ old_addError(self, test, exc_info)
+
+
+def new_addFailure(self, test, exc_info):
+ if isinstance(exc_info, Failure):
+ if self.options.verbose > 2:
+ print " (%.3f ms)" % (time.time() - self._start_time)
+
+ self.failures.append((test, exc_info.getTraceback()))
+
+ if not RUNNING_TESTS:
+ print
+ print_traceback("Failure in test %s" % test, exc_info)
+
+ if self.options.post_mortem:
+ zope.testing.testrunner.post_mortem(exc_info)
+
+ self.test_width = self.last_width = 0
+ else:
+ old_addFailure(self, test, exc_info)
+
+
+def setUp():
+ # IMPORTANT - call the tearDown method below if you use this method!!!
+ #
+
+ # Monkey-patch the twisted.trial so has to catch and handle
+ # twisted.python.failure.Failure object correctly.
+ zope.testing.testrunner.TestResult.addError = new_addError
+ zope.testing.testrunner.TestResult.addFailure = new_addFailure
+
+def tearDown():
+ # IMPORTANT - call this method!!!!!!!!!!!!!!!!!!!!!!!!!!!
+ #
+
+ # Un-Monkey-patch the twisted.trial so has to catch and handle
+ # twisted.python.failure.Failure object correctly.
+ zope.testing.testrunner.TestResult.addError = new_addError
+ zope.testing.testrunner.TestResult.addFailure = new_addFailure
+
+ # Something funny happens with threads, the twisted reactor and zope.
+ # This fixes it.
+ util._Janitor.do_cleanThreads()
+
+#
+# Now just make sure that all this does what it says.
+#
+
+class TestZopeTests(unittest.TestCase):
+
+ def setUp(self):
+ setUp()
+
+ def tearDown(self):
+ tearDown()
+
+ def test_error(self):
+ raise Exception, "this test is a broken zope test :-)"
+
+ def test_failure(self):
+ self.assert_(False, "I am a failed zope test")
+
+ def test_assert_ok(self):
+ self.assert_(True, "I am a good test")
+
+
+class TestTrialTests(twisted.trial.unittest.TestCase):
+
+ def setUp(self):
+ setUp()
+
+ def tearDown(self):
+ tearDown()
+
+ def test_error(self):
+ raise Exception, "this test is a broken trial test :-)"
+
+ def test_failure(self):
+ self.assert_(False, "I am a failed trial test")
+
+ def test_assert_ok(self):
+ self.assert_(True, "I am a good test")
+
+
+class Options(object):
+ # simple object to simpilutate the minium zope.testing options.
+ progress = False
+ verbose = 0
+ post_mortem = None
+
+
+old_print_traceback = zope.testing.testrunner.print_traceback
+def new_print_traceback(msg, exc_info):
+ # don't print out anything when running the test tests.
+ pass
+
+
+class TestTrialIntegration(unittest.TestCase):
+
+ def run_test(self, name, tests):
+ global RUNNING_TESTS
+
+ RUNNING_TESTS = True
+ zope.testing.testrunner.print_traceback = new_print_traceback
+
+ result = zope.testing.testrunner.TestResult(Options(), tests)
+ for test in tests:
+ test(result)
+
+ RUNNING_TESTS = False
+ zope.testing.testrunner.print_traceback = old_print_traceback
+
+ return result
+
+ def test_trial_tests(self):
+ suite = unittest.makeSuite(TestTrialTests)
+ result = self.run_test('', suite)
+ self._assertResults(
+ 'zope.app.twisted.ftp.tests.test_zopetrial.TestTrialTests', result)
+
+ def test_zope_tests(self):
+ suite = unittest.makeSuite(TestZopeTests)
+ result = self.run_test('', suite)
+ self._assertResults(
+ 'zope.app.twisted.ftp.tests.test_zopetrial.TestZopeTests', result)
+
+ def _assertResults(self, basetest, result):
+ # errors
+ self.assertEqual(len(result.failures), 1)
+ self.assertEqual(result.failures[0][0].id(),
+ '%s.test_failure' % basetest)
+
+ # failures
+ self.assertEqual(len(result.errors), 1)
+ self.assertEqual(result.errors[0][0].id(), '%s.test_error' % basetest)
+
+ # ok
+ self.assertEqual(result.testsRun, 3) # ok, error, failure
+
+
+def test_suite():
+ suite = unittest.TestSuite()
+ suite.addTest(unittest.makeSuite(TestTrialIntegration))
+
+ return suite
+
+if __name__ == '__main__':
+ test_suite()
Property changes on: Zope3/branches/mkerrin-remove_trial_tests/src/zope/app/twisted/ftp/tests/test_zopetrial.py
___________________________________________________________________
Name: svn:keywords
+ Id
Name: svn:eol-style
+ native
Modified: Zope3/branches/mkerrin-remove_trial_tests/test.py
===================================================================
--- Zope3/branches/mkerrin-remove_trial_tests/test.py 2006-02-21 14:39:58 UTC (rev 41728)
+++ Zope3/branches/mkerrin-remove_trial_tests/test.py 2006-02-21 14:46:49 UTC (rev 41729)
@@ -57,6 +57,9 @@
# Get rid of twisted.conch.ssh warning
warnings.filterwarnings(
'ignore', 'PyCrypto', RuntimeWarning, 'twisted[.]conch[.]ssh')
+warnings.filterwarnings(
+ 'ignore', '', DeprecationWarning,
+ '(zope[.]app[.]twisted[.]ftp|twisted[.]test[.]test_ftp)')
result = testrunner.run(defaults)
More information about the Zope3-Checkins
mailing list