Annotation of ExtFile/tests/testCacheable.py, revision 1.1.1.1

1.1       dwinter     1: #
                      2: # Tests the ZCacheable implementation
                      3: #
                      4: 
                      5: import os, sys
                      6: if __name__ == '__main__':
                      7:     execfile(os.path.join(sys.path[0], 'framework.py'))
                      8: 
                      9: from Testing import ZopeTestCase
                     10: 
                     11: ZopeTestCase.installProduct('ExtFile')
                     12: ZopeTestCase.utils.startZServer(4)
                     13: 
                     14: from Products.ExtFile.tests.ExtFileTestCase import ExtFileTestCase
                     15: from Products.ExtFile.tests.ExtFileTestCase import gifImage, notImage
                     16: 
                     17: from Testing.ZopeTestCase import Functional
                     18: from Testing.ZopeTestCase import user_name
                     19: from Testing.ZopeTestCase import user_password
                     20: user_auth = '%s:%s' % (user_name, user_password)
                     21: 
                     22: from OFS.SimpleItem import SimpleItem
                     23: from OFS.Cache import ZCM_MANAGERS
                     24: from webdav.common import rfc1123_date
                     25: from DateTime import DateTime
                     26: from StringIO import StringIO
                     27: from Products.ExtFile.ExtImage import UPLOAD_RESIZE
                     28: from Products.ExtFile.ExtFile import IStreamIterator
                     29: 
                     30: from Products.ExtFile.ExtFile import ChangePermission
                     31: from AccessControl.Permissions import ftp_access as FTPAccess
                     32: 
                     33: 
                     34: class DummyCache:
                     35:     def __init__(self):
                     36:         self.data = ''
                     37:         self.called = []
                     38:     def ZCache_set(self, ob, data, *args, **kw):
                     39:         self.called.append('set')
                     40:         self.data = data
                     41:     def ZCache_get(self, ob, *args, **kw):
                     42:         self.called.append('get')
                     43:         return self.data
                     44:     def ZCache_invalidate(self, ob):
                     45:         self.called.append('invalidate')
                     46:         self.data = ''
                     47: 
                     48: class DummyCacheManager(SimpleItem):
                     49:     meta_type = 'Dummy Cache Manager'
                     50:     def __init__(self, id):
                     51:         self.id = id
                     52:         self.cache = DummyCache()
                     53:     def ZCacheManager_getCache(self):
                     54:         return self.cache
                     55: 
                     56: 
                     57: class ExtFileCacheableTests(Functional, ExtFileTestCase):
                     58: 
                     59:     def afterSetUp(self):
                     60:         ExtFileTestCase.afterSetUp(self)
                     61:         # Set up cache
                     62:         self.folder.cache = DummyCacheManager('cache')
                     63:         setattr(self.folder, ZCM_MANAGERS, ('cache',))
                     64:         self.cache = self.folder.cache.cache
                     65:         # Set up cached file
                     66:         self.addExtFile(id='file.exe', file=notImage)
                     67:         self.file.ZCacheable_setManagerId('cache')
                     68:         self.file_path = self.file.absolute_url(1)
                     69: 
                     70:     def test304ResponseSetsCache(self):
                     71:         response = self.publish(self.file_path,
                     72:                                 env={'HTTP_IF_MODIFIED_SINCE': rfc1123_date(DateTime()+7)},
                     73:                                 basic=user_auth)
                     74:         self.assertEqual(response.getStatus(), 304)
                     75:         self.assertEqual(self.cache.data, None)
                     76:         self.assertEqual(self.cache.called, ['set'])
                     77: 
                     78:     def test200ResponseSetsCache(self):
                     79:         response = self.publish(self.file_path, basic=user_auth)
                     80:         self.assertEqual(response.getStatus(), 200)
                     81:         self.assertEqual(self.cache.data, None)
                     82:         self.assertEqual(self.cache.called, ['set'])
                     83: 
                     84:     def testManageUploadInvalidatesCache(self):
                     85:         self.file.manage_upload(open(gifImage, 'rb'))
                     86:         self.assertEqual(self.cache.data, '')
                     87:         self.assertEqual(self.cache.called, ['invalidate'])
                     88: 
                     89:     def testManageFileUploadInvalidatesCache(self):
                     90:         self.file.manage_file_upload(gifImage)
                     91:         self.assertEqual(self.cache.data, '')
                     92:         self.assertEqual(self.cache.called, ['invalidate'])
                     93: 
                     94:     def testManageHTTPUploadInvalidatesCache(self):
                     95:         self.file.manage_http_upload(url=self.app.GifImage.absolute_url())
                     96:         self.assertEqual(self.cache.data, '')
                     97:         self.assertEqual(self.cache.called, ['invalidate'])
                     98: 
                     99:     def testPUTInvalidatesCache(self):
                    100:         self.setPermissions([ChangePermission])
                    101:         response = self.publish(self.file_path,
                    102:                                 request_method='PUT',
                    103:                                 env={'CONTENT_TYPE': 'image/gif'},
                    104:                                 stdin=open(gifImage, 'rb'),
                    105:                                 basic=user_auth)
                    106:         self.assertEqual(response.getStatus(), 204)
                    107:         self.assertEqual(self.cache.data, '')
                    108:         self.assertEqual(self.cache.called, ['invalidate'])
                    109: 
                    110:     def testManageEditInvalidatesCache(self):
                    111:         self.file.manage_editExtFile(title='Foo')
                    112:         self.assertEqual(self.cache.data, '')
                    113:         self.assertEqual(self.cache.called, ['invalidate'])
                    114: 
                    115:     def testRedirectDefaultViewReturns304(self):
                    116:         self.file.manage_file_upload(gifImage)
                    117:         self.file.redirect_default_view = 1
                    118:         response = self.publish(self.file_path,
                    119:                                 env={'HTTP_IF_MODIFIED_SINCE': rfc1123_date(DateTime()+7)},
                    120:                                 basic=user_auth)
                    121:         self.assertEqual(response.getStatus(), 304)
                    122:         self.assertEqual(response.getHeader('Location'), None)
                    123: 
                    124:     def testRedirectDefaultViewReturns302(self):
                    125:         self.file.redirect_default_view = 1
                    126:         saved = os.environ.copy()
                    127:         os.environ['EXTFILE_STATIC_PATH'] = '/static'
                    128:         try:
                    129:             response = self.publish(self.file_path, basic=user_auth)
                    130:         finally:
                    131:             os.environ = saved
                    132:         self.assertEqual(response.getStatus(), 302)
                    133:         self.assertEqual(response.getHeader('Location'),
                    134:                          self.app.REQUEST.SERVER_URL+'/static/file.exe')
                    135: 
                    136:     if IStreamIterator is not None:
                    137: 
                    138:         def testStreamIterator(self):
                    139:             response = self.publish(self.file_path, basic=user_auth)
                    140:             self.assertEqual(response.getStatus(), 200)
                    141:             # FIXME
                    142:             self.failUnless(response.getBody().startswith('<Products.ExtFile.ExtFile.stream_iterator'))
                    143:             self.assertEqual(self.cache.called, ['set'])
                    144: 
                    145:         def testFTPgetStreamIterator(self):
                    146:             self.setPermissions([FTPAccess])
                    147:             response = self.publish(self.file_path+'/manage_FTPget',
                    148:                                     basic=user_auth)
                    149:             self.assertEqual(response.getStatus(), 200)
                    150:             # FIXME
                    151:             self.failUnless(response.getBody().startswith('<Products.ExtFile.ExtFile.stream_iterator'))
                    152:             self.assertEqual(self.cache.called, ['set'])
                    153: 
                    154:         def testNoRequestNoStreamIterator(self):
                    155:             body = self.file.index_html()
                    156:             self.assertEqual(self.app.REQUEST.RESPONSE.getStatus(), 200)
                    157:             # Not taken from cache
                    158:             self.assertEqual(body, open(notImage, 'rb').read())
                    159:             self.assertEqual(self.cache.called, [])
                    160: 
                    161: 
                    162: class ExtImageCacheableTests(ExtFileCacheableTests):
                    163: 
                    164:     def afterSetUp(self):
                    165:         ExtFileTestCase.afterSetUp(self)
                    166:         # Set up cache
                    167:         self.folder.cache = DummyCacheManager('cache')
                    168:         setattr(self.folder, ZCM_MANAGERS, ('cache',))
                    169:         self.cache = self.folder.cache.cache
                    170:         # Set up cached image
                    171:         self.addExtImage(id='file.exe', file=notImage)
                    172:         self.image.ZCacheable_setManagerId('cache')
                    173:         self.file = self.image
                    174:         self.file_path = self.image.absolute_url(1)
                    175: 
                    176:     def testManageUploadPreviewInvalidatesCache(self):
                    177:         self.image.manage_upload(open(gifImage, 'rb'), is_preview=1)
                    178:         self.assertEqual(self.cache.data, '')
                    179:         self.assertEqual(self.cache.called, ['invalidate'])
                    180: 
                    181:     def testManageUploadPreviewResizeInvalidatesCache(self):
                    182:         self.image.manage_upload(open(gifImage, 'rb'), is_preview=1, create_prev=UPLOAD_RESIZE)
                    183:         self.assertEqual(self.cache.data, '')
                    184:         self.assertEqual(self.cache.called, ['invalidate'])
                    185: 
                    186:     def testManageFileUploadPreviewInvalidatesCache(self):
                    187:         self.image.manage_file_upload(gifImage, is_preview=1)
                    188:         self.assertEqual(self.cache.data, '')
                    189:         self.assertEqual(self.cache.called, ['invalidate'])
                    190: 
                    191:     def testManageFileUploadPreviewResizeInvalidatesCache(self):
                    192:         self.image.manage_file_upload(gifImage, is_preview=1, create_prev=UPLOAD_RESIZE)
                    193:         self.assertEqual(self.cache.data, '')
                    194:         self.assertEqual(self.cache.called, ['invalidate'])
                    195: 
                    196:     def testManageHTTPUploadPreviewInvalidatesCache(self):
                    197:         self.image.manage_http_upload(url=self.app.GifImage.absolute_url(), is_preview=1)
                    198:         self.assertEqual(self.cache.data, '')
                    199:         self.assertEqual(self.cache.called, ['invalidate'])
                    200: 
                    201:     def testManageCreatePrevInvalidatesCache(self):
                    202:         self.image.manage_file_upload(gifImage)
                    203:         self.cache.ZCache_invalidate(None)
                    204:         self.cache.called = []
                    205:         self.image.manage_create_prev(maxx=100, maxy=100, ratio=1)
                    206:         self.assertEqual(self.cache.data, '')
                    207:         self.assertEqual(self.cache.called, ['invalidate'])
                    208: 
                    209:     def testManageDelPrevInvalidatesCache(self):
                    210:         self.image.manage_file_upload(gifImage)
                    211:         self.image.manage_create_prev(maxx=100, maxy=100, ratio=1)
                    212:         self.cache.ZCache_invalidate(None)
                    213:         self.cache.called = []
                    214:         self.image.manage_del_prev()
                    215:         self.assertEqual(self.cache.data, '')
                    216:         self.assertEqual(self.cache.called, ['invalidate'])
                    217: 
                    218: 
                    219: def test_suite():
                    220:     from unittest import TestSuite, makeSuite
                    221:     suite = TestSuite()
                    222:     suite.addTest(makeSuite(ExtFileCacheableTests))
                    223:     suite.addTest(makeSuite(ExtImageCacheableTests))
                    224:     return suite
                    225: 
                    226: if __name__ == '__main__':
                    227:     framework()
                    228: 

FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>