Return to testCacheable.py CVS log

Up to [Repository] / ExtFile / tests

Auf der Basis http://www.zope.org/Members/shh/ExtFile Version 1.5.4

mit zlog ersetzt durch logging

    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: