[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