"""
$RCSfile: TestCacheManager.py,v $
Unit tests for ZopeXMLMethods product

Author: Craeg Strong <cstrong@arielpartners.com>
Modified by Philipp von Weitershausen <philikon@philikon.de>

Release: 1.0
"""

__cvstag__  = '$Name:  $'[6:-2]
__date__    = '$Date: 2005/09/06 02:08:59 $'[6:-2]
__version__ = '$Revision: 1.4 $'[10:-2]

import os, sys, time

if __name__ == '__main__':
    execfile(os.path.join(sys.path[0], 'framework.py'))

# Load fixture
from Testing import ZopeTestCase

from transaction import manager

# ZopeXMLMethods
from Products.ZopeXMLMethods import CacheManager, XSLTMethod

ZopeTestCase.installProduct('ZopeXMLMethods')
ZopeTestCase.installProduct('PageTemplates')

from Testing.ZopeTestCase import folder_name, standard_permissions
host, port = ZopeTestCase.utils.startZServer()
base = 'http://%s:%d/%s' %(host, port, folder_name)

class CacheManagerTestCase(ZopeTestCase.ZopeTestCase):
    """

    Unit tests for the Cache Manager within the ZopeXMLMethod Zope
    Product.  These tests use the ZopeTestCase enhanced testing
    product.

    """

    ################################################################
    # Fixture
    ################################################################

    def beforeSetUp(self):
        manager.begin()

    def afterSetUp(self):
        '''Add object to default fixture'''
        pass

    def beforeClose(self):
        # Commit the cleared app
        manager.commit()

    def afterClear(self):
        '''Clean up after myself'''
        pass

    ################################################################
    # Test Cases
    ################################################################

    def test_01(self):
        "Ensure that cache manager can turn caching on/off in a batch"
        xsltFile     = open('testfiles/simple.xsl','rb')
        self.folder.manage_addDTMLMethod('aTransformer', '', xsltFile)
        
        xmlFile      = open('testfiles/simple.xml', 'rb')
        self.folder.manage_addDTMLMethod('aSource', '', xmlFile)

        xform  = XSLTMethod.addInstance(self.folder,id='aXSLTMethod',
                                        xslTransformerId='aTransformer')
        xform2 = XSLTMethod.addInstance(self.folder,id='aXSLTMethod2',
                                        xslTransformerId='aTransformer')
        xform3 = XSLTMethod.addInstance(self.folder,id='aXSLTMethod3',
                                        xslTransformerId='aTransformer')

        self.failUnless(xform.isCachingOn())
        self.failUnless(xform2.isCachingOn())
        self.failUnless(xform3.isCachingOn())
        
        cm = CacheManager.addInstance(self.folder,'cm')
        cm.batchSetCachingOff(self.app.REQUEST)
        
        self.failIf(xform.isCachingOn())
        self.failIf(xform2.isCachingOn())
        self.failIf(xform3.isCachingOn())
        
        cm.batchSetCachingOn(self.app.REQUEST)
        self.failUnless(xform.isCachingOn())
        self.failUnless(xform2.isCachingOn())
        self.failUnless(xform3.isCachingOn())

        # cleanup
        message = cm.clearCache()

    def test_02(self):
        "Ensure basic caching works"
        xsltFile     = open('testfiles/simple.xsl','rb')
        self.folder.manage_addDTMLMethod('aTransformer', '', xsltFile)
        
        xmlFile      = open('testfiles/simple.xml', 'rb')
        self.folder.manage_addDTMLMethod('aSource', '', xmlFile)
        
        xform = XSLTMethod.addInstance(self.folder,id='aXSLTMethod',
                                         xslTransformerId='aTransformer')
        self.failUnless(xform.isCachingOn())
        
        cm = CacheManager.addInstance(self.folder, 'cm')
        cm.cachePrefix = os.path.join('testfiles', 'Test')

        xform = self.folder.aSource.aXSLTMethod
        transformationPath = os.path.join('aSource', 'aXSLTMethod')

        url = os.path.join(base, transformationPath)
        self.app.REQUEST['URL'] = url
        result = xform(xform,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world")
        
        cacheFileTS = cm.cacheFileTimeStamp(url)
        
        cacheFile = '_'.join( (cm.cachePrefix , folder_name,
                               'aSource', 'aXSLTMethod') )
        self.failUnless(os.path.exists(cacheFile))
        files = cm.listCacheFiles()
        self.failUnless(cacheFile in files)
        
        self.assertEquals(cacheFileTS, os.path.getmtime(cacheFile))

        # cleanup
        message = cm.clearCache()
        self.failIf(os.path.exists(cacheFile))
        
    def test_03(self):
        "Ensure that cache manager can cache multiple xforms in a batch"
        xsltFile     = open('testfiles/simple.xsl','rb')
        self.folder.manage_addDTMLMethod('aTransformer', '', xsltFile)
        
        xmlFile      = open('testfiles/simple.xml', 'rb')
        self.folder.manage_addDTMLMethod('aSource', '', xmlFile)
        
        xform1 = XSLTMethod.addInstance(self.folder,id='aXSLTMethod1',title='',description='',
                                          xslTransformerId='aTransformer')
        xform2 = XSLTMethod.addInstance(self.folder,id='aXSLTMethod2',title='',description='',
                                         xslTransformerId='aTransformer')
        xform3 = XSLTMethod.addInstance(self.folder,id='aXSLTMethod3',title='',description='',
                                         xslTransformerId='aTransformer')
        
        self.failUnless(xform1.isCachingOn())
        self.failUnless(xform2.isCachingOn())
        self.failUnless(xform3.isCachingOn())
        
        cm = CacheManager.addInstance(self.folder,'cm')
        cm.cachePrefix = os.path.join('testfiles', 'Test')

        xform1 = self.folder.aSource.aXSLTMethod1
        transformationPath1 = os.path.join('aSource', 'aXSLTMethod1')

        url = os.path.join(base, transformationPath1)
        self.app.REQUEST['URL'] = url
        result = xform1(xform1,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world")

        xform2 = self.folder.aSource.aXSLTMethod2
        transformationPath2 = os.path.join('aSource', 'aXSLTMethod2')

        url = os.path.join(base, transformationPath2)
        self.app.REQUEST['URL'] = url
        result = xform2(xform2,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world")

        xform3 = self.folder.aSource.aXSLTMethod3
        transformationPath3 = os.path.join('aSource', 'aXSLTMethod3')

        url = os.path.join(base, transformationPath3)
        self.app.REQUEST['URL'] = url
        result = xform3(xform3,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world")

        files = cm.listCacheFiles()
        cacheFile1 = '_'.join( (cm.cachePrefix , folder_name,
                               'aSource', 'aXSLTMethod1') )
        self.failUnless(os.path.exists(cacheFile1))
        self.failUnless(cacheFile1 in files)        
        cacheFile2 = '_'.join( (cm.cachePrefix , folder_name,
                               'aSource', 'aXSLTMethod2') )
        self.failUnless(os.path.exists(cacheFile2))
        self.failUnless(cacheFile2 in files)        
        cacheFile3 = '_'.join( (cm.cachePrefix , folder_name,
                               'aSource', 'aXSLTMethod3') )
        self.failUnless(os.path.exists(cacheFile3))
        self.failUnless(cacheFile3 in files)                

        # cleanup
        message = cm.clearCache()
        self.failIf(os.path.exists(cacheFile1))
        self.failIf(os.path.exists(cacheFile2))
        self.failIf(os.path.exists(cacheFile3))

    def test_04(self):
        "Ensure that cache manager can recover when cache files are deleted by external party"
        xsltFile     = open('testfiles/simple.xsl','rb')
        self.folder.manage_addDTMLMethod('aTransformer', '', xsltFile)
        
        xmlFile      = open('testfiles/simple.xml', 'rb')
        self.folder.manage_addDTMLMethod('aSource', '', xmlFile)
        
        xform1 = XSLTMethod.addInstance(self.folder,id='aXSLTMethod1',
                                          xslTransformerId='aTransformer')
        xform2 = XSLTMethod.addInstance(self.folder,id='aXSLTMethod2',
                                         xslTransformerId='aTransformer')
        xform3 = XSLTMethod.addInstance(self.folder,id='aXSLTMethod3',
                                         xslTransformerId='aTransformer')
        
        self.failUnless(xform1.isCachingOn())
        self.failUnless(xform2.isCachingOn())
        self.failUnless(xform3.isCachingOn())
        
        cm = CacheManager.addInstance(self.folder,'cm')
        cm.cachePrefix = os.path.join('testfiles', 'Test')

        xform1 = self.folder.aSource.aXSLTMethod1
        transformationPath1 = os.path.join('aSource', 'aXSLTMethod1')

        url = os.path.join(base, transformationPath1)
        self.app.REQUEST['URL'] = url
        result = xform1(xform1,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world")

        xform2 = self.folder.aSource.aXSLTMethod2
        transformationPath2 = os.path.join('aSource', 'aXSLTMethod2')

        url = os.path.join(base, transformationPath2)
        self.app.REQUEST['URL'] = url
        result = xform2(xform2,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world")

        xform3 = self.folder.aSource.aXSLTMethod3
        transformationPath3 = os.path.join('aSource', 'aXSLTMethod3')

        url = os.path.join(base, transformationPath3)
        self.app.REQUEST['URL'] = url
        result = xform3(xform3,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world")

        files = cm.listCacheFiles()
        cacheFile1 = '_'.join( (cm.cachePrefix , folder_name,
                                'aSource', 'aXSLTMethod1') )
        self.failUnless(os.path.exists(cacheFile1))
        self.failUnless(cacheFile1 in files)        
        cacheFile2 = '_'.join( (cm.cachePrefix , folder_name,
                                'aSource', 'aXSLTMethod2') )
        self.failUnless(os.path.exists(cacheFile2))
        self.failUnless(cacheFile2 in files)        
        cacheFile3 = '_'.join( (cm.cachePrefix , folder_name,
                                'aSource', 'aXSLTMethod3') )
        self.failUnless(os.path.exists(cacheFile3))
        self.failUnless(cacheFile3 in files)

        os.unlink(cacheFile1)
        # gone from cache, but should not matter
        result = xform1(xform1,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world")

        os.unlink(cacheFile2)
        os.unlink(cacheFile3)
        
        # cleanup
        message = cm.clearCache()
        # message is ignorable
        self.failIf(message.find("not removed, error") == -1)
        self.failIf(os.path.exists(cacheFile1))
        self.failIf(os.path.exists(cacheFile2))
        self.failIf(os.path.exists(cacheFile3))
        
    def test_05(self):
        "Ensure that removing cache manager causes no errors"

        xsltFile     = open('testfiles/simple.xsl','rb')
        self.folder.manage_addDTMLMethod('aTransformer', '', xsltFile)
        
        xmlFile      = open('testfiles/simple.xml', 'rb')
        self.folder.manage_addDTMLMethod('aSource', '', xmlFile)
        
        xform1 = XSLTMethod.addInstance(self.folder,id='aXSLTMethod1',
                                          xslTransformerId='aTransformer')
        xform2 = XSLTMethod.addInstance(self.folder,id='aXSLTMethod2',
                                          xslTransformerId='aTransformer')
        
        self.failUnless(xform1.isCachingOn())
        self.failUnless(xform2.isCachingOn())
        
        cm = CacheManager.addInstance(self.folder,'cm')
        cm.cachePrefix = os.path.join('testfiles', 'Test')

        xform1 = self.folder.aSource.aXSLTMethod1
        transformationPath1 = os.path.join('aSource', 'aXSLTMethod1')

        url1 = os.path.join(base, transformationPath1)
        self.app.REQUEST['URL'] = url1
        result = xform1(xform1,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world")

        xform2 = self.folder.aSource.aXSLTMethod2
        transformationPath2 = os.path.join('aSource', 'aXSLTMethod2')

        url2 = os.path.join(base, transformationPath2)
        self.app.REQUEST['URL'] = url2
        result = xform2(xform2,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world")

        # do it again, for good measure. Should get from the cache now
        self.app.REQUEST['URL'] = url1
        self.assertEquals(xform1(xform1, self.app.REQUEST), "Hello, world")
        self.app.REQUEST['URL'] = url2
        self.assertEquals(xform1(xform2, self.app.REQUEST), "Hello, world")

        self.failUnless(xform1.findCacheManager())
        self.failUnless(xform2.findCacheManager())

        cm.clearCache()

        # remove the cache.  make sure it is truly dead.
        self.folder._delObject('cm')
        del cm
        cm = None

        # should not be a problem, cache manager is optional
        self.app.REQUEST['URL'] = url1
        self.assertEquals(xform1(xform1, self.app.REQUEST), "Hello, world")
        self.app.REQUEST['URL'] = url2
        self.assertEquals(xform1(xform2, self.app.REQUEST), "Hello, world")

        self.failIf(xform1.findCacheManager())
        self.failIf(xform2.findCacheManager())

    def test_06(self):
        "Ensure two objects with same ID, in different folders, can be cached"
        
        xsltFile     = open('testfiles/simple.xsl','rb')
        self.folder.manage_addDTMLMethod('aTransformer', '', xsltFile)
        
        xmlFile      = open('testfiles/simple.xml', 'rb')
        self.folder.manage_addDTMLMethod('aSource', '', xmlFile)
        
        self.folder.manage_addFolder('folder1')
        folder1 = self.folder['folder1']
        xform1 = XSLTMethod.addInstance(folder1,id='aXSLTMethod1',
                                          xslTransformerId='aTransformer')
        
        self.folder.manage_addFolder('folder2')
        folder2 = self.folder['folder2']
        xform2 = XSLTMethod.addInstance(folder2,id='aXSLTMethod2',
                                          xslTransformerId='aTransformer')
        self.failUnless(xform1.isCachingOn())
        self.failUnless(xform2.isCachingOn())
        
        cm = CacheManager.addInstance(self.folder,'cm')
        cm.cachePrefix = os.path.join('testfiles', 'Test')

        transformationPath1 = os.path.join('aSource',
                                           'folder1',
                                           'aXSLTMethod1')
        url1 = os.path.join(base, transformationPath1)
        self.app.REQUEST['URL'] = url1
        xform1  = self.folder.aSource.folder1.aXSLTMethod1
        result1 = xform1(xform1, self.app.REQUEST)

        transformationPath2 = os.path.join('aSource',
                                           'folder2',
                                           'aXSLTMethod2')

        url2 = os.path.join(base, transformationPath2)
        self.app.REQUEST['URL'] = url2
        xform2  = self.folder.aSource.folder2.aXSLTMethod2
        result2 = xform2(xform2, self.app.REQUEST)
        
        self.assertEquals(result1, result2)

        files = cm.listCacheFiles()
        cacheFile1 = '_'.join( (cm.cachePrefix , folder_name,
                                'aSource', 'folder1', 'aXSLTMethod1') )
        self.failUnless(os.path.exists(cacheFile1))
        self.failUnless(cacheFile1 in files)        

        cacheFile2 = '_'.join( (cm.cachePrefix , folder_name,
                                'aSource', 'folder2', 'aXSLTMethod2') )
        self.failUnless(os.path.exists(cacheFile2))
        self.failUnless(cacheFile2 in files)        
        
        # cleanup
        message = cm.clearCache()
        self.failIf(os.path.exists(cacheFile1))
        self.failIf(os.path.exists(cacheFile2))

    def test_07(self):
        "Ensure that changing to a different transformer invalidates the cache"
        xsltFile     = open('testfiles/simple.xsl','rb')
        self.folder.manage_addDTMLMethod('tran1', '', xsltFile)

        xsltFile     = open('testfiles/alternate.xsl','rb')
        self.folder.manage_addDTMLMethod('tran2', '', xsltFile)

        xmlFile      = open('testfiles/simple.xml', 'rb')
        self.folder.manage_addDTMLMethod('aSource', '', xmlFile)
        
        xform = XSLTMethod.addInstance(self.folder,id='aXSLTMethod',
                                         xslTransformerId='tran1')
        self.failUnless(xform.isCachingOn())
        
        cm = CacheManager.addInstance(self.folder,'cm')
        cm.cachePrefix = os.path.join('testfiles', 'Test')

        xform = self.folder.aSource.aXSLTMethod
        transformationPath = os.path.join('aSource', 'aXSLTMethod')
        
        url = os.path.join(base, transformationPath)
        self.app.REQUEST['URL'] = url
        result = xform(xform,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world")
        
        # do it again, for good measure
        result = xform(xform,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world")        

        # First sleep for 4 seconds to ensure we get a different
        # number from bobobase_modification_time()
        time.sleep(4)
        # Change the contents

        # now change the transformer
        xform.editTransform('tran2')
        result = xform(xform,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world and goodbye")

        # cleanup
        message = cm.clearCache()

    def test_08(self):
        "Ensure that changing to a different XML doc invalidates the cache"
        
        xsltFile     = open('testfiles/simple.xsl','rb')
        self.folder.manage_addDTMLMethod('tran', '', xsltFile)

        xmlFile      = open('testfiles/simple.xml', 'rb')
        self.folder.manage_addDTMLMethod('src1', '', xmlFile)

        xmlFile      = open('testfiles/alternate.xml', 'rb')
        self.folder.manage_addDTMLMethod('src2', '', xmlFile)

        xform = XSLTMethod.addInstance(self.folder,id='aXSLTMethod',
                                         xslTransformerId='tran')
        self.failUnless(xform.isCachingOn())
        
        cm = CacheManager.addInstance(self.folder,'cm')
        cm.cachePrefix = os.path.join('testfiles', 'Test')

        xform = self.folder.src1.aXSLTMethod
        transformationPath = os.path.join('src1', 'tran')
        
        url = os.path.join(base, transformationPath)
        self.app.REQUEST['URL'] = url
        result = xform(xform,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world")
        
        # do it again, for good measure
        result = xform(xform,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world")        

        self.assertEquals(xform.isCacheFileUpToDate(self.app.REQUEST), 1)

        # now change the XML document
        xform = self.folder.src2.aXSLTMethod
        transformationPath = os.path.join('src2', 'tran')        

        url = os.path.join(base, transformationPath)
        self.app.REQUEST['URL'] = url
        
        self.assertEquals(xform.isCacheFileUpToDate(self.app.REQUEST), 0)

        result = xform(xform,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, yourself")

        self.assertEquals(xform.isCacheFileUpToDate(self.app.REQUEST), 1)        

        # cleanup
        cm.clearCache()

    def test_09(self):
        "Ensure that editing the source XML doc invalidates the cache"
        xsltFile     = open('testfiles/simple.xsl','rb')
        self.folder.manage_addDTMLMethod('tran', '', xsltFile)

        xmlFile      = open('testfiles/simple.xml', 'rb')
        self.folder.manage_addDTMLMethod('src', '', xmlFile)
        src = self.folder['src']

        xform = XSLTMethod.addInstance(self.folder,id='aXSLTMethod',
                                         xslTransformerId='tran')
        self.failUnless(xform.isCachingOn())
        
        cm = CacheManager.addInstance(self.folder,'cm')
        cm.cachePrefix = os.path.join('testfiles', 'Test')

        xform = self.folder.src.aXSLTMethod
        transformationPath = os.path.join('src', 'tran')
        
        url = os.path.join(base, transformationPath)
        self.app.REQUEST['URL'] = url
        result = xform(xform,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world")
        
        # First sleep for 4 seconds to ensure we get a different
        # number from bobobase_modification_time()
        time.sleep(4)
        # Change the contents
        src.manage_edit("<?xml version=\"1.0\"?><!DOCTYPE document><document><b>Hello, zope</b></document>", "My Title")
        self.assertEquals(xform(xform,self.app.REQUEST), "Hello, zope")

        # cleanup
        cm.clearCache()

    def test_10(self):
        "Ensure that editing the transformer invalidates the cache"
        xsltFile     = open('testfiles/simple.xsl','rb')
        self.folder.manage_addDTMLMethod('tran', '', xsltFile)
        tran = self.folder['tran']
        
        xmlFile      = open('testfiles/simple.xml', 'rb')
        self.folder.manage_addDTMLMethod('src', '', xmlFile)

        xform = XSLTMethod.addInstance(self.folder,id='aXSLTMethod',
                                         xslTransformerId='tran')
        self.failUnless(xform.isCachingOn())
        
        cm = CacheManager.addInstance(self.folder,'cm')
        cm.cachePrefix = os.path.join('testfiles', 'Test')

        xform = self.folder.src.aXSLTMethod
        transformationPath = os.path.join('src', 'tran')
        
        url = os.path.join(base, transformationPath)
        self.app.REQUEST['URL'] = url
        result = xform(xform,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world")

        # First sleep for 4 seconds to ensure we get a different
        # number from bobobase_modification_time()
        time.sleep(4)
        # Change the contents
        xsltFile     = open('testfiles/alternate.xsl','rb')
        
        tran.manage_edit(xsltFile, "My Title")
        result = xform(xform,self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world and goodbye")

        # cleanup
        cm.clearCache()

    def test_11(self):
        "Test XSLT Pipelines with caching"

        cm = CacheManager.addInstance(self.folder,'cm')
        cm.cachePrefix = os.path.join('testfiles', 'Test')

        xsltFile    = open('testfiles/pipeline1.xsl','rb')
        self.folder.manage_addDTMLMethod('xslt1', '', xsltFile)

        xsltFile    = open('testfiles/pipeline2.xsl','rb')
        self.folder.manage_addDTMLMethod('xslt2', '', xsltFile)

        xsltFile    = open('testfiles/pipeline3.xsl','rb')
        self.folder.manage_addDTMLMethod('xslt3', '', xsltFile)

        xmlFile      = open('testfiles/simple.xml', 'rb')
        self.folder.manage_addDTMLMethod('aSource', '', xmlFile)

        XSLTMethod.addInstance(self.folder,id='tran1',
                                 xslTransformerId='xslt1')

        XSLTMethod.addInstance(self.folder,id='tran2',
                                 xslTransformerId='xslt2')

        XSLTMethod.addInstance(self.folder,id='tran3',
                                 xslTransformerId='xslt3')

        text = "<div tal:replace=\"here/aSource/tran1/tran2/tran3\">Oh, crap</div>"

        self.folder.manage_addProduct['PageTemplates'].manage_addPageTemplate('aZPT',
                                                                              '',text)
        aZPT   = self.folder['aZPT']
        result = aZPT(aZPT, self.app.REQUEST)
        self.assertEquals(result.strip(), "Hello, world")

        message = cm.clearCache()

    ################################################################
    # Test Runner Setup
    ################################################################
            
if __name__ == '__main__':
    from Products.ZopeXMLMethods.processors.ProcessorRegistry import ProcessorRegistry
    from Products.ZopeXMLMethods.processors.interfaces import IXSLTProcessor
    
    # Specify which processor to use
    #ProcessorRegistry.CandidateProcessors = ( 'FourSuiteProcessor' )
        
    # report which processor we are using
    print "Using processor:", ProcessorRegistry.defaultName(IXSLTProcessor)

    framework(descriptions=1, verbosity=2)
else:
    # While framework.py provides its own test_suite() 
    # method the testrunner utility does not.
    import unittest
    def test_suite():
        suite = unittest.TestSuite()
        suite.addTest(unittest.makeSuite(CacheManagerTestCase))
        return suite
