[Zope-Checkins] CVS: Zope3/lib/python/Zope/Server/VFS/tests - testPublisherFilesystem.py:1.1.2.1 ReadFilesystemTests.py:1.1.2.2 WriteFilesystemTests.py:1.1.2.2

[Shane Hathaway]

Update of /cvs-repository/Zope3/lib/python/Zope/Server/VFS/tests
In directory cvs.zope.org:/tmp/cvs-serv8322/VFS/tests

Modified Files:
      Tag: Zope3-Server-Branch
	ReadFilesystemTests.py WriteFilesystemTests.py 
Added Files:
      Tag: Zope3-Server-Branch
	testPublisherFilesystem.py 
Log Message:
Functioning, tested VFS publication


=== Added File Zope3/lib/python/Zope/Server/VFS/tests/testPublisherFilesystem.py ===
##############################################################################
#
# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (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.
#
##############################################################################
"""

$Id: testPublisherFilesystem.py,v 1.1.2.1 2002/04/12 20:51:13 shane Exp $
"""


import unittest
from StringIO import StringIO

from Zope.Server.VFS.PublisherFileSystem import PublisherFileSystem
from Zope.Publisher.VFS.VFSRequest import VFSRequest
from Zope.Publisher.DefaultPublication import DefaultPublication
from Zope.Publisher.VFS.IVFSFilePublisher import IVFSFilePublisher
from Zope.Publisher.VFS.IVFSDirectoryPublisher import IVFSDirectoryPublisher
from Zope.Publisher.mapply import mapply

from WriteFilesystemTests import WriteFilesystemTests


class VFSPublication (DefaultPublication):
    # This class will not be needed if we move callObject().

    def callObject(self, request, ob):
        method = getattr(ob, request.method)
        return mapply(method, request.getPositionalArguments(), request)
    
    def traverseName(self, request, ob, name):
        return ob.publishTraverse(request, name)


class TestFile:

    __implements__ = IVFSFilePublisher

    def __init__(self, data=''):
        self.data = data

    def publishTraverse(self, request, name):
        """See IVFSPublisher."""
        raise OSError, 'Cannot traverse TestFiles'

    def isdir(self):
        """See IVFSObjectPublisher."""
        return 0

    def isfile(self):
        """See IVFSObjectPublisher."""
        return 1

    def stat(self):
        """See IVFSObjectPublisher."""
        raise NotImplementedError

    def read(self, mode, outstream, start=0, end=-1):
        """See IVFSFilePublisher."""
        if end >= 0:
            s = self.data[start:end]
        else:
            s = self.data[start:]
        outstream.write(s)
        
    def write(self, mode, instream, start=0):
        """See IVFSFilePublisher."""
        s = instream.read()
        if 'a' in mode:
            self.data = self.data + s
        else:
            self.data = self.data[:start] + s + self.data[start + len(s):]




class TestDirectory:

    __implements__ = IVFSDirectoryPublisher

    def __init__(self, items={}):
        self.items = items.copy()

    def publishTraverse(self, request, name):
        """See IVFSPublisher."""
        return self.items[name]

    def isdir(self):
        """See IVFSObjectPublisher."""
        return 1

    def isfile(self):
        """See IVFSObjectPublisher."""
        return 0

    def stat(self):
        """See IVFSObjectPublisher."""
        raise NotImplementedError

    def exists(self, name):
        """See IVFSDirectoryPublisher."""
        return self.items.has_key(name)

    def listdir(self, with_stats=0, pattern='*'):
        """See IVFSDirectoryPublisher."""
        if with_stats or pattern != '*':
            raise NotImplementedError
        return self.items.keys()

    def mkdir(self, name, mode=0777):
        """See IVFSDirectoryPublisher."""
        self.items[name] = TestDirectory()

    def remove(self, name):
        """See IVFSDirectoryPublisher."""
        del self.items[name]

    def rmdir(self, name):
        """See IVFSDirectoryPublisher."""
        del self.items[name]

    def rename(self, old, new):
        """See IVFSDirectoryPublisher."""
        if self.items.has_key(new):
            raise OSError, 'Name conflict'
        self.items[new] = self.items[old]
        del self.items[old]

    def writefile(self, name, mode, instream, start=0):
        """See IVFSDirectoryPublisher."""
        if not self.items.has_key(name):
            self.items[name] = TestFile()
        self.items[name].write(mode, instream, start)

    def check_writable(self, name):
        """See IVFSDirectoryPublisher."""
        if not self.items.has_key(name):
            return 1
        return self.items[name].isfile()


class PublisherFileSystemTests(unittest.TestCase, WriteFilesystemTests):
    """This test is constructed in a way that it builds up a directory
       structure, whcih is removed at the end.
    """

    filesystem_class = PublisherFileSystem

    # XXX for now, the publisher always eats exceptions.
    check_exceptions = 0

    def setUp(self):

        app = TestDirectory()
        
        pub = VFSPublication(app)

        def request_factory(input_stream, output_steam, env):
            request = VFSRequest(input_stream, output_steam, env)
            request.setPublication(pub)
            return request

        self.filesystem = PublisherFileSystem(None, request_factory)
        self.filesystem.mkdir(self.dir_name)
        s = StringIO(self.file_contents)
        self.filesystem.writefile(self.file_name, 'w', s)

    def tearDown(self):
        pass



def test_suite():
    loader = unittest.TestLoader()
    return loader.loadTestsFromTestCase(PublisherFileSystemTests)

if __name__=='__main__':
    unittest.TextTestRunner().run( test_suite() )


=== Zope3/lib/python/Zope/Server/VFS/tests/ReadFilesystemTests.py 1.1.2.1 => 1.1.2.2 ===
     file_contents = 'Lengthen your stride'
 
+    check_exceptions = 1
+
 
     def testExists(self):
         self.failUnless(self.filesystem.exists(self.dir_name))


=== Zope3/lib/python/Zope/Server/VFS/tests/WriteFilesystemTests.py 1.1.2.1 => 1.1.2.2 ===
 
 
-    def testCheckWriteable(self):
-        # Can't overwrite a directory.
-        self.assertRaises(
-            IOError, self.filesystem.check_writable, self.dir_name)
+    def testCheckWritable(self):
+        if self.check_exceptions:
+            # Can't overwrite a directory.
+            self.assertRaises(
+                IOError, self.filesystem.check_writable, self.dir_name)
         # Can overwrite a file.
         try:
             self.filesystem.check_writable(self.file_name)