File:  [Repository] / ExtFile / tests / testTransactions.py
Revision 1.1.1.1 (vendor branch): download - view: text, annotated - select for diffs - revision graph
Wed Jan 24 16:53:50 2007 UTC (17 years, 5 months ago) by dwinter
Branches: first, MAIN
CVS tags: release, HEAD
Auf der Basis http://www.zope.org/Members/shh/ExtFile Version 1.5.4

mit zlog ersetzt durch logging


    1: #
    2: # Tests transactional behavior of ExtFiles
    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, jpegImage, notImage
   16: from Products.ExtFile.tests.ExtFileTestCase import copymove_perms
   17: from Products.ExtFile.ExtImage import NO_PREVIEW, GENERATE, UPLOAD_NORESIZE, UPLOAD_RESIZE
   18: from Products.ExtFile import transaction, TM
   19: from Acquisition import aq_base
   20: 
   21: 
   22: class TestTransactions(ExtFileTestCase):
   23:     '''Test ExtFile/ExtImage transaction awareness.'''
   24: 
   25:     def beforeClose(self):
   26:         transaction.commit()  # Commit the cleaned-up fixture
   27: 
   28:     def testAddFileCommit(self):
   29:         self.addExtFile(id='file', file=gifImage)
   30:         transaction.commit()
   31:         self.failUnless(self._exists('file.exe'))
   32: 
   33:     def testAddFileAbort(self):
   34:         # Aborting the transaction leaves the repository empty
   35:         self.addExtFile(id='file', file=gifImage)
   36:         transaction.savepoint(1)   # Wuhuu, force object rollback
   37:         transaction.abort()
   38:         self.failIf(hasattr(aq_base(self.folder), 'file'))
   39:         self.failIf(self._exists('file.exe'))
   40:         self.failIf(self._exists('file.exe.tmp'))
   41: 
   42:     def testAddImageCommit(self):
   43:         self.addExtImage(id='image', file=gifImage)
   44:         transaction.commit()
   45:         self.failUnless(self._exists('image.gif'))
   46: 
   47:     def testAddImageAbort(self):
   48:         # Aborting the transaction leaves the repository empty
   49:         self.addExtImage(id='image', file=gifImage)
   50:         transaction.savepoint(1)   # Wuhuu, force object rollback
   51:         transaction.abort()
   52:         self.failIf(hasattr(aq_base(self.folder), 'image'))
   53:         self.failIf(self._exists('image.gif'))
   54:         self.failIf(self._exists('image.gif.tmp'))
   55: 
   56:     def testAddImageAbortWithPreview(self):
   57:         # Aborting the transaction leaves the repository empty
   58:         self.addExtImage(id='image', file=gifImage)
   59:         self.image.manage_create_prev(maxx=10, maxy=10)
   60:         transaction.savepoint(1)   # Wuhuu, force object rollback
   61:         transaction.abort()
   62:         self.failIf(hasattr(aq_base(self.folder), 'image'))
   63:         self.failIf(self._exists('image.gif'))
   64:         self.failIf(self._exists('image.gif.tmp'))
   65:         self.failIf(self._exists('image.jpg'))
   66:         self.failIf(self._exists('image.jpg.tmp'))
   67: 
   68:     def testAddMoreThanOneFileInSeparateTransactions(self):
   69:         # First file
   70:         self.addExtFile(id='fred', file=notImage)
   71:         transaction.commit()
   72:         self.failUnless(self._exists('fred.exe'))
   73:         self.failIf(self._exists('fred.exe.tmp'))
   74:         # Second file
   75:         self.addExtFile(id='barney', file=notImage)
   76:         transaction.commit()
   77:         self.failUnless(self._exists('barney.exe'))
   78:         self.failIf(self._exists('barney.exe.tmp'))
   79:         # Third file
   80:         self.addExtFile(id='betty', file=notImage)
   81:         transaction.commit()
   82:         self.failUnless(self._exists('betty.exe'))
   83:         self.failIf(self._exists('betty.exe.tmp'))
   84: 
   85:     def testUploadFileIntoExistingInSeparateTransactions(self):
   86:         # Create a file 'fred'
   87:         self.addExtFile(id='fred', file=notImage)
   88:         self.failUnless(TM.contains(self.file))
   89:         self.failUnless(self._exists('fred.exe.tmp'))
   90:         transaction.commit()
   91:         self.failIf(TM.contains(self.file))
   92:         self.failUnless(self._exists('fred.exe'))
   93:         self.failIf(self._exists('fred.exe.tmp'))
   94:         # Upload new file into 'fred'
   95:         self.file.manage_file_upload(file=gifImage)
   96:         self.failUnless(TM.contains(self.file))
   97:         self.failUnless(self._exists('fred.exe'))
   98:         self.failUnless(self._exists('fred.exe.tmp'))
   99:         transaction.commit()
  100:         self.failIf(TM.contains(self.file))
  101:         self.failUnless(self._exists('fred.exe'))
  102:         self.failIf(self._exists('fred.exe.tmp'))
  103: 
  104: 
  105: class TestTransactionManager(ExtFileTestCase):
  106: 
  107:     def afterSetUp(self):
  108:         ExtFileTestCase.afterSetUp(self)
  109:         self.addExtFile(id='file', file=notImage)
  110:         self.file._register()
  111: 
  112:     def beforeClose(self):
  113:         transaction.commit()  # Commit the cleaned-up fixture
  114: 
  115:     def testBegin(self):
  116:         self.assertEqual(self.file._v_begin_called, 1)
  117: 
  118:     def testFinish(self):
  119:         transaction.commit()
  120:         self.assertEqual(self.file._v_finish_called, 1)
  121: 
  122:     def testAbort(self):
  123:         transaction.abort()
  124:         self.assertEqual(self.file._v_abort_called, 1)
  125: 
  126: 
  127: class TestExtFileTransactions(ExtFileTestCase):
  128: 
  129:     def afterSetUp(self):
  130:         ExtFileTestCase.afterSetUp(self)
  131:         self.setPermissions(copymove_perms)
  132:         self.folder.manage_addFolder('subfolder')
  133:         self.subfolder = self.folder.subfolder
  134: 
  135:     def beforeClose(self):
  136:         transaction.commit()  # Commit the cleaned-up fixture
  137: 
  138:     def testManageFileUploadCreatesTempFile(self):
  139:         self.addExtFile(id='file', file='')
  140:         self.file.manage_file_upload(file=notImage)
  141:         self.failUnless(self._exists('file.exe.tmp'), 'Missing .tmp file')
  142:         self.failIf(self._exists('file.exe'), 'No .tmp file used')
  143: 
  144:     def testManageHTTPUploadCreatesTempFile(self):
  145:         self.addExtFile(id='file', file='')
  146:         self.file.manage_http_upload(url=self.app.NotImage.absolute_url())
  147:         self.failUnless(self._exists('file.exe.tmp'), 'Missing .tmp file')
  148:         self.failIf(self._exists('file.exe'), 'No .tmp file used')
  149: 
  150:     def testPUTCreatesTempFile(self):
  151:         self.addExtFile(id='file', file='')
  152:         request = self.app.REQUEST
  153:         request['BODYFILE'] = open(notImage, 'rb')
  154:         self.file.PUT(request, request.RESPONSE)
  155:         self.failUnless(self._exists('file.exe.tmp'), 'Missing .tmp file')
  156:         self.failIf(self._exists('file.exe'), 'No .tmp file used')
  157: 
  158:     def testFinishCommitsTempFile(self):
  159:         self.addExtFile(id='file', file=notImage)
  160:         self.failUnless(self._exists('file.exe.tmp'))
  161:         self.file._finish()
  162:         self.failUnless(self._exists('file.exe'))
  163:         self.failIf(self._exists('file.exe.tmp'))
  164: 
  165:     def testAbortNukesTempFile(self):
  166:         self.addExtFile(id='file', file=notImage)
  167:         self.failUnless(self._exists('file.exe.tmp'))
  168:         self.file._abort()
  169:         self.failIf(self._exists('file.exe.tmp'))
  170:         self.failIf(self._exists('file.exe'))
  171: 
  172:     def testUndoCreatesTempFile(self):
  173:         self.addExtFile(id='file', file=notImage)
  174:         self.file._finish()
  175:         os.rename(self._fsname('file.exe'), self._fsname('file.exe.undo'))
  176:         self.file._undo()
  177:         self.failIf(self._exists('file.exe.undo'))
  178:         self.failUnless(self._exists('file.exe.tmp'))
  179: 
  180:     def testIsBrokenUsesTempFile(self):
  181:         self.addExtFile(id='file', file=notImage)
  182:         self.failUnless(self._exists('file.exe.tmp'))
  183:         self.failIf(self._exists('file.exe'))
  184:         self.failIf(self.file.is_broken())
  185: 
  186:     def testIsBrokenUsesMainFileIfTempFileNotPresent(self):
  187:         self.addExtFile(id='file', file=notImage)
  188:         self.file._finish()
  189:         self.failUnless(self._exists('file.exe'))
  190:         self.failIf(self._exists('file.exe.tmp'))
  191:         self.failIf(self.file.is_broken())
  192: 
  193:     def testIsBrokenTriesToUndoIfMainFileNotPresent(self):
  194:         self.addExtFile(id='file', file=notImage)
  195:         os.rename(self._fsname('file.exe.tmp'), self._fsname('file.exe.undo'))
  196:         self.failUnless(self._exists('file.exe.undo'))
  197:         self.failIf(self._exists('file.exe'))
  198:         self.failIf(self.file.is_broken())
  199:         self.failUnless(self._exists('file.exe.tmp'))
  200:         self.failIf(self._exists('file.exe.undo'))
  201: 
  202:     def testIsBrokenReturnsTrueIfBroken(self):
  203:         self.addExtFile(id='file', file=notImage)
  204:         os.remove(self._fsname('file.exe.tmp'))
  205:         self.failUnless(self.file.is_broken())
  206: 
  207:     def testGetSizeUsesTempFile(self):
  208:         self.addExtFile(id='file', file=notImage)
  209:         self.failUnless(self._exists('file.exe.tmp'))
  210:         self.failIf(self._exists('file.exe'))
  211:         self.failIfEqual(self.file.get_size(), 0)
  212: 
  213:     def testGetSizeUsesMainFileIfTempFileNotPresent(self):
  214:         self.addExtFile(id='file', file=notImage)
  215:         self.file._finish()
  216:         self.failUnless(self._exists('file.exe'))
  217:         self.failIf(self._exists('file.exe.tmp'))
  218:         self.failIfEqual(self.file.get_size(), 0)
  219: 
  220:     def testGetSizeTriesToUndoIfMainFileNotPresent(self):
  221:         self.addExtFile(id='file', file=notImage)
  222:         self.file._finish()
  223:         os.rename(self._fsname('file.exe'), self._fsname('file.exe.undo'))
  224:         self.failUnless(self._exists('file.exe.undo'))
  225:         self.failIf(self._exists('file.exe'))
  226:         self.failIfEqual(self.file.get_size(), 0)
  227:         self.failUnless(self._exists('file.exe.tmp'))
  228:         self.failIf(self._exists('file.exe.undo'))
  229: 
  230:     def testGetSizeReturnsZeroIfBroken(self):
  231:         self.addExtFile(id='file', file=notImage)
  232:         os.remove(self._fsname('file.exe.tmp'))
  233:         self.assertEqual(self.file.get_size(), 0)
  234: 
  235:     def testManageFileUploadRegistersWithTM(self):
  236:         self.addExtFile(id='file', file='')
  237:         self.assertEqual(getattr(self.file, '_v_begin_called', 0), 0)
  238:         self.file.manage_file_upload(file=notImage)
  239:         self.assertEqual(getattr(self.file, '_v_begin_called', 0), 1)
  240: 
  241:     def testManageHTTPUploadRegistersWithTM(self):
  242:         self.addExtFile(id='file', file='')
  243:         self.assertEqual(getattr(self.file, '_v_begin_called', 0), 0)
  244:         self.file.manage_http_upload(url=self.app.NotImage.absolute_url())
  245:         self.assertEqual(getattr(self.file, '_v_begin_called', 0), 1)
  246: 
  247:     def testPUTRegistersWithTM(self):
  248:         self.addExtFile(id='file', file='')
  249:         request = self.app.REQUEST
  250:         request['BODYFILE'] = open(notImage, 'rb')
  251:         self.assertEqual(getattr(self.file, '_v_begin_called', 0), 0)
  252:         self.file.PUT(request, request.RESPONSE)
  253:         self.assertEqual(getattr(self.file, '_v_begin_called', 0), 1)
  254: 
  255:     def testUndoRegistersWithTM(self):
  256:         self.addExtFile(id='file', file=notImage)
  257:         os.rename(self._fsname('file.exe.tmp'), self._fsname('file.exe.undo'))
  258:         self.file._v_begin_called = 0   # Clear
  259:         TM.remove(self.file)            # Clear
  260:         self.assertEqual(getattr(self.file, '_v_begin_called', 0), 0)
  261:         self.file._undo()
  262:         self.assertEqual(getattr(self.file, '_v_begin_called', 0), 1)
  263:         self.failUnless(self._exists('file.exe.tmp'))
  264: 
  265:     def testGetNewUfnSkipsTempFile(self):
  266:         self.addExtFile(id='file', file=notImage)
  267:         self.failUnless(self._exists('file.exe.tmp'))
  268:         fn = self.file._get_new_ufn(self.file.filename)
  269:         self.assertEqual(fn, ['file.1.exe'])
  270: 
  271:     def testGetFileToServeDoesNotUseTempFile(self):
  272:         self.addExtFile(id='file', file=notImage)
  273:         self.failUnless(self._exists('file.exe.tmp'))
  274:         fn = self.file._get_file_to_serve()[0]
  275:         self.assertEqual(fn, ['file.exe'])
  276: 
  277:     def testManageBeforeDeleteUsesTempFile(self):
  278:         self.addExtFile(id='file', file=notImage)
  279:         self.failUnless(self._exists('file.exe.tmp'))
  280:         self.folder._delObject('file')
  281:         self.failUnless(self._exists('file.exe.undo'))
  282:         self.failIf(self._exists('file.exe'))
  283: 
  284:     def testManageBeforeDeleteNukesMainFileIfTempFilePresent(self):
  285:         self.addExtFile(id='file', file=notImage)
  286:         self.file._finish()
  287:         self.file.manage_file_upload(file=notImage)
  288:         self.failUnless(self._exists('file.exe'))
  289:         self.failUnless(self._exists('file.exe.tmp'))
  290:         self.folder._delObject('file')
  291:         self.failUnless(self._exists('file.exe.undo'))
  292:         self.failIf(self._exists('file.exe'))
  293: 
  294:     def testManageBeforeDeleteUsesMainFileIfTempFileNotPresent(self):
  295:         self.addExtFile(id='file', file=notImage)
  296:         self.file._finish()
  297:         self.failUnless(self._exists('file.exe'))
  298:         self.failIf(self._exists('file.exe.tmp'))
  299:         self.folder._delObject('file')
  300:         self.failUnless(self._exists('file.exe.undo'))
  301:         self.failIf(self._exists('file.exe'))
  302: 
  303:     def testManageAfterCloneCreatesTempFile(self):
  304:         self.addExtFile(id='file', file=notImage)
  305:         self.file._finish()
  306:         transaction.savepoint(1) # Need a _p_oid
  307:         cb = self.folder.manage_copyObjects(['file'])
  308:         self.subfolder.manage_pasteObjects(cb)
  309:         self.failUnless(self._exists('file.exe'))       # original
  310:         self.failUnless(self._exists('file.1.exe.tmp'))  # copy
  311:         self.assertEqual(self.subfolder.file.filename, ['file.1.exe'])
  312: 
  313:     def testManageAfterCloneUsesTempFileAsSource(self):
  314:         self.addExtFile(id='file', file=notImage)
  315:         transaction.savepoint(1) # Need a _p_oid
  316:         self.subfolder.manage_clone(self.file, 'file')
  317:         self.failUnless(self._exists('file.exe.tmp'))   # original
  318:         self.failUnless(self._exists('file.1.exe.tmp'))  # copy
  319: 
  320:     def testManageAfterCloneUsesMainFileIfTempFileNotPresent(self):
  321:         self.addExtFile(id='file', file=notImage)
  322:         self.file._finish()
  323:         transaction.savepoint(1) # Need a _p_oid
  324:         self.subfolder.manage_clone(self.file, 'file')
  325:         self.failUnless(self._exists('file.exe'))       # original
  326:         self.failUnless(self._exists('file.1.exe.tmp'))  # copy
  327: 
  328:     def testManageAfterCloneTriesToUndoIfTempFileNotPresent(self):
  329:         self.addExtFile(id='file', file=notImage)
  330:         self.file._finish()
  331:         os.rename(self._fsname('file.exe'), self._fsname('file.exe.undo'))
  332:         transaction.savepoint(1) # Need a _p_oid
  333:         self.subfolder.manage_clone(self.file, 'file')
  334:         self.failUnless(self._exists('file.exe.tmp'))   # restored original
  335:         self.failUnless(self._exists('file.1.exe.tmp'))  # copy
  336: 
  337:     def testManageAfterCloneRegistersWithTM(self):
  338:         self.addExtFile(id='file', file=notImage)
  339:         self.file._finish()
  340:         self.file._v_begin_called = 0   # Clear
  341:         TM.remove(self.file)            # Clear
  342:         transaction.savepoint(1) # Need a _p_oid
  343:         self.assertEqual(getattr(self.subfolder.file, '_v_begin_called', 0), 0)
  344:         self.subfolder.manage_clone(self.file, 'file')
  345:         self.assertEqual(getattr(self.subfolder.file, '_v_begin_called', 0), 1)
  346: 
  347: 
  348: class TestExtImageTransactions(ExtFileTestCase):
  349: 
  350:     def afterSetUp(self):
  351:         ExtFileTestCase.afterSetUp(self)
  352:         self.setPermissions(copymove_perms)
  353:         self.folder.manage_addFolder('subfolder')
  354:         self.subfolder = self.folder.subfolder
  355: 
  356:     def beforeClose(self):
  357:         transaction.commit()  # Commit the cleaned-up fixture
  358: 
  359:     def testManageFileUploadCreatesTempImage(self):
  360:         self.addExtImage(id='image', file='')
  361:         self.image.manage_file_upload(file=gifImage)
  362:         self.failUnless(self._exists('image.gif.tmp'), 'Missing .tmp file')
  363:         self.failIf(self._exists('image.gif'), 'No .tmp file used')
  364: 
  365:     def testManageFileUploadCreatesTempPreview(self):
  366:         self.addExtImage(id='image', file='')
  367:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  368:         self.failUnless(self._exists('image.jpg.tmp'), 'Missing .tmp file')
  369:         self.failIf(self._exists('image.jpg'), 'No .tmp file used')
  370: 
  371:     def testManageFileUploadCreatesTempPreviewIfResize(self):
  372:         self.addExtImage(id='image', file='')
  373:         self.image.manage_file_upload(file=jpegImage, is_preview=1, create_prev=UPLOAD_RESIZE, maxx=10, maxy=10)
  374:         self.failUnless(self._exists('image.jpg.tmp'), 'Missing .tmp file')
  375:         self.failIf(self._exists('image.jpg'), 'No .tmp file used')
  376: 
  377:     def testManageFileUploadCreatesTempPreviewIfGenerate(self):
  378:         self.addExtImage(id='image', file='')
  379:         self.image.manage_file_upload(file=jpegImage, create_prev=GENERATE, maxx=10, maxy=10)
  380:         self.failUnless(self._exists('image.jpg.tmp'), 'Missing .tmp file') # main image
  381:         self.failIf(self._exists('image.jpg'), 'No .tmp file used')
  382:         self.failUnless(self._exists('image.1.jpg.tmp'), 'Missing .tmp file') # preview
  383:         self.failIf(self._exists('image.1.jpg'), 'No .tmp file used')
  384: 
  385:     def testManageHTTPUploadCreatesTempImage(self):
  386:         self.addExtImage(id='image', file='')
  387:         self.image.manage_http_upload(url=self.app.GifImage.absolute_url())
  388:         self.failUnless(self._exists('image.gif.tmp'), 'Missing .tmp file')
  389:         self.failIf(self._exists('image.gif'), 'No .tmp file used')
  390: 
  391:     def testManageHTTPUploadCreatesTempPreview(self):
  392:         self.addExtImage(id='image', file='')
  393:         self.image.manage_http_upload(url=self.app.GifImage.absolute_url(), is_preview=1)
  394:         self.failUnless(self._exists('image.gif.tmp'), 'Missing .tmp file')
  395:         self.failIf(self._exists('image.gif'), 'No .tmp file used')
  396: 
  397:     #def testManageHTTPUploadCreatesTempPreviewIfAutogen(self):
  398:     #    self.addExtImage(id='image', file='')
  399:     #    self.image.manage_file_upload(file=jpegImage, is_preview=1)
  400:     #    preview_size = self.image.get_prev_size()
  401:     #    self.image.manage_http_upload(url=self.app.GifImage.absolute_url())
  402:     #    self.failUnless(self._exists('image.gif.tmp'), 'Missing .tmp file') # main image
  403:     #    self.failIf(self._exists('image.gif'), 'No .tmp file used')
  404:     #    self.failUnless(self._exists('image.jpg.tmp'), 'Missing .tmp file') # preview
  405:     #    self.failIf(self._exists('image.jpg'), 'No .tmp file used')
  406:     #    # The preview file should no longer be the same
  407:     #    self.failIfEqual(preview_size, self.image.get_prev_size())
  408: 
  409:     def testPUTCreatesTempImage(self):
  410:         self.addExtImage(id='image', file='')
  411:         request = self.app.REQUEST
  412:         request['BODYFILE'] = open(jpegImage, 'rb')
  413:         self.image.PUT(request, request.RESPONSE)
  414:         self.failUnless(self._exists('image.jpg.tmp'), 'Missing .tmp file')
  415:         self.failIf(self._exists('image.jpg'), 'No .tmp file used')
  416: 
  417:     def testPUTCreatesTempPreviewIfAutogen(self):
  418:         self.addExtImage(id='image', file='')
  419:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  420:         preview_size = self.image.get_prev_size()
  421:         request = self.app.REQUEST
  422:         request['BODYFILE'] = open(gifImage, 'rb')
  423:         self.image.PUT(request, request.RESPONSE)
  424:         self.failUnless(self._exists('image.gif.tmp'), 'Missing .tmp file') # main image
  425:         self.failIf(self._exists('image.gif'), 'No .tmp file used')
  426:         self.failUnless(self._exists('image.jpg.tmp'), 'Missing .tmp file') # preview
  427:         self.failIf(self._exists('image.jpg'), 'No .tmp file used')
  428:         # The preview file should no longer be the same
  429:         self.failIfEqual(preview_size, self.image.get_prev_size())
  430: 
  431:     def testFinishCommitsTempImage(self):
  432:         self.addExtImage(id='image', file=jpegImage)
  433:         self.failUnless(self._exists('image.jpg.tmp'))
  434:         self.image._finish()
  435:         self.failUnless(self._exists('image.jpg'))
  436:         self.failIf(self._exists('image.jpg.tmp'))
  437: 
  438:     def testFinishCommitsTempPreview(self):
  439:         self.addExtImage(id='image', file=jpegImage)
  440:         self.image.manage_create_prev(maxx=10, maxy=10)
  441:         self.failUnless(self._exists('image.jpg.tmp'))
  442:         self.failUnless(self._exists('image.1.jpg.tmp'))
  443:         self.image._finish()
  444:         self.failUnless(self._exists('image.jpg'))
  445:         self.failIf(self._exists('image.jpg.tmp'))
  446:         self.failUnless(self._exists('image.1.jpg'))
  447:         self.failIf(self._exists('image.1.jpg.tmp'))
  448: 
  449:     def testAbortNukesTempImage(self):
  450:         self.addExtImage(id='image', file=jpegImage)
  451:         self.failUnless(self._exists('image.jpg.tmp'))
  452:         self.image._abort()
  453:         self.failIf(self._exists('image.jpg'))
  454:         self.failIf(self._exists('image.jpg.tmp'))
  455: 
  456:     def testAbortNukesTempPreview(self):
  457:         self.addExtImage(id='image', file=jpegImage)
  458:         self.image.manage_create_prev(maxx=10, maxy=10)
  459:         self.failUnless(self._exists('image.jpg.tmp'))
  460:         self.failUnless(self._exists('image.1.jpg.tmp'))
  461:         self.image._abort()
  462:         self.failIf(self._exists('image.jpg'))
  463:         self.failIf(self._exists('image.jpg.tmp'))
  464:         self.failIf(self._exists('image.1.jpg'))
  465:         self.failIf(self._exists('image.1.jpg.tmp'))
  466: 
  467:     def testUndoCreatesTempImage(self):
  468:         self.addExtImage(id='image', file=jpegImage)
  469:         self.image._finish()
  470:         os.rename(self._fsname('image.jpg'), self._fsname('image.jpg.undo'))
  471:         self.image._undo()
  472:         self.failIf(self._exists('image.jpg.undo'))
  473:         self.failUnless(self._exists('image.jpg.tmp'))
  474: 
  475:     def testUndoCreatesTempPreview(self):
  476:         self.addExtImage(id='image', file=jpegImage)
  477:         self.image.manage_create_prev(maxx=10, maxy=10)
  478:         self.image._finish()
  479:         os.rename(self._fsname('image.jpg'), self._fsname('image.jpg.undo'))
  480:         os.rename(self._fsname('image.1.jpg'), self._fsname('image.1.jpg.undo'))
  481:         self.image._undo()
  482:         self.failIf(self._exists('image.jpg.undo'))
  483:         self.failUnless(self._exists('image.jpg.tmp'))
  484:         self.failIf(self._exists('image.1.jpg.undo'))
  485:         self.failUnless(self._exists('image.1.jpg.tmp'))
  486: 
  487:     def testIsBrokenReturnsTrueIfPreviewBroken(self):
  488:         self.addExtImage(id='image', file=jpegImage)
  489:         self.image.manage_create_prev(maxx=10, maxy=10)
  490:         os.remove(self._fsname('image.1.jpg.tmp'))
  491:         self.failUnless(self.image.is_broken())
  492: 
  493:     def testManageFileUploadRegistersPreviewWithTM(self):
  494:         self.addExtImage(id='image', file='')
  495:         self.assertEqual(getattr(self.image, '_v_begin_called', 0), 0)
  496:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  497:         self.assertEqual(getattr(self.image, '_v_begin_called', 0), 1)
  498: 
  499:     def testManageFileUploadRegistersPreviewWithTMIfResize(self):
  500:         self.addExtImage(id='image', file='')
  501:         self.assertEqual(getattr(self.image, '_v_begin_called', 0), 0)
  502:         self.image.manage_file_upload(file=jpegImage, is_preview=1, create_prev=UPLOAD_RESIZE, maxx=10, maxy=10)
  503:         self.assertEqual(getattr(self.image, '_v_begin_called', 0), 1)
  504: 
  505:     # XXX: Not testable atm
  506:     def DISABLED_testManageFileUploadRegistersPreviewWithTMIfGenerate(self):
  507:         self.addExtImage(id='image', file='')
  508:         self.assertEqual(getattr(self.image, '_v_begin_called', 0), 0)
  509:         self.image.manage_file_upload(file=jpegImage, create_prev=GENERATE, maxx=10, maxy=10)
  510:         self.assertEqual(getattr(self.image, '_v_begin_called', 0), 1)
  511: 
  512:     def testManageHTTPUploadRegistersPreviewWithTM(self):
  513:         self.addExtImage(id='image', file='')
  514:         self.assertEqual(getattr(self.image, '_v_begin_called', 0), 0)
  515:         self.image.manage_http_upload(url=self.app.GifImage.absolute_url(), is_preview=1)
  516:         self.assertEqual(getattr(self.image, '_v_begin_called', 0), 1)
  517: 
  518:     # XXX: Not testable atm
  519:     def DISABLED_testManageHTTPUploadRegistersPreviewWithTMIfAutogen(self):
  520:         self.addExtImage(id='image', file='')
  521:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  522:         self.image._v_begin_called = 0  # Clear
  523:         TM.remove(self.image)           # Clear
  524:         self.assertEqual(getattr(self.image, '_v_begin_called', 0), 0)
  525:         self.image.manage_http_upload(url=self.app.GifImage.absolute_url())
  526:         self.assertEqual(getattr(self.image, '_v_begin_called', 0), 1)
  527: 
  528:     # XXX: Not testable atm
  529:     def DISABLED_testPUTRegistersPreviewWithTMIfAutogen(self):
  530:         self.addExtImage(id='image', file='')
  531:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  532:         self.image._v_begin_called = 0  # Clear
  533:         TM.remove(self.image)           # Clear
  534:         request = self.app.REQUEST
  535:         request['BODYFILE'] = open(jpegImage, 'rb')
  536:         self.assertEqual(getattr(self.image, '_v_begin_called', 0), 0)
  537:         self.image.PUT(request, request.RESPONSE)
  538:         self.assertEqual(getattr(self.image, '_v_begin_called', 0), 1)
  539: 
  540:     def testUndoRegistersPreviewWithTM(self):
  541:         self.addExtImage(id='image', file='')
  542:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  543:         os.rename(self._fsname('image.jpg.tmp'), self._fsname('image.jpg.undo'))
  544:         self.image._v_begin_called = 0  # Clear
  545:         TM.remove(self.image)           # Clear
  546:         self.assertEqual(getattr(self.image, '_v_begin_called', 0), 0)
  547:         self.image._undo()
  548:         self.assertEqual(getattr(self.image, '_v_begin_called', 0), 1)
  549:         self.failUnless(self._exists('image.jpg.tmp'))
  550: 
  551:     def testManageBeforeDeleteUsesTempPreview(self):
  552:         self.addExtImage(id='image', file='')
  553:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  554:         self.failUnless(self._exists('image.jpg.tmp'))
  555:         self.folder._delObject('image')
  556:         self.failUnless(self._exists('image.jpg.undo'))
  557:         self.failIf(self._exists('image.jpg.tmp'))
  558:         self.failIf(self._exists('image.jpg'))
  559: 
  560:     def testManageBeforeDeleteNukesPreviewIfTempPreviewPresent(self):
  561:         self.addExtImage(id='image', file='')
  562:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  563:         self.image._finish()
  564:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  565:         self.failUnless(self._exists('image.jpg'))
  566:         self.failUnless(self._exists('image.jpg.tmp'))
  567:         self.folder._delObject('image')
  568:         self.failUnless(self._exists('image.jpg.undo'))
  569:         self.failIf(self._exists('image.jpg.tmp'))
  570:         self.failIf(self._exists('image.jpg'))
  571: 
  572:     def testManageBeforeDeleteUsesPreviewIfTempPreviewNotPresent(self):
  573:         self.addExtImage(id='image', file='')
  574:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  575:         self.image._finish()
  576:         self.failUnless(self._exists('image.jpg'))
  577:         self.failIf(self._exists('image.jpg.tmp'))
  578:         self.folder._delObject('image')
  579:         self.failUnless(self._exists('image.jpg.undo'))
  580:         self.failIf(self._exists('image.jpg.tmp'))
  581:         self.failIf(self._exists('image.jpg'))
  582: 
  583:     def testManageAfterCloneCreatesTempPreview(self):
  584:         self.addExtImage(id='image', file='')
  585:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  586:         self.image._finish()
  587:         transaction.savepoint(1) # Need a _p_oid
  588:         cb = self.folder.manage_copyObjects(['image'])
  589:         self.subfolder.manage_pasteObjects(cb)
  590:         self.failUnless(self._exists('image.jpg'))      # original
  591:         self.failUnless(self._exists('image.1.jpg.tmp')) # copy
  592:         self.assertEqual(self.subfolder.image.prev_filename, ['image.1.jpg'])
  593: 
  594:     def testManageAfterCloneUsesTempPreviewAsSource(self):
  595:         self.addExtImage(id='image', file='')
  596:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  597:         transaction.savepoint(1) # Need a _p_oid
  598:         self.subfolder.manage_clone(self.image, 'image')
  599:         self.failUnless(self._exists('image.jpg.tmp'))  # original
  600:         self.failUnless(self._exists('image.1.jpg.tmp')) # copy
  601: 
  602:     def testManageAfterCloneUsesPreviewIfTempPreviewNotPresent(self):
  603:         self.addExtImage(id='image', file='')
  604:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  605:         self.image._finish()
  606:         transaction.savepoint(1) # Need a _p_oid
  607:         self.subfolder.manage_clone(self.image, 'image')
  608:         self.failUnless(self._exists('image.jpg'))      # original
  609:         self.failUnless(self._exists('image.1.jpg.tmp')) # copy
  610: 
  611:     def testManageAfterCloneTriesToUndoIfTempPreviewNotPresent(self):
  612:         self.addExtImage(id='image', file='')
  613:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  614:         self.image._finish()
  615:         os.rename(self._fsname('image.jpg'), self._fsname('image.jpg.undo'))
  616:         transaction.savepoint(1) # Need a _p_oid
  617:         self.subfolder.manage_clone(self.image, 'image')
  618:         self.failUnless(self._exists('image.jpg.tmp'))  # restored original
  619:         self.failUnless(self._exists('image.1.jpg.tmp')) # copy
  620: 
  621:     def testManageAfterCloneUsesTempPreviewIfOneFile(self):
  622:         # XXX: Fishy. It seems this tests an impossible state
  623:         self.addExtImage(id='image', file=jpegImage)
  624:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  625:         # Simulate main and preview being one file
  626:         self.image.prev_filename = ['image.2.jpg']
  627:         transaction.savepoint(1) # Need a _p_oid
  628:         self.subfolder.manage_clone(self.image, 'image')
  629:         self.failUnless(self._exists('image.jpg.tmp'))  # original
  630:         self.failUnless(self._exists('image.1.jpg.tmp')) # original preview
  631:         self.failUnless(self._exists('image.2.jpg.tmp')) # copy
  632:         self.assertEqual(self.subfolder.image.prev_filename, ['image.2.jpg'])
  633:         self.assertEqual(self.subfolder.image.filename, ['image.2.jpg'])
  634: 
  635:     # XXX: Not testable atm
  636:     def DISABLED_testManageAfterCloneRegistersWithTM(self):
  637:         pass
  638: 
  639:     def testManageCreatePrevCreatesTempFile(self):
  640:         self.addExtImage(id='image', file=jpegImage)
  641:         self.image.manage_create_prev(maxx=10, maxy=10)
  642:         self.failUnless(self._exists('image.jpg.tmp'))
  643:         self.failUnless(self._exists('image.1.jpg.tmp')) # preview
  644:         self.assertEqual(self.image.prev_filename, ['image.1.jpg'])
  645: 
  646:     def testCreatePrevRegistersWithTM(self):
  647:         self.addExtImage(id='image', file=jpegImage)
  648:         self.assertEqual(getattr(self.image, '_v_begin_called', 0), 1)
  649:         self.image._v_begin_called = 0  # Clear
  650:         TM.remove(self.image)           # Clear
  651:         self.assertEqual(getattr(self.image, '_v_begin_called', 0), 0)
  652:         self.image.manage_create_prev(maxx=10, maxy=10)
  653:         self.assertEqual(getattr(self.image, '_v_begin_called', 0), 1)
  654: 
  655:     def testManageDelPrevUsesTempFile(self):
  656:         self.addExtImage(id='image', file='')
  657:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  658:         self.failUnless(self._exists('image.jpg.tmp'))
  659:         self.image.manage_del_prev()
  660:         self.failUnless(self._exists('image.jpg.undo'))
  661:         self.failIf(self._exists('image.jpg.tmp'))
  662:         self.failIf(self._exists('image.jpg'))
  663: 
  664:     def testManageDelPrevNukesPreviewIfTempPreviewPresent(self):
  665:         self.addExtImage(id='image', file='')
  666:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  667:         self.image._finish()
  668:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  669:         self.failUnless(self._exists('image.jpg'))
  670:         self.failUnless(self._exists('image.jpg.tmp'))
  671:         self.image.manage_del_prev()
  672:         self.failUnless(self._exists('image.jpg.undo'))
  673:         self.failIf(self._exists('image.jpg.tmp'))
  674:         self.failIf(self._exists('image.jpg'))
  675: 
  676:     def testManageDelPrevUsesPreviewIfTempPreviewNotPresent(self):
  677:         self.addExtImage(id='image', file='')
  678:         self.image.manage_file_upload(file=jpegImage, is_preview=1)
  679:         self.image._finish()
  680:         self.failUnless(self._exists('image.jpg'))
  681:         self.failIf(self._exists('image.jpg.tmp'))
  682:         self.image.manage_del_prev()
  683:         self.failUnless(self._exists('image.jpg.undo'))
  684:         self.failIf(self._exists('image.jpg.tmp'))
  685:         self.failIf(self._exists('image.jpg'))
  686: 
  687: 
  688: def test_suite():
  689:     from unittest import TestSuite, makeSuite
  690:     suite = TestSuite()
  691:     suite.addTest(makeSuite(TestTransactions))
  692:     suite.addTest(makeSuite(TestTransactionManager))
  693:     suite.addTest(makeSuite(TestExtFileTransactions))
  694:     suite.addTest(makeSuite(TestExtImageTransactions))
  695:     return suite
  696: 
  697: if __name__ == '__main__':
  698:     framework()
  699: 

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