diff --git a/plone/app/caching/browser/controlpanel.py b/plone/app/caching/browser/controlpanel.py index 5e4583d..0977731 100644 --- a/plone/app/caching/browser/controlpanel.py +++ b/plone/app/caching/browser/controlpanel.py @@ -56,7 +56,8 @@ def update(self): self.registry = getUtility(IRegistry) self.settings = self.registry.forInterface(ICacheSettings) self.ploneSettings = self.registry.forInterface(IPloneCacheSettings) - self.purgingSettings = self.registry.forInterface(ICachePurgingSettings) + self.purgingSettings = self.registry.forInterface( + ICachePurgingSettings) self.ramCache = queryUtility(IRAMCache) if self.request.method == 'POST': @@ -113,7 +114,8 @@ def publishTraverse(self, request, name): if self.editGlobal: - operation = queryUtility(ICachingOperationType, name=self.editOperationName) + operation = queryUtility( + ICachingOperationType, name=self.editOperationName) if operation is None: raise NotFound(self, operation) @@ -127,7 +129,8 @@ def publishTraverse(self, request, name): if self.editRuleset and self.editOperationName and not self.editRulesetName: self.editRulesetName = name - operation = queryUtility(ICachingOperationType, name=self.editOperationName) + operation = queryUtility( + ICachingOperationType, name=self.editOperationName) if operation is None: raise NotFound(self, self.operationName) @@ -255,7 +258,8 @@ def processSave(self): for domain in domains: if not _isuri(domain): - self.errors['domain'] = _(u"Invalid URL: ${url}", mapping={'url': domain}) + self.errors['domain'] = _( + u"Invalid URL: ${url}", mapping={'url': domain}) # RAM cache settings @@ -265,7 +269,8 @@ def processSave(self): self.errors['ramCacheMaxEntries'] = _(u"An integer is required.") else: if ramCacheMaxEntries < 0: - self.errors['ramCacheMaxEntries'] = _(u"A positive number is required.") + self.errors['ramCacheMaxEntries'] = _( + u"A positive number is required.") try: ramCacheMaxAge = int(ramCacheMaxAge) @@ -273,19 +278,23 @@ def processSave(self): self.errors['ramCacheMaxAge'] = _(u"An integer is required.") else: if ramCacheMaxAge < 0: - self.errors['ramCacheMaxAge'] = _(u"A positive number is required.") + self.errors['ramCacheMaxAge'] = _( + u"A positive number is required.") try: ramCacheCleanupInterval = int(ramCacheCleanupInterval) except (ValueError, TypeError,): - self.errors['ramCacheCleanupInterval'] = _(u"An integer is required.") + self.errors['ramCacheCleanupInterval'] = _( + u"An integer is required.") else: if ramCacheMaxAge < 0: - self.errors['ramCacheCleanupInterval'] = _(u"A positive number is required.") + self.errors['ramCacheCleanupInterval'] = _( + u"A positive number is required.") # Check for errors if self.errors: - IStatusMessage(self.request).addStatusMessage(_(u"There were errors."), "error") + IStatusMessage(self.request).addStatusMessage( + _(u"There were errors."), "error") return # Save settings @@ -301,9 +310,11 @@ def processSave(self): self.purgingSettings.virtualHosting = virtualHosting self.purgingSettings.domains = domains - self.ramCache.update(ramCacheMaxEntries, ramCacheMaxAge, ramCacheCleanupInterval) + self.ramCache.update(ramCacheMaxEntries, + ramCacheMaxAge, ramCacheCleanupInterval) - IStatusMessage(self.request).addStatusMessage(_(u"Changes saved."), "info") + IStatusMessage(self.request).addStatusMessage( + _(u"Changes saved."), "info") # Rule types - used as the index column @property @@ -368,7 +379,8 @@ def contentTypesLookup(self): types = {} portal_types = getToolByName(self.context, 'portal_types') for fti in portal_types.objectValues(): - types[fti.id] = dict(title=fti.title or fti.id, description=fti.description) + types[fti.id] = dict(title=fti.title or fti.id, + description=fti.description) return types # Sorted lists (e.g. for drop-downs) @@ -377,7 +389,8 @@ def contentTypesLookup(self): @memoize def operationTypes(self): operations = [v for k, v in self.operationTypesLookup.items()] - operations.sort(lambda x, y: (cmp(x['sort'], y['sort']) or cmp(x['title'], y['title']))) + operations.sort(lambda x, y: ( + cmp(x['sort'], y['sort']) or cmp(x['title'], y['title']))) return operations @property @@ -463,7 +476,8 @@ def processImport(self): self.errors['profile'] = _(u"You must select a profile to import.") if self.errors: - IStatusMessage(self.request).addStatusMessage(_(u"There were errors."), "error") + IStatusMessage(self.request).addStatusMessage( + _(u"There were errors."), "error") return portal_setup = getToolByName(self.context, 'portal_setup') @@ -477,7 +491,8 @@ def processImport(self): # Import the new profile portal_setup.runAllImportStepsFromProfile("profile-%s" % profile) - IStatusMessage(self.request).addStatusMessage(_(u"Import complete."), "info") + IStatusMessage(self.request).addStatusMessage( + _(u"Import complete."), "info") @property @memoize @@ -490,6 +505,7 @@ def profiles(self): class Purge(BaseView): """The purge control panel """ + def update(self): self.purgeLog = [] @@ -505,7 +521,8 @@ def processPurge(self): self.errors['urls'] = _(u"No URLs or paths entered.") if self.errors: - IStatusMessage(self.request).addStatusMessage(_(u"There were errors."), "error") + IStatusMessage(self.request).addStatusMessage( + _(u"There were errors."), "error") return purger = getUtility(IPurger) @@ -581,11 +598,14 @@ def update(self): def processPurge(self): if self.ramCache is None: - IStatusMessage(self.request).addStatusMessage(_(u"RAM cache not installed."), "error") + IStatusMessage(self.request).addStatusMessage( + _(u"RAM cache not installed."), "error") if self.errors: - IStatusMessage(self.request).addStatusMessage(_(u"There were errors."), "error") + IStatusMessage(self.request).addStatusMessage( + _(u"There were errors."), "error") return self.ramCache.invalidateAll() - IStatusMessage(self.request).addStatusMessage(_(u"Cache purged."), "info") + IStatusMessage(self.request).addStatusMessage( + _(u"Cache purged."), "info") diff --git a/plone/app/caching/browser/edit.py b/plone/app/caching/browser/edit.py index 7c472b3..bf4ec1a 100644 --- a/plone/app/caching/browser/edit.py +++ b/plone/app/caching/browser/edit.py @@ -55,7 +55,8 @@ class EditForm(form.Form): template = ViewPageTemplateFile('edit.pt') - # Keep the ZPublisher happy - would normally be done by the ZCML registration + # Keep the ZPublisher happy - would normally be done by the ZCML + # registration __name__ = 'cache-operation-edit' def __init__(self, context, request, operationName, operation, rulesetName=None, ruleset=None): @@ -70,7 +71,8 @@ def update(self): self.registry = getUtility(IRegistry) - # If we were using plone.z3cform, this would be done with z2.switch_on() + # If we were using plone.z3cform, this would be done with + # z2.switch_on() if not IFormLayer.providedBy(self.request): alsoProvides(self.request, IFormLayer) @@ -88,18 +90,22 @@ def update(self): fieldName = "%s.%s" % (prefix, option) if self.rulesetName: - rulesetFieldName = "%s.%s.%s" % (prefix, self.rulesetName, option) + rulesetFieldName = "%s.%s.%s" % ( + prefix, self.rulesetName, option) if rulesetFieldName in self.registry.records: - newField = self.cloneField(self.registry.records[rulesetFieldName].field) + newField = self.cloneField(self.registry.records[ + rulesetFieldName].field) newField.__name__ = rulesetFieldName elif fieldName in self.registry.records: - newField = self.cloneField(self.registry.records[fieldName].field) + newField = self.cloneField( + self.registry.records[fieldName].field) newField.__name__ = rulesetFieldName else: if fieldName in self.registry.records: - newField = self.cloneField(self.registry.records[fieldName].field) + newField = self.cloneField( + self.registry.records[fieldName].field) newField.__name__ = fieldName if newField is not None: @@ -142,10 +148,12 @@ def getContent(self): # as necessary in applyChanges() if self.rulesetName: - rulesetRecordName = "%s.%s.%s" % (prefix, self.rulesetName, option,) + rulesetRecordName = "%s.%s.%s" % ( + prefix, self.rulesetName, option,) if rulesetRecordName in self.registry.records: - context[rulesetRecordName] = self.registry[rulesetRecordName] + context[rulesetRecordName] = self.registry[ + rulesetRecordName] elif recordName in self.registry.records: context[rulesetRecordName] = self.registry[recordName] @@ -171,12 +179,15 @@ def applyChanges(self, data): # Strip the ruleset name out, leaving the original key - this # must exist, otherwise getContent() would not have put it in # the data dictionary - globalKey = self.operation.prefix + key[len(self.operation.prefix) + len(self.rulesetName) + 1:] + globalKey = self.operation.prefix + \ + key[len(self.operation.prefix) + + len(self.rulesetName) + 1:] assert globalKey in self.registry.records # Create a new record with a FieldRef field = self.registry.records[globalKey].field - self.registry.records[key] = Record(FieldRef(globalKey, field), value) + self.registry.records[key] = Record( + FieldRef(globalKey, field), value) else: self.registry[key] = value @@ -205,11 +216,11 @@ def cloneField(self, field): def title(self): if self.rulesetName: return _(u"Edit ${operation} options for Ruleset: ${ruleset}", - mapping={'operation': self.operation.title, - 'ruleset': self.ruleset.title}) + mapping={'operation': self.operation.title, + 'ruleset': self.ruleset.title}) else: return _(u"Edit ${operation} options", - mapping={'operation': self.operation.title}) + mapping={'operation': self.operation.title}) @property def description(self): @@ -224,23 +235,30 @@ def save(self, action): self.status = self.formErrorsMessage return self.applyChanges(data) - IStatusMessage(self.request).addStatusMessage(_(u"Changes saved."), "info") - self.request.response.redirect("%s/@@caching-controlpanel#detailed-settings" % self.context.absolute_url()) + IStatusMessage(self.request).addStatusMessage( + _(u"Changes saved."), "info") + self.request.response.redirect( + "%s/@@caching-controlpanel#detailed-settings" % self.context.absolute_url()) @button.buttonAndHandler(_(u"Cancel"), name="cancel") def cancel(self, action): - IStatusMessage(self.request).addStatusMessage(_(u"Edit cancelled."), type="info") - self.request.response.redirect("%s/@@caching-controlpanel#detailed-settings" % self.context.absolute_url()) + IStatusMessage(self.request).addStatusMessage( + _(u"Edit cancelled."), type="info") + self.request.response.redirect( + "%s/@@caching-controlpanel#detailed-settings" % self.context.absolute_url()) return '' @button.buttonAndHandler(_(u"Delete settings (use defaults)"), name="clear") def clear(self, action): for key in self.getContent().keys(): - assert key.startswith("%s.%s." % (self.operation.prefix, self.rulesetName,)) + assert key.startswith("%s.%s." % ( + self.operation.prefix, self.rulesetName,)) if key in self.registry.records: del self.registry.records[key] - IStatusMessage(self.request).addStatusMessage(_(u"Ruleset-specific settings removed."), type="info") - self.request.response.redirect("%s/@@caching-controlpanel#detailed-settings" % self.context.absolute_url()) + IStatusMessage(self.request).addStatusMessage( + _(u"Ruleset-specific settings removed."), type="info") + self.request.response.redirect( + "%s/@@caching-controlpanel#detailed-settings" % self.context.absolute_url()) return '' diff --git a/plone/app/caching/interfaces.py b/plone/app/caching/interfaces.py index ea762b4..e6eac09 100644 --- a/plone/app/caching/interfaces.py +++ b/plone/app/caching/interfaces.py @@ -7,6 +7,7 @@ _ = zope.i18nmessageid.MessageFactory('plone.app.caching') + class ICacheProfiles(Interface): """Marker interface for extension profiles that contain cache settings. These will primarily include a ``registry.xml`` file to configure cache @@ -27,6 +28,7 @@ class ICacheProfiles(Interface): available for installation in the cache settings control panel. """ + class IPloneCacheSettings(Interface): """Settings stored in the registry. @@ -36,32 +38,35 @@ class IPloneCacheSettings(Interface): """ templateRulesetMapping = schema.Dict( - title=_(u"Page template/ruleset mapping"), - description=_(u"Maps skin layer page template names to ruleset names"), - key_type=schema.ASCIILine(title=_(u"Page template name")), - value_type=schema.DottedName(title=_(u"Ruleset name")), - ) + title=_(u"Page template/ruleset mapping"), + description=_(u"Maps skin layer page template names to ruleset names"), + key_type=schema.ASCIILine(title=_(u"Page template name")), + value_type=schema.DottedName(title=_(u"Ruleset name")), + ) contentTypeRulesetMapping = schema.Dict( - title=_(u"Content type/ruleset mapping"), - description=_(u"Maps content type names to ruleset names"), - key_type=schema.ASCIILine(title=_(u"Content type name")), - value_type=schema.DottedName(title=_(u"Ruleset name")), - ) + title=_(u"Content type/ruleset mapping"), + description=_(u"Maps content type names to ruleset names"), + key_type=schema.ASCIILine(title=_(u"Content type name")), + value_type=schema.DottedName(title=_(u"Ruleset name")), + ) purgedContentTypes = schema.Tuple( - title=_(u"Content types to purge"), - description=_(u"List content types which should be purged when modified"), - value_type=schema.ASCIILine(title=_(u"Content type name")), - default=('File', 'Image', 'News Item', ), - ) + title=_(u"Content types to purge"), + description=_( + u"List content types which should be purged when modified"), + value_type=schema.ASCIILine(title=_(u"Content type name")), + default=('File', 'Image', 'News Item', ), + ) cacheStopRequestVariables = schema.Tuple( - title=_(u"Request variables that prevent caching"), - description=_(u"Variables in the request that prevent caching if present"), - value_type=schema.ASCIILine(title=_(u"Request variables")), - default=('statusmessages', 'SearchableText',), - ) + title=_(u"Request variables that prevent caching"), + description=_( + u"Variables in the request that prevent caching if present"), + value_type=schema.ASCIILine(title=_(u"Request variables")), + default=('statusmessages', 'SearchableText',), + ) + class IETagValue(Interface): """ETag component builder @@ -76,6 +81,7 @@ def __call__(): """Return the ETag component, as a string. """ + class IRAMCached(Interface): """Marker interface applied to the request if it should be RAM cached. diff --git a/plone/app/caching/lastmodified.py b/plone/app/caching/lastmodified.py index 2ecf1dd..5570e1b 100644 --- a/plone/app/caching/lastmodified.py +++ b/plone/app/caching/lastmodified.py @@ -28,9 +28,6 @@ class IDCTimes(Interface): pass - - - @implementer(ILastModified) @adapter(IPageTemplate) def PageTemplateDelegateLastModified(template): @@ -40,6 +37,7 @@ def PageTemplateDelegateLastModified(template): """ return ILastModified(template.__parent__, None) + @implementer(ILastModified) @adapter(FSPageTemplate) def FSPageTemplateDelegateLastModified(template): @@ -49,6 +47,7 @@ def FSPageTemplateDelegateLastModified(template): """ return PageTemplateDelegateLastModified(template) + @implementer(ILastModified) class PersistentLastModified(object): """General ILastModified adapter for persistent objects that have a @@ -68,11 +67,13 @@ def __call__(self): return datetime.fromtimestamp(mtime, tzlocal()) return None + class OFSFileLastModified(PersistentLastModified): """ILastModified adapter for OFS.Image.File """ adapts(File) + @implementer(ILastModified) class ImageScaleLastModified(object): """ILastModified adapter for Products.Archetypes.Field.Image @@ -88,6 +89,7 @@ def __call__(self): return ILastModified(parent)() return None + @implementer(ILastModified) class FSObjectLastModified(object): """ILastModified adapter for FSFile and FSImage @@ -105,6 +107,7 @@ def __call__(self): mtime = self.context._file_mod_time return datetime.fromtimestamp(mtime, tzlocal()) + @implementer(ILastModified) class CatalogableDublinCoreLastModified(object): """ILastModified adapter for ICatalogableDublinCore, which includes @@ -121,6 +124,7 @@ def __call__(self): return None return modified.asdatetime() + @implementer(ILastModified) class DCTimesLastModified(object): """ILastModified adapter for zope.dublincore IDCTimes @@ -133,6 +137,7 @@ def __init__(self, context): def __call__(self): return self.context.modified + @implementer(ILastModified) class ResourceLastModified(object): """ILastModified for Zope 3 style browser resources @@ -148,6 +153,7 @@ def __call__(self): return datetime.fromtimestamp(lmt, tzlocal()) return None + @implementer(ILastModified) class CookedFileLastModified(object): """ILastModified for Resource Registry `cooked` files diff --git a/plone/app/caching/lookup.py b/plone/app/caching/lookup.py index 4232249..a3a5f5d 100644 --- a/plone/app/caching/lookup.py +++ b/plone/app/caching/lookup.py @@ -65,7 +65,8 @@ def __call__(self): if registry is None: return None - ploneCacheSettings = registry.forInterface(IPloneCacheSettings, check=False) + ploneCacheSettings = registry.forInterface( + IPloneCacheSettings, check=False) # 2. Get the name of the published object name = getattr(self.published, '__name__', None) @@ -89,16 +90,20 @@ def __call__(self): # 4.1.1. Get the default view of the parent content object defaultView = getObjectDefaultView(parent) - # 4.1.2. If the name of the published object is the same as the default view of the parent: + # 4.1.2. If the name of the published object is the same as the + # default view of the parent: if defaultView == name: - # 4.1.2.1. Look up the parent type in the content type mapping + # 4.1.2.1. Look up the parent type in the content type + # mapping if ploneCacheSettings.contentTypeRulesetMapping is not None: - ruleset = ploneCacheSettings.contentTypeRulesetMapping.get(parentPortalType, None) + ruleset = ploneCacheSettings.contentTypeRulesetMapping.get( + parentPortalType, None) if ruleset is not None: return ruleset - # 4.1.2.2. Look up a ruleset on the parent object and return + # 4.1.2.2. Look up a ruleset on the parent object and + # return ruleset = lookup(parent) if ruleset is not None: return ruleset diff --git a/plone/app/caching/operations/default.py b/plone/app/caching/operations/default.py index 3337f5d..2e97c51 100644 --- a/plone/app/caching/operations/default.py +++ b/plone/app/caching/operations/default.py @@ -33,6 +33,7 @@ except ImportError: HAVE_RESOURCE_REGISTRIES = False + @implementer(ICachingOperation) @provider(ICachingOperationType) class BaseCaching(object): @@ -67,7 +68,8 @@ class BaseCaching(object): u"to use one of the other simpler operations (Strong caching, " u"Moderate caching, Weak caching, or No caching).") prefix = 'plone.app.caching.baseCaching' - options = ('maxage','smaxage','etags','lastModified','ramCache', 'vary', 'anonOnly') + options = ('maxage', 'smaxage', 'etags', 'lastModified', + 'ramCache', 'vary', 'anonOnly') # Default option values maxage = smaxage = etags = vary = None @@ -80,9 +82,9 @@ def __init__(self, published, request): def interceptResponse(self, rulename, response, class_=None): options = lookupOptions(class_ or self.__class__, rulename) - etags = options.get('etags') or self.etags - anonOnly = options.get('anonOnly', self.anonOnly) - ramCache = options.get('ramCache', self.ramCache) + etags = options.get('etags') or self.etags + anonOnly = options.get('anonOnly', self.anonOnly) + ramCache = options.get('ramCache', self.ramCache) lastModified = options.get('lastModified', self.lastModified) # Add the ``anonymousOrRandom`` etag if we are anonymous only @@ -93,7 +95,8 @@ def interceptResponse(self, rulename, response, class_=None): etags = tuple(etags) + ('anonymousOrRandom',) etag = getETagAnnotation(self.published, self.request, keys=etags) - lastModified = getLastModifiedAnnotation(self.published, self.request, lastModified=lastModified) + lastModified = getLastModifiedAnnotation( + self.published, self.request, lastModified=lastModified) # Check for cache stop request variables if cacheStop(self.request, rulename): @@ -106,10 +109,12 @@ def interceptResponse(self, rulename, response, class_=None): # Check if this is in the ram cache if ramCache: context = getContext(self.published) - portal_state = getMultiAdapter((context, self.request), name=u'plone_portal_state') + portal_state = getMultiAdapter( + (context, self.request), name=u'plone_portal_state') if portal_state.anonymous(): - cached = fetchFromRAMCache(self.request, etag=etag, lastModified=lastModified) + cached = fetchFromRAMCache( + self.request, etag=etag, lastModified=lastModified) if cached is not None: return cachedResponse(self.published, self.request, response, *cached) @@ -118,13 +123,13 @@ def interceptResponse(self, rulename, response, class_=None): def modifyResponse(self, rulename, response, class_=None): options = lookupOptions(class_ or self.__class__, rulename) - maxage = options.get('maxage', self.maxage) - smaxage = options.get('smaxage', self.smaxage) - etags = options.get('etags') or self.etags + maxage = options.get('maxage', self.maxage) + smaxage = options.get('smaxage', self.smaxage) + etags = options.get('etags') or self.etags anonOnly = options.get('anonOnly', self.anonOnly) ramCache = options.get('ramCache', self.ramCache) - vary = options.get('vary', self.vary) + vary = options.get('vary', self.vary) # Add the ``anonymousOrRandom`` etag if we are anonymous only if anonOnly: @@ -134,7 +139,8 @@ def modifyResponse(self, rulename, response, class_=None): etags = tuple(etags) + ('anonymousOrRandom',) etag = getETagAnnotation(self.published, self.request, etags) - lastModified = getLastModifiedAnnotation(self.published, self.request, options['lastModified']) + lastModified = getLastModifiedAnnotation( + self.published, self.request, options['lastModified']) # Check for cache stop request variables if cacheStop(self.request, rulename): @@ -154,7 +160,8 @@ def modifyResponse(self, rulename, response, class_=None): if etags is not None: if 'userid' in etags or 'anonymousOrRandom' in etags or 'roles' in etags: context = getContext(self.published) - portal_state = getMultiAdapter((context, self.request), name=u'plone_portal_state') + portal_state = getMultiAdapter( + (context, self.request), name=u'plone_portal_state') public = portal_state.anonymous() public = public and visibleToRole(self.published, role='Anonymous') @@ -163,10 +170,11 @@ def modifyResponse(self, rulename, response, class_=None): maxage = smaxage = 0 setCacheHeaders(self.published, self.request, response, maxage=maxage, smaxage=smaxage, - etag=etag, lastModified=lastModified, vary=vary) + etag=etag, lastModified=lastModified, vary=vary) if ramCache and public: - cacheInRAM(self.published, self.request, response, etag=etag, lastModified=lastModified) + cacheInRAM(self.published, self.request, response, + etag=etag, lastModified=lastModified) @provider(ICachingOperationType) @@ -187,13 +195,14 @@ class WeakCaching(BaseCaching): sort = 3 # Configurable options - options = ('etags','lastModified','ramCache','vary', 'anonOnly') + options = ('etags', 'lastModified', 'ramCache', 'vary', 'anonOnly') # Default option values maxage = 0 smaxage = etags = vary = None lastModified = ramCache = anonOnly = False + @provider(ICachingOperationType) class ModerateCaching(BaseCaching): """Moderate caching operation. A subclass of the generic BaseCaching @@ -211,7 +220,8 @@ class ModerateCaching(BaseCaching): sort = 2 # Configurable options - options = ('smaxage','etags','lastModified','ramCache','vary', 'anonOnly') + options = ('smaxage', 'etags', 'lastModified', + 'ramCache', 'vary', 'anonOnly') # Default option values maxage = 0 @@ -219,6 +229,7 @@ class ModerateCaching(BaseCaching): etags = vary = None lastModified = ramCache = anonOnly = False + @provider(ICachingOperationType) class StrongCaching(BaseCaching): """Strong caching operation. A subclass of the generic BaseCaching @@ -234,7 +245,8 @@ class StrongCaching(BaseCaching): sort = 1 # Configurable options - options = ('maxage','smaxage','etags','lastModified','ramCache','vary', 'anonOnly') + options = ('maxage', 'smaxage', 'etags', 'lastModified', + 'ramCache', 'vary', 'anonOnly') # Default option values maxage = 86400 @@ -261,7 +273,9 @@ def modifyResponse(self, rulename, response): doNotCache(self.published, self.request, response) return - super(ResourceRegistriesCaching, self).modifyResponse(rulename, response, class_=StrongCaching) + super(ResourceRegistriesCaching, self).modifyResponse( + rulename, response, class_=StrongCaching) + @implementer(ICachingOperation) @provider(ICachingOperationType) diff --git a/plone/app/caching/operations/etags.py b/plone/app/caching/operations/etags.py index a82b987..cff76e6 100644 --- a/plone/app/caching/operations/etags.py +++ b/plone/app/caching/operations/etags.py @@ -27,7 +27,8 @@ def __init__(self, published, request): def __call__(self): context = getContext(self.published) - portal_state = queryMultiAdapter((context, self.request), name=u'plone_portal_state') + portal_state = queryMultiAdapter( + (context, self.request), name=u'plone_portal_state') if portal_state is None: return None @@ -37,6 +38,7 @@ def __call__(self): return member.getId() + @implementer(IETagValue) class Roles(object): """The ``roles`` etag component, returning the current user's roles, @@ -50,7 +52,8 @@ def __init__(self, published, request): def __call__(self): context = getContext(self.published) - portal_state = queryMultiAdapter((context, self.request), name=u'plone_portal_state') + portal_state = queryMultiAdapter( + (context, self.request), name=u'plone_portal_state') if portal_state is None: return None @@ -63,6 +66,7 @@ def __call__(self): return ';'.join(sorted(member.getRolesInContext(context))) + @implementer(IETagValue) class Language(object): """The ``language`` etag component, returning the value of the @@ -77,6 +81,7 @@ def __init__(self, published, request): def __call__(self): return self.request.get('HTTP_ACCEPT_LANGUAGE', '') + @implementer(IETagValue) class UserLanguage(object): """The ``userLanguage`` etag component, returning the user's preferred @@ -90,12 +95,14 @@ def __init__(self, published, request): def __call__(self): context = getContext(self.published) - portal_state = queryMultiAdapter((context, self.request), name=u'plone_portal_state') + portal_state = queryMultiAdapter( + (context, self.request), name=u'plone_portal_state') if portal_state is None: return None return portal_state.language() + @implementer(IETagValue) class LastModified(object): """The ``lastModified`` etag component, returning the last modification @@ -113,6 +120,7 @@ def __call__(self): return None return str(time.mktime(lastModified.utctimetuple())) + @implementer(IETagValue) class CatalogCounter(object): """The ``catalogCounter`` etag component, returning a counter which is @@ -132,6 +140,7 @@ def __call__(self): return str(tools.catalog().getCounter()) + @implementer(IETagValue) class ObjectLocked(object): """The ``locked`` etag component, returning 1 or 0 depending on whether @@ -145,11 +154,13 @@ def __init__(self, published, request): def __call__(self): context = getContext(self.published) - context_state = queryMultiAdapter((context, self.request), name=u'plone_context_state') + context_state = queryMultiAdapter( + (context, self.request), name=u'plone_context_state') if context_state is None: return None return str(int(context_state.is_locked())) + @implementer(IETagValue) class Skin(object): """The ``skin`` etag component, returning the current skin name. @@ -173,6 +184,7 @@ def __call__(self): return portal_skins.getDefaultSkin() + @implementer(IETagValue) class ResourceRegistries(object): """The ``resourceRegistries`` etag component, returning the most recent @@ -204,6 +216,7 @@ def __call__(self): mtimes.sort() return str(mtimes[-1]) + @implementer(IETagValue) class AnonymousOrRandom(object): """The ``anonymousOrRandom`` etag component. This is normally added @@ -219,13 +232,15 @@ def __init__(self, published, request): def __call__(self): context = getContext(self.published) - portal_state = queryMultiAdapter((context, self.request), name=u'plone_portal_state') + portal_state = queryMultiAdapter( + (context, self.request), name=u'plone_portal_state') if portal_state is None: return None if portal_state.anonymous(): return None return "%s%d" % (time.time(), random.randint(0, 1000)) + @implementer(IETagValue) class CopyCookie(object): """The ``copy`` etag component, returning 1 or 0 depending on whether diff --git a/plone/app/caching/operations/ramcache.py b/plone/app/caching/operations/ramcache.py index 2d7671f..0a82287 100644 --- a/plone/app/caching/operations/ramcache.py +++ b/plone/app/caching/operations/ramcache.py @@ -29,20 +29,20 @@ def __init__(self, published, request): def transformUnicode(self, result, encoding): if self.responseIsSuccess() and IRAMCached.providedBy(self.request): storeResponseInRAMCache(self.request, self.request.response, - result.encode(encoding)) + result.encode(encoding)) return None def transformBytes(self, result, encoding): if self.responseIsSuccess() and IRAMCached.providedBy(self.request): storeResponseInRAMCache(self.request, self.request.response, - result) + result) return None def transformIterable(self, result, encoding): if self.responseIsSuccess() and IRAMCached.providedBy(self.request): result = ''.join(result) storeResponseInRAMCache(self.request, self.request.response, - result) + result) # as we have iterated the iterable, we must return a new one return iter(result) return None diff --git a/plone/app/caching/operations/utils.py b/plone/app/caching/operations/utils.py index 2e51c88..f9ab7b0 100644 --- a/plone/app/caching/operations/utils.py +++ b/plone/app/caching/operations/utils.py @@ -47,6 +47,7 @@ # as any additional keyword parameters required. # + def setCacheHeaders(published, request, response, maxage=None, smaxage=None, etag=None, lastModified=None, vary=None): """General purpose dispatcher to set various cache headers @@ -59,19 +60,20 @@ def setCacheHeaders(published, request, response, maxage=None, smaxage=None, eta if maxage: cacheInBrowserAndProxy(published, request, response, maxage, smaxage=smaxage, - etag=etag, lastModified=lastModified, vary=vary) + etag=etag, lastModified=lastModified, vary=vary) elif smaxage: cacheInProxy(published, request, response, smaxage, - etag=etag, lastModified=lastModified, vary=vary) + etag=etag, lastModified=lastModified, vary=vary) elif etag or lastModified: cacheInBrowser(published, request, response, - etag=etag, lastModified=lastModified) + etag=etag, lastModified=lastModified) else: doNotCache(published, request, response) + def doNotCache(published, request, response): """Set response headers to ensure that the response is not cached by web browsers or caching proxies. @@ -87,6 +89,7 @@ def doNotCache(published, request, response): response.setHeader('Expires', formatDateTime(getExpiration(0))) response.setHeader('Cache-Control', 'max-age=0, must-revalidate, private') + def cacheInBrowser(published, request, response, etag=None, lastModified=None): """Set response headers to indicate that browsers should cache the response but expire immediately and revalidate the cache on every @@ -100,7 +103,7 @@ def cacheInBrowser(published, request, response, etag=None, lastModified=None): """ if etag is not None: - response.setHeader('ETag', '"%s"' %etag, literal=1) + response.setHeader('ETag', '"%s"' % etag, literal=1) if lastModified is not None: response.setHeader('Last-Modified', formatDateTime(lastModified)) @@ -110,6 +113,7 @@ def cacheInBrowser(published, request, response, etag=None, lastModified=None): response.setHeader('Expires', formatDateTime(getExpiration(0))) response.setHeader('Cache-Control', 'max-age=0, must-revalidate, private') + def cacheInProxy(published, request, response, smaxage, etag=None, lastModified=None, vary=None): """Set headers to cache the response in a caching proxy. @@ -125,13 +129,15 @@ def cacheInProxy(published, request, response, smaxage, etag=None, lastModified= del response.headers['last-modified'] if etag is not None: - response.setHeader('ETag', '"%s"' %etag, literal=1) + response.setHeader('ETag', '"%s"' % etag, literal=1) if vary is not None: response.setHeader('Vary', vary) response.setHeader('Expires', formatDateTime(getExpiration(0))) - response.setHeader('Cache-Control', 'max-age=0, s-maxage=%d, must-revalidate' % smaxage) + response.setHeader( + 'Cache-Control', 'max-age=0, s-maxage=%d, must-revalidate' % smaxage) + def cacheInBrowserAndProxy(published, request, response, maxage, smaxage=None, etag=None, lastModified=None, vary=None): """Set headers to cache the response in the browser and caching proxy if @@ -150,7 +156,7 @@ def cacheInBrowserAndProxy(published, request, response, maxage, smaxage=None, e del response.headers['last-modified'] if etag is not None: - response.setHeader('ETag', '"%s"' %etag, literal=1) + response.setHeader('ETag', '"%s"' % etag, literal=1) if vary is not None: response.setHeader('Vary', vary) @@ -158,11 +164,13 @@ def cacheInBrowserAndProxy(published, request, response, maxage, smaxage=None, e response.setHeader('Expires', formatDateTime(getExpiration(maxage))) if smaxage is not None: - maxage = '%s, s-maxage=%s' %(maxage, smaxage) + maxage = '%s, s-maxage=%s' % (maxage, smaxage) # Substituting proxy-validate in place of must=revalidate here because of Safari bug # https://bugs.webkit.org/show_bug.cgi?id=13128 - response.setHeader('Cache-Control', 'max-age=%s, proxy-revalidate, public' % maxage) + response.setHeader( + 'Cache-Control', 'max-age=%s, proxy-revalidate, public' % maxage) + def cacheInRAM(published, request, response, etag=None, lastModified=None, annotationsKey=PAGE_CACHE_ANNOTATION_KEY): """Set a flag indicating that the response for the given request @@ -192,6 +200,7 @@ def cacheInRAM(published, request, response, etag=None, lastModified=None, annot annotations[annotationsKey] = key alsoProvides(request, IRAMCached) + def cachedResponse(published, request, response, status, headers, body, gzip=False): """Returned a cached page. Modifies the response (status and headers) and returns the cached body. @@ -219,6 +228,7 @@ def cachedResponse(published, request, response, status, headers, body, gzip=Fal return body + def notModified(published, request, response, etag=None, lastModified=None): """Return a ``304 NOT MODIFIED`` response. Modifies the response (status) and returns an empty body to indicate the request should be interrupted. @@ -283,6 +293,7 @@ def cacheStop(request, rulename): return True return False + def isModified(request, etag=None, lastModified=None): """Return True or False depending on whether the published resource has been modified. @@ -327,7 +338,7 @@ def isModified(request, etag=None, lastModified=None): latest headers. I interpret this as the spec rule that the etags do NOT match, and therefor we must not return a 304. """ - if ifNoneMatch and etag==None: + if ifNoneMatch and etag == None: return True # Check the modification date @@ -372,6 +383,7 @@ def visibleToRole(published, role, permission='View'): # Basic helper functions # + def getContext(published, marker=(IContentish, ISiteRoot,)): """Given a published object, attempt to look up a context @@ -402,6 +414,7 @@ def checkType(context): return published + def formatDateTime(dt): """Format a Python datetime object as an RFC1123 date. @@ -415,6 +428,7 @@ def formatDateTime(dt): return wsgiref.handlers.format_date_time(time.mktime(dt.timetuple())) + def parseDateTime(str): """Return a Python datetime object from an an RFC1123 date. @@ -437,6 +451,7 @@ def parseDateTime(str): return dt + def getLastModifiedAnnotation(published, request, lastModified=True): """Try to get the last modified date from a request annotation if available, otherwise try to get it from published object @@ -458,6 +473,7 @@ def getLastModifiedAnnotation(published, request, lastModified=True): return dt + def getLastModified(published, lastModified=True): """Get a last modified date or None. @@ -483,6 +499,7 @@ def getLastModified(published, lastModified=True): return dt + def getExpiration(maxage): """Get an expiration date as a datetime in the local timezone. @@ -496,6 +513,7 @@ def getExpiration(maxage): else: return now - datetime.timedelta(days=3650) + def getETagAnnotation(published, request, keys=(), extraTokens=()): """Try to get the ETag from a request annotation if available, otherwise try to get it from published object @@ -517,6 +535,7 @@ def getETagAnnotation(published, request, keys=(), extraTokens=()): return etag + def getETag(published, request, keys=(), extraTokens=()): """Calculate an ETag. @@ -536,9 +555,11 @@ def getETag(published, request, keys=(), extraTokens=()): tokens = [] noTokens = True for key in keys: - component = queryMultiAdapter((published, request), IETagValue, name=key) + component = queryMultiAdapter( + (published, request), IETagValue, name=key) if component is None: - logger.warning("Could not find value adapter for ETag component %s", key) + logger.warning( + "Could not find value adapter for ETag component %s", key) tokens.append('') else: value = component() @@ -561,6 +582,7 @@ def getETag(published, request, keys=(), extraTokens=()): return etag + def parseETags(text, allowWeak=True, _result=None): """Parse a header value into a list of etags. Handles fishy quoting and other browser quirks. @@ -586,13 +608,13 @@ def parseETags(text, allowWeak=True, _result=None): m = etagQuote.match(text) if m: # Match quoted etag (spec-observing client) - l = len(m.group(1)) + l = len(m.group(1)) value = (m.group(2) or '') + (m.group(3) or '') else: # Match non-quoted etag (lazy client) m = etagNoQuote.match(text) if m: - l = len(m.group(1)) + l = len(m.group(1)) value = (m.group(2) or '') + (m.group(3) or '') else: return result @@ -613,6 +635,7 @@ def parseETags(text, allowWeak=True, _result=None): # RAM cache management # + def getRAMCache(globalKey=PAGE_CACHE_KEY): """Get a RAM cache instance for the given key. The return value is ``None`` if no RAM cache can be found, or a mapping object supporting at least @@ -629,6 +652,7 @@ def getRAMCache(globalKey=PAGE_CACHE_KEY): return chooser(globalKey) + def getRAMCacheKey(request, etag=None, lastModified=None): """Calculate the cache key for pages cached in RAM. @@ -647,13 +671,14 @@ def getRAMCacheKey(request, etag=None, lastModified=None): request.get('SERVER_URL', ''), request.get('PATH_INFO', ''), request.get('QUERY_STRING', ''), - ) + ) if etag: resourceKey = '|' + etag + '||' + resourceKey if lastModified: resourceKey = '|' + str(lastModified) + '||' + resourceKey return resourceKey + def storeResponseInRAMCache(request, response, result, globalKey=PAGE_CACHE_KEY, annotationsKey=PAGE_CACHE_ANNOTATION_KEY): """Store the given response in the RAM cache. @@ -698,6 +723,7 @@ def storeResponseInRAMCache(request, response, result, globalKey=PAGE_CACHE_KEY, cache[key] = (status, headers, result, gzipFlag) + def fetchFromRAMCache(request, etag=None, lastModified=None, globalKey=PAGE_CACHE_KEY, default=None): """Return a page cached in RAM, or None if it cannot be found. diff --git a/plone/app/caching/purge.py b/plone/app/caching/purge.py index 80a50fc..3221e92 100644 --- a/plone/app/caching/purge.py +++ b/plone/app/caching/purge.py @@ -28,6 +28,7 @@ except ImportError: HAVE_AT = False + @implementer(IPurgePaths) class ContentPurgePaths(object): """Paths to purge for content items @@ -84,6 +85,7 @@ def getRelativePaths(self): def getAbsolutePaths(self): return [] + @implementer(IPurgePaths) class DiscussionItemPurgePaths(object): """Paths to purge for Discussion Item. @@ -113,7 +115,8 @@ def getRelativePaths(self): if rewriter is None: yield relativePath else: - rewrittenPaths = rewriter(relativePath) or [] # None -> [] + rewrittenPaths = rewriter( + relativePath) or [] # None -> [] for rewrittenPath in rewrittenPaths: yield rewrittenPath @@ -163,8 +166,8 @@ def getRelativePaths(self): def fieldFilter(field): return ((IBlobField.providedBy(field) or - IFileField.providedBy(field) or - IImageField.providedBy(field)) + IFileField.providedBy(field) or + IImageField.providedBy(field)) and not ITextField.providedBy(field)) seenDownloads = False @@ -191,11 +194,13 @@ def getAbsolutePaths(self): # Event redispatch for content items - we check the list of content items # instead of the marker interface + @adapter(IContentish, IObjectModifiedEvent) def purgeOnModified(object, event): if isPurged(object): notify(Purge(object)) + @adapter(IContentish, IObjectMovedEvent) def purgeOnMovedOrRemoved(object, event): # Don't purge when added diff --git a/plone/app/caching/setuphandlers.py b/plone/app/caching/setuphandlers.py index 9d2f5c1..5dd25d7 100644 --- a/plone/app/caching/setuphandlers.py +++ b/plone/app/caching/setuphandlers.py @@ -11,6 +11,7 @@ def enableExplicitMode(): if registry is not None: registry.explicit = True + def importVarious(context): if not context.readDataFile('plone.app.caching.txt'): diff --git a/plone/app/caching/testing.py b/plone/app/caching/testing.py index 623631e..765a58d 100644 --- a/plone/app/caching/testing.py +++ b/plone/app/caching/testing.py @@ -19,12 +19,12 @@ except ImportError: # so we can run tests on plone 4.3 from plone.keyring.interfaces import IKeyManager + def _getKeyring(username): manager = getUtility(IKeyManager) return manager['_system'] - @implementer(IPurger) class FauxPurger(object): @@ -56,7 +56,8 @@ def setUpZope(self, app, configurationContext): # Load ZCML import plone.app.caching - xmlconfig.file('configure.zcml', plone.app.caching, context=configurationContext) + xmlconfig.file('configure.zcml', plone.app.caching, + context=configurationContext) # Install fake purger self.oldPurger = getUtility(IPurger) @@ -65,7 +66,8 @@ def setUpZope(self, app, configurationContext): def setUpPloneSite(self, portal): applyProfile(portal, 'plone.app.caching:default') - portal['portal_workflow'].setDefaultChain('simple_publication_workflow') + portal['portal_workflow'].setDefaultChain( + 'simple_publication_workflow') def tearDownZope(self, app): # Store old purger diff --git a/plone/app/caching/tests/test_etags.py b/plone/app/caching/tests/test_etags.py index 504d24d..66b75f7 100644 --- a/plone/app/caching/tests/test_etags.py +++ b/plone/app/caching/tests/test_etags.py @@ -27,11 +27,13 @@ class DummyContext(Explicit): pass + class DummyPublished(object): def __init__(self, parent=None): self.__parent__ = parent + class TestETags(unittest.TestCase): layer = UNIT_TESTING @@ -94,7 +96,6 @@ def member(self): self.assertEqual('bob', etag()) - # Roles def test_Roles_anonymous(self): @@ -156,7 +157,6 @@ def member(self): self.assertEqual('Manager;Member', etag()) - # Language def test_Language_no_header(self): @@ -306,7 +306,6 @@ def catalog(self): self.assertEqual('10', etag()) - # ObjectLocked def test_ObjectLocked_true(self): @@ -357,7 +356,6 @@ def is_locked(self): self.assertEqual('0', etag()) - # Skin def test_Skin_request_variable(self): diff --git a/plone/app/caching/tests/test_integration.py b/plone/app/caching/tests/test_integration.py index ab852e9..77cc985 100644 --- a/plone/app/caching/tests/test_integration.py +++ b/plone/app/caching/tests/test_integration.py @@ -197,7 +197,8 @@ def test_auto_purge_content_types(self): browser.open(editURL) - browser.getControl(name='form.widgets.IDublinCore.title').value = u"Title 1" + browser.getControl( + name='form.widgets.IDublinCore.title').value = u"Title 1" browser.getControl('Save').click() self.assertEqual([], self.purger._sync) @@ -212,7 +213,8 @@ def test_auto_purge_content_types(self): transaction.commit() browser.open(editURL) - browser.getControl(name='form.widgets.IDublinCore.title').value = u"Title 2" + browser.getControl( + name='form.widgets.IDublinCore.title').value = u"Title 2" browser.getControl('Save').click() self.assertEqual([], self.purger._sync) @@ -227,7 +229,8 @@ def test_auto_purge_content_types(self): transaction.commit() browser.open(editURL) - browser.getControl(name='form.widgets.IDublinCore.title').value = u"Title 3" + browser.getControl( + name='form.widgets.IDublinCore.title').value = u"Title 3" browser.getControl('Save').click() self.assertEqual([], self.purger._sync) @@ -242,7 +245,8 @@ def test_auto_purge_content_types(self): transaction.commit() browser.open(editURL) - browser.getControl(name='form.widgets.IDublinCore.title').value = u"Title 4" + browser.getControl( + name='form.widgets.IDublinCore.title').value = u"Title 4" browser.getControl('Save').click() self.assertEqual([], self.purger._sync) diff --git a/plone/app/caching/tests/test_lastmodified.py b/plone/app/caching/tests/test_lastmodified.py index 028ab5f..975566a 100644 --- a/plone/app/caching/tests/test_lastmodified.py +++ b/plone/app/caching/tests/test_lastmodified.py @@ -24,6 +24,7 @@ def oldstate(self, obj, tid): def register(self, object): pass + class TestLastModified(unittest.TestCase): layer = UNIT_TESTING @@ -52,8 +53,7 @@ class Dummy(Persistent, Explicit): zpt = ZopePageTemplate('zpt').__of__(d) self.assertEqual(None, ILastModified(zpt)()) - - timestamp = 987654321.0 # time stamp (in UTC) + timestamp = 987654321.0 # time stamp (in UTC) # equivalent in local time, which is what the last-modified adapter # should return mod = datetime.datetime.fromtimestamp(timestamp, tzlocal()) @@ -76,8 +76,7 @@ class Dummy(Persistent, Explicit): zpt = FSPageTemplate('zpt', __file__).__of__(d) self.assertEqual(None, ILastModified(zpt)()) - - timestamp = 987654321.0 # time stamp (in UTC) + timestamp = 987654321.0 # time stamp (in UTC) # equivalent in local time, which is what the last-modified adapter # should return mod = datetime.datetime.fromtimestamp(timestamp, tzlocal()) @@ -91,8 +90,8 @@ def test_OFSFileLastModified_File(self): dummy = File('dummy', 'Dummy', 'data') self.assertEqual(None, ILastModified(dummy)()) - timestamp = 987654321.0 # time stamp (in UTC) - ts = TimeStamp(*time.gmtime(timestamp)[:6]) # corresponding TimeStamp + timestamp = 987654321.0 # time stamp (in UTC) + ts = TimeStamp(*time.gmtime(timestamp)[:6]) # corresponding TimeStamp # equivalent in local time, which is what the last-modified adapter # should return @@ -108,8 +107,8 @@ def test_OFSFileLastModified_Image(self): dummy = Image('dummy', 'Dummy', 'data') self.assertEqual(None, ILastModified(dummy)()) - timestamp = 987654321.0 # time stamp (in UTC) - ts = TimeStamp(*time.gmtime(timestamp)[:6]) # corresponding TimeStamp + timestamp = 987654321.0 # time stamp (in UTC) + ts = TimeStamp(*time.gmtime(timestamp)[:6]) # corresponding TimeStamp # equivalent in local time, which is what the last-modified adapter # should return @@ -127,19 +126,22 @@ def test_FSObjectLastModified_FSFile(self): modtime = float(os.path.getmtime(__file__)) mod = datetime.datetime.fromtimestamp(modtime, tzlocal()) - format = "%y%m%d%H%M%s" # see note in test_FSObjectLastModified_FSImage - self.assertEqual(mod.strftime(format), ILastModified(dummy)().strftime(format)) + format = "%y%m%d%H%M%s" # see note in test_FSObjectLastModified_FSImage + self.assertEqual(mod.strftime(format), + ILastModified(dummy)().strftime(format)) def test_FSObjectLastModified_FSImage(self): from Products.CMFCore.FSImage import FSImage - dummy = FSImage('dummy', __file__) # not really an image, but anyway + dummy = FSImage('dummy', __file__) # not really an image, but anyway modtime = float(os.path.getmtime(__file__)) mod = datetime.datetime.fromtimestamp(modtime, tzlocal()) # different filesystems seem to handle datetime differently. some use microseconds - # and others don't so to make jenkins happy lets omit the microseconds factor + # and others don't so to make jenkins happy lets omit the microseconds + # factor format = "%y%m%d%H%M%s" - self.assertEqual(mod.strftime(format), ILastModified(dummy)().strftime(format)) + self.assertEqual(mod.strftime(format), + ILastModified(dummy)().strftime(format)) def test_CatalogableDublinCoreLastModified(self): from Products.CMFCore.interfaces import ICatalogableDublinCore @@ -151,7 +153,8 @@ class Dummy(object): _mod = None def modified(self): - if self._mod is not None: return DateTime.DateTime(self._mod) + if self._mod is not None: + return DateTime.DateTime(self._mod) return None d = Dummy() diff --git a/plone/app/caching/tests/test_lookup.py b/plone/app/caching/tests/test_lookup.py index d57f954..13bd411 100644 --- a/plone/app/caching/tests/test_lookup.py +++ b/plone/app/caching/tests/test_lookup.py @@ -30,16 +30,18 @@ def __init__(self, portal_type='testtype', defaultView='defaultView'): def defaultView(self): return self._defaultView + class DummyNotContent(Explicit): pass + class DummyFTI(object): def __init__(self, portal_type, viewAction=''): self.id = portal_type self._actions = { - 'object/view': {'url': viewAction}, - } + 'object/view': {'url': viewAction}, + } def getActionInfo(self, name): return self._actions[name] @@ -51,6 +53,7 @@ def queryMethodID(self, id, default=None, context=None): return '@@defaultView' return default + @implementer(IDynamicType) class DummyNotBrowserDefault(Explicit): @@ -61,19 +64,24 @@ def __init__(self, portal_type='testtype', viewAction=''): def getTypeInfo(self): return DummyFTI(self.portal_type, self._viewAction) + class DummyResponse(dict): def addHeader(self, name, value): self.setdefault(name, []).append(value) + class DummyRequest(dict): + def __init__(self, published, response): self['PUBLISHED'] = published self.response = response + class DummyView(BrowserView): __name__ = 'defaultView' + class TestContentItemLookup(unittest.TestCase): layer = IMPLICIT_RULESET_REGISTRY_UNIT_TESTING @@ -218,7 +226,8 @@ def test_parent_not_IBrowserDefault_methodid(self): ploneSettings.templateRulesetMapping = {} ploneSettings.contentTypeRulesetMapping = {'testtype': 'rule1'} - published = ZopePageTemplate('defaultView').__of__(DummyNotBrowserDefault('testtype', 'string:${object_url}/view')) + published = ZopePageTemplate('defaultView').__of__( + DummyNotBrowserDefault('testtype', 'string:${object_url}/view')) request = DummyRequest(published, DummyResponse()) self.assertEqual('rule1', ContentItemLookup(published, request)()) @@ -231,7 +240,8 @@ def test_parent_not_IBrowserDefault_default_method(self): ploneSettings.templateRulesetMapping = {} ploneSettings.contentTypeRulesetMapping = {'testtype': 'rule1'} - published = ZopePageTemplate('defaultView').__of__(DummyNotBrowserDefault('testtype', 'string:${object_url}/')) + published = ZopePageTemplate('defaultView').__of__( + DummyNotBrowserDefault('testtype', 'string:${object_url}/')) request = DummyRequest(published, DummyResponse()) self.assertEqual('rule1', ContentItemLookup(published, request)()) @@ -244,7 +254,8 @@ def test_parent_not_IBrowserDefault_actiononly(self): ploneSettings.templateRulesetMapping = {} ploneSettings.contentTypeRulesetMapping = {'testtype': 'rule1'} - published = ZopePageTemplate('defaultView').__of__(DummyNotBrowserDefault('testtype', 'string:${object_url}/defaultView')) + published = ZopePageTemplate('defaultView').__of__( + DummyNotBrowserDefault('testtype', 'string:${object_url}/defaultView')) request = DummyRequest(published, DummyResponse()) self.assertEqual('rule1', ContentItemLookup(published, request)()) diff --git a/plone/app/caching/tests/test_operation_default.py b/plone/app/caching/tests/test_operation_default.py index 8c79997..f2fe0ca 100644 --- a/plone/app/caching/tests/test_operation_default.py +++ b/plone/app/caching/tests/test_operation_default.py @@ -66,7 +66,8 @@ def test_last_modified_no_etags(self): self.assertFalse('Etag' in browser.headers) # now set up etags and make sure that a header is added - self.registry['plone.app.caching.weakCaching.etags'] = ('lastModified',) + self.registry['plone.app.caching.weakCaching.etags'] = ( + 'lastModified',) import transaction transaction.commit() browser.open("%s/dummy-content/edit?_authenticator=%s" % ( diff --git a/plone/app/caching/tests/test_operation_parameters.py b/plone/app/caching/tests/test_operation_parameters.py index 2bf0b2c..c49f2e2 100644 --- a/plone/app/caching/tests/test_operation_parameters.py +++ b/plone/app/caching/tests/test_operation_parameters.py @@ -58,25 +58,31 @@ def test_anon_only(self): # Publish the folder and page self.portal.portal_workflow.doActionFor(self.portal['f1'], 'publish') - self.portal.portal_workflow.doActionFor(self.portal['f1']['d1'], 'publish') + self.portal.portal_workflow.doActionFor( + self.portal['f1']['d1'], 'publish') # Set pages to have weak caching and test anonymous - self.cacheSettings.operationMapping = {'plone.content.itemView': 'plone.app.caching.weakCaching'} + self.cacheSettings.operationMapping = { + 'plone.content.itemView': 'plone.app.caching.weakCaching'} transaction.commit() # View the page as anonymous browser = Browser(self.app) browser.open(self.portal['f1']['d1'].absolute_url()) - self.assertEqual('plone.content.itemView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.itemView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) self.assertTrue(testText in browser.contents) - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) # Set pages to have moderate caching so that we can see the difference # between logged in and anonymous - self.cacheSettings.operationMapping = {'plone.content.itemView': 'plone.app.caching.moderateCaching'} + self.cacheSettings.operationMapping = { + 'plone.content.itemView': 'plone.app.caching.moderateCaching'} self.registry['plone.app.caching.moderateCaching.smaxage'] = 60 self.registry['plone.app.caching.moderateCaching.vary'] = 'X-Anonymous' self.registry['plone.app.caching.moderateCaching.anonOnly'] = True @@ -86,27 +92,35 @@ def test_anon_only(self): # View the page as anonymous browser = Browser(self.app) browser.open(self.portal['f1']['d1'].absolute_url()) - self.assertEqual('plone.content.itemView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.moderateCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.itemView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.moderateCaching', + browser.headers['X-Cache-Operation']) self.assertTrue(testText in browser.contents) - self.assertEqual('max-age=0, s-maxage=60, must-revalidate', browser.headers['Cache-Control']) + self.assertEqual('max-age=0, s-maxage=60, must-revalidate', + browser.headers['Cache-Control']) self.assertEqual('X-Anonymous', browser.headers['Vary']) self.assertFalse('Etag' in browser.headers) # View the page as logged-in browser = Browser(self.app) - browser.addHeader('Authorization', 'Basic %s:%s' % (TEST_USER_NAME, TEST_USER_PASSWORD,)) + browser.addHeader('Authorization', 'Basic %s:%s' % + (TEST_USER_NAME, TEST_USER_PASSWORD,)) browser.open(self.portal['f1']['d1'].absolute_url()) - self.assertEqual('plone.content.itemView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.moderateCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.itemView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.moderateCaching', + browser.headers['X-Cache-Operation']) self.assertTrue(testText in browser.contents) - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) self.assertTrue('Etag' in browser.headers) # Set pages to have strong caching so that we can see the difference # between logged in and anonymous - self.cacheSettings.operationMapping = {'plone.content.itemView': 'plone.app.caching.strongCaching'} + self.cacheSettings.operationMapping = { + 'plone.content.itemView': 'plone.app.caching.strongCaching'} self.registry['plone.app.caching.strongCaching.vary'] = 'X-Anonymous' self.registry['plone.app.caching.strongCaching.anonOnly'] = True transaction.commit() @@ -114,21 +128,28 @@ def test_anon_only(self): # View the page as anonymous browser = Browser(self.app) browser.open(self.portal['f1']['d1'].absolute_url()) - self.assertEqual('plone.content.itemView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.strongCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.itemView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.strongCaching', + browser.headers['X-Cache-Operation']) self.assertTrue(testText in browser.contents) - self.assertEqual('max-age=86400, proxy-revalidate, public', browser.headers['Cache-Control']) + self.assertEqual('max-age=86400, proxy-revalidate, public', + browser.headers['Cache-Control']) self.assertEqual('X-Anonymous', browser.headers['Vary']) self.assertFalse('Etag' in browser.headers) # View the page as logged-in browser = Browser(self.app) - browser.addHeader('Authorization', 'Basic %s:%s' % (TEST_USER_NAME, TEST_USER_PASSWORD,)) + browser.addHeader('Authorization', 'Basic %s:%s' % + (TEST_USER_NAME, TEST_USER_PASSWORD,)) browser.open(self.portal['f1']['d1'].absolute_url()) - self.assertEqual('plone.content.itemView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.strongCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.itemView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.strongCaching', + browser.headers['X-Cache-Operation']) self.assertTrue(testText in browser.contents) - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) self.assertTrue('Etag' in browser.headers) # Check an edge case that has had a problem in the past: @@ -141,4 +162,5 @@ def test_anon_only(self): # View the page as anonymous browser = Browser(self.app) browser.open(self.portal['f1']['d1'].absolute_url()) - self.assertEqual('max-age=0, s-maxage=60, must-revalidate', browser.headers['Cache-Control']) + self.assertEqual('max-age=0, s-maxage=60, must-revalidate', + browser.headers['Cache-Control']) diff --git a/plone/app/caching/tests/test_operation_utils.py b/plone/app/caching/tests/test_operation_utils.py index a5d90d6..8fba6c3 100644 --- a/plone/app/caching/tests/test_operation_utils.py +++ b/plone/app/caching/tests/test_operation_utils.py @@ -31,6 +31,7 @@ class DummyPublished(object): def __init__(self, parent=None): self.__parent__ = parent + def normalize_response_cache(value): # Zope < 2.13 incorrectly includes the HTTP status as a header; # Zope 2.13 does not @@ -62,7 +63,8 @@ def test_doNotCache(self): doNotCache(published, request, response) self.assertEqual(200, response.getStatus()) - self.assertEqual('max-age=0, must-revalidate, private', response.getHeader('Cache-Control')) + self.assertEqual('max-age=0, must-revalidate, private', + response.getHeader('Cache-Control')) self.assertEqual(None, response.getHeader('Last-Modified')) expires = dateutil.parser.parse(response.getHeader('Expires')) @@ -77,18 +79,19 @@ def test_doNotCache_deletes_last_modified(self): published = DummyPublished() now = datetime.datetime.now(dateutil.tz.tzlocal()) - response.setHeader('Last-Modified', wsgiref.handlers.format_date_time(time.mktime(now.timetuple()))) + response.setHeader( + 'Last-Modified', wsgiref.handlers.format_date_time(time.mktime(now.timetuple()))) doNotCache(published, request, response) self.assertEqual(200, response.getStatus()) - self.assertEqual('max-age=0, must-revalidate, private', response.getHeader('Cache-Control')) + self.assertEqual('max-age=0, must-revalidate, private', + response.getHeader('Cache-Control')) self.assertEqual(None, response.getHeader('Last-Modified')) expires = dateutil.parser.parse(response.getHeader('Expires')) self.assertTrue(now > expires) - # cacheInBrowser() def test_cacheInBrowser_no_etag_no_last_modified(self): @@ -104,7 +107,8 @@ def test_cacheInBrowser_no_etag_no_last_modified(self): cacheInBrowser(published, request, response) self.assertEqual(200, response.getStatus()) - self.assertEqual('max-age=0, must-revalidate, private', response.getHeader('Cache-Control')) + self.assertEqual('max-age=0, must-revalidate, private', + response.getHeader('Cache-Control')) self.assertEqual(None, response.getHeader('Last-Modified')) self.assertEqual(None, response.getHeader('ETag', literal=1)) @@ -125,7 +129,8 @@ def test_cacheInBrowser_etag(self): cacheInBrowser(published, request, response, etag=etag) self.assertEqual(200, response.getStatus()) - self.assertEqual('max-age=0, must-revalidate, private', response.getHeader('Cache-Control')) + self.assertEqual('max-age=0, must-revalidate, private', + response.getHeader('Cache-Control')) self.assertEqual(None, response.getHeader('Last-Modified')) self.assertEqual('"|foo|bar|"', response.getHeader('ETag', literal=1)) @@ -141,12 +146,14 @@ def test_cacheInBrowser_lastModified(self): published = DummyPublished() now = datetime.datetime.now(dateutil.tz.tzlocal()) - nowFormatted = wsgiref.handlers.format_date_time(time.mktime(now.timetuple())) + nowFormatted = wsgiref.handlers.format_date_time( + time.mktime(now.timetuple())) cacheInBrowser(published, request, response, lastModified=now) self.assertEqual(200, response.getStatus()) - self.assertEqual('max-age=0, must-revalidate, private', response.getHeader('Cache-Control')) + self.assertEqual('max-age=0, must-revalidate, private', + response.getHeader('Cache-Control')) self.assertEqual(nowFormatted, response.getHeader('Last-Modified')) self.assertEqual(None, response.getHeader('ETag', literal=1)) @@ -164,19 +171,21 @@ def test_cacheInBrowser_lastModified_and_etag(self): now = datetime.datetime.now(dateutil.tz.tzlocal()) etag = "|foo|bar|" - nowFormatted = wsgiref.handlers.format_date_time(time.mktime(now.timetuple())) + nowFormatted = wsgiref.handlers.format_date_time( + time.mktime(now.timetuple())) - cacheInBrowser(published, request, response, etag=etag, lastModified=now) + cacheInBrowser(published, request, response, + etag=etag, lastModified=now) self.assertEqual(200, response.getStatus()) - self.assertEqual('max-age=0, must-revalidate, private', response.getHeader('Cache-Control')) + self.assertEqual('max-age=0, must-revalidate, private', + response.getHeader('Cache-Control')) self.assertEqual(nowFormatted, response.getHeader('Last-Modified')) self.assertEqual('"|foo|bar|"', response.getHeader('ETag', literal=1)) expires = dateutil.parser.parse(response.getHeader('Expires')) self.assertTrue(now > expires) - # cacheInProxy() def test_cacheInProxy_minimal(self): @@ -192,7 +201,8 @@ def test_cacheInProxy_minimal(self): cacheInProxy(published, request, response, smaxage=60) self.assertEqual(200, response.getStatus()) - self.assertEqual('max-age=0, s-maxage=60, must-revalidate', response.getHeader('Cache-Control')) + self.assertEqual('max-age=0, s-maxage=60, must-revalidate', + response.getHeader('Cache-Control')) self.assertEqual(None, response.getHeader('Last-Modified')) self.assertEqual(None, response.getHeader('ETag', literal=1)) self.assertEqual(None, response.getHeader('Vary')) @@ -212,12 +222,15 @@ def test_cacheInProxy_full(self): etag = '|foo|bar|' vary = 'Accept-Language' - nowFormatted = wsgiref.handlers.format_date_time(time.mktime(now.timetuple())) + nowFormatted = wsgiref.handlers.format_date_time( + time.mktime(now.timetuple())) - cacheInProxy(published, request, response, smaxage=60, etag=etag, lastModified=now, vary=vary) + cacheInProxy(published, request, response, smaxage=60, + etag=etag, lastModified=now, vary=vary) self.assertEqual(200, response.getStatus()) - self.assertEqual('max-age=0, s-maxage=60, must-revalidate', response.getHeader('Cache-Control')) + self.assertEqual('max-age=0, s-maxage=60, must-revalidate', + response.getHeader('Cache-Control')) self.assertEqual(nowFormatted, response.getHeader('Last-Modified')) self.assertEqual('"|foo|bar|"', response.getHeader('ETag', literal=1)) self.assertEqual('Accept-Language', response.getHeader('Vary')) @@ -225,7 +238,6 @@ def test_cacheInProxy_full(self): expires = dateutil.parser.parse(response.getHeader('Expires')) self.assertTrue(now > expires) - # cacheInBrowserAndProxy() def test_cacheInBrowserAndProxy_minimal(self): @@ -241,7 +253,8 @@ def test_cacheInBrowserAndProxy_minimal(self): cacheInBrowserAndProxy(published, request, response, maxage=60) self.assertEqual(200, response.getStatus()) - self.assertEqual('max-age=60, proxy-revalidate, public', response.getHeader('Cache-Control')) + self.assertEqual('max-age=60, proxy-revalidate, public', + response.getHeader('Cache-Control')) self.assertEqual(None, response.getHeader('Last-Modified')) self.assertEqual(None, response.getHeader('ETag', literal=1)) self.assertEqual(None, response.getHeader('Vary')) @@ -262,12 +275,15 @@ def test_cacheInBrowserAndProxy_full(self): etag = '|foo|bar|' vary = 'Accept-Language' - nowFormatted = wsgiref.handlers.format_date_time(time.mktime(now.timetuple())) + nowFormatted = wsgiref.handlers.format_date_time( + time.mktime(now.timetuple())) - cacheInBrowserAndProxy(published, request, response, maxage=60, etag=etag, lastModified=now, vary=vary) + cacheInBrowserAndProxy(published, request, response, + maxage=60, etag=etag, lastModified=now, vary=vary) self.assertEqual(200, response.getStatus()) - self.assertEqual('max-age=60, proxy-revalidate, public', response.getHeader('Cache-Control')) + self.assertEqual('max-age=60, proxy-revalidate, public', + response.getHeader('Cache-Control')) self.assertEqual(nowFormatted, response.getHeader('Last-Modified')) self.assertEqual('"|foo|bar|"', response.getHeader('ETag', literal=1)) self.assertEqual('Accept-Language', response.getHeader('Vary')) @@ -282,7 +298,6 @@ def test_cacheInBrowserAndProxy_full(self): "%s is not > %s" % (timedelta, datetime.timedelta(seconds=61)) ) - # cacheInRAM() def test_cacheInRAM_no_etag(self): @@ -303,7 +318,8 @@ def test_cacheInRAM_no_etag(self): cacheInRAM(published, request, response) annotations = IAnnotations(request) - self.assertEqual("http://example.com/foo?", annotations[PAGE_CACHE_ANNOTATION_KEY]) + self.assertEqual("http://example.com/foo?", + annotations[PAGE_CACHE_ANNOTATION_KEY]) self.assertTrue(IRAMCached.providedBy(request)) def test_cacheInRAM_etag(self): @@ -325,7 +341,8 @@ def test_cacheInRAM_etag(self): cacheInRAM(published, request, response, etag=etag) annotations = IAnnotations(request) - self.assertEqual("||foo|bar|||http://example.com/foo?", annotations[PAGE_CACHE_ANNOTATION_KEY]) + self.assertEqual("||foo|bar|||http://example.com/foo?", + annotations[PAGE_CACHE_ANNOTATION_KEY]) self.assertTrue(IRAMCached.providedBy(request)) def test_cacheInRAM_etag_alternate_key(self): @@ -343,10 +360,12 @@ def test_cacheInRAM_etag_alternate_key(self): assert not IRAMCached.providedBy(response) - cacheInRAM(published, request, response, etag=etag, annotationsKey="alt.key") + cacheInRAM(published, request, response, + etag=etag, annotationsKey="alt.key") annotations = IAnnotations(request) - self.assertEqual("||foo|bar|||http://example.com/foo?", annotations["alt.key"]) + self.assertEqual("||foo|bar|||http://example.com/foo?", + annotations["alt.key"]) self.assertTrue(IRAMCached.providedBy(request)) @@ -378,7 +397,8 @@ def test_cachedResponse(self): response.setHeader('X-Bar', 'qux') response.setStatus(200) - body = cachedResponse(published, request, response, 404, headers, u"body") + body = cachedResponse(published, request, + response, 404, headers, u"body") self.assertEqual(u"body", body) self.assertEqual(404, response.getStatus()) @@ -410,7 +430,8 @@ def test_cachedResponse_gzip_off(self): assert response.enableHTTPCompression(query=True) - body = cachedResponse(published, request, response, 404, headers, u"body", 0) + body = cachedResponse(published, request, response, + 404, headers, u"body", 0) self.assertEqual(u"body", body) self.assertEqual(404, response.getStatus()) @@ -444,7 +465,8 @@ def test_cachedResponse_gzip_on(self): assert not response.enableHTTPCompression(query=True) - body = cachedResponse(published, request, response, 404, headers, u"body", 1) + body = cachedResponse(published, request, response, + 404, headers, u"body", 1) self.assertEqual(u"body", body) self.assertEqual(404, response.getStatus()) @@ -485,7 +507,8 @@ def test_notModified_full(self): now = datetime.datetime.now(dateutil.tz.tzlocal()) etag = "|foo|bar|" - body = notModified(published, request, response, etag=etag, lastModified=now) + body = notModified(published, request, response, + etag=etag, lastModified=now) self.assertEqual(u"", body) self.assertEqual(etag, response.getHeader('ETag', literal=1)) @@ -494,6 +517,7 @@ def test_notModified_full(self): self.assertEqual(None, response.getHeader('Cache-Control')) self.assertEqual(304, response.getStatus()) + class CacheCheckHelpersTest(unittest.TestCase): layer = UNIT_TESTING @@ -521,9 +545,11 @@ def test_isModified_no_headers_with_keys(self): request = HTTPRequest(StringIO(), environ, response) etag = '|foo' - lastModified = datetime.datetime(2010, 11, 24, 3, 4, 5, 6, dateutil.tz.tzutc()) + lastModified = datetime.datetime( + 2010, 11, 24, 3, 4, 5, 6, dateutil.tz.tzutc()) - self.assertEqual(True, isModified(request, etag=etag, lastModified=lastModified)) + self.assertEqual(True, isModified( + request, etag=etag, lastModified=lastModified)) def test_isModified_ims_invalid_date(self): from plone.app.caching.operations.utils import isModified @@ -534,7 +560,8 @@ def test_isModified_ims_invalid_date(self): request.environ['HTTP_IF_MODIFIED_SINCE'] = 'blah' - lastModified = datetime.datetime(2010, 11, 24, 3, 4, 5, 6, dateutil.tz.tzutc()) + lastModified = datetime.datetime( + 2010, 11, 24, 3, 4, 5, 6, dateutil.tz.tzutc()) self.assertEqual(True, isModified(request, lastModified=lastModified)) @@ -545,9 +572,11 @@ def test_isModified_ims_modified(self): response = HTTPResponse() request = HTTPRequest(StringIO(), environ, response) - request.environ['HTTP_IF_MODIFIED_SINCE'] = 'Tue, 24 Nov 2009 03:04:05 GMT' + request.environ[ + 'HTTP_IF_MODIFIED_SINCE'] = 'Tue, 24 Nov 2009 03:04:05 GMT' - lastModified = datetime.datetime(2010, 11, 24, 3, 4, 5, 6, dateutil.tz.tzutc()) + lastModified = datetime.datetime( + 2010, 11, 24, 3, 4, 5, 6, dateutil.tz.tzutc()) self.assertEqual(True, isModified(request, lastModified=lastModified)) @@ -558,9 +587,11 @@ def test_isModified_ims_not_modified(self): response = HTTPResponse() request = HTTPRequest(StringIO(), environ, response) - request.environ['HTTP_IF_MODIFIED_SINCE'] = 'Thu, 24 Nov 2011 03:04:05 GMT' + request.environ[ + 'HTTP_IF_MODIFIED_SINCE'] = 'Thu, 24 Nov 2011 03:04:05 GMT' - lastModified = datetime.datetime(2010, 11, 24, 3, 4, 5, 6, dateutil.tz.tzutc()) + lastModified = datetime.datetime( + 2010, 11, 24, 3, 4, 5, 6, dateutil.tz.tzutc()) self.assertEqual(False, isModified(request, lastModified=lastModified)) @@ -571,9 +602,11 @@ def test_isModified_ims_not_modified_two_dates(self): response = HTTPResponse() request = HTTPRequest(StringIO(), environ, response) - request.environ['HTTP_IF_MODIFIED_SINCE'] = 'Thu, 24 Nov 2011 03:04:05 GMT; Thu, 24 Nov 2011 03:04:05' + request.environ[ + 'HTTP_IF_MODIFIED_SINCE'] = 'Thu, 24 Nov 2011 03:04:05 GMT; Thu, 24 Nov 2011 03:04:05' - lastModified = datetime.datetime(2010, 11, 24, 3, 4, 5, 6, dateutil.tz.tzutc()) + lastModified = datetime.datetime( + 2010, 11, 24, 3, 4, 5, 6, dateutil.tz.tzutc()) self.assertEqual(False, isModified(request, lastModified=lastModified)) @@ -584,12 +617,15 @@ def test_isModified_ims_not_modified_etag_no_inm_header(self): response = HTTPResponse() request = HTTPRequest(StringIO(), environ, response) - request.environ['HTTP_IF_MODIFIED_SINCE'] = 'Thu, 24 Nov 2011 03:04:05 GMT' + request.environ[ + 'HTTP_IF_MODIFIED_SINCE'] = 'Thu, 24 Nov 2011 03:04:05 GMT' etag = '|foo' - lastModified = datetime.datetime(2010, 11, 24, 3, 4, 5, 6, dateutil.tz.tzutc()) + lastModified = datetime.datetime( + 2010, 11, 24, 3, 4, 5, 6, dateutil.tz.tzutc()) - self.assertEqual(True, isModified(request, etag=etag, lastModified=lastModified)) + self.assertEqual(True, isModified( + request, etag=etag, lastModified=lastModified)) def test_isModified_inm_no_tags(self): from plone.app.caching.operations.utils import isModified @@ -690,8 +726,7 @@ def test_isModified_inm_match_update(self): etag = None self.assertEqual(True, isModified(request, etag=etag, - lastModified='doesnt_really_matter')) - + lastModified='doesnt_really_matter')) def test_isModified_inm_match_multiple(self): from plone.app.caching.operations.utils import isModified @@ -727,7 +762,8 @@ def test_isModified_inm_match_ignores_ims_if_no_last_modified_date(self): request = HTTPRequest(StringIO(), environ, response) request.environ['HTTP_IF_NONE_MATCH'] = '"|foo"' - request.environ['HTTP_IF_MODIFIED_SINCE'] = 'Tue, 24 Nov 2009 03:04:05 GMT' + request.environ[ + 'HTTP_IF_MODIFIED_SINCE'] = 'Tue, 24 Nov 2009 03:04:05 GMT' etag = '|foo' @@ -741,12 +777,15 @@ def test_isModified_inm_match_modified(self): request = HTTPRequest(StringIO(), environ, response) request.environ['HTTP_IF_NONE_MATCH'] = '"|foo"' - request.environ['HTTP_IF_MODIFIED_SINCE'] = 'Tue, 24 Nov 2009 03:04:05 GMT' + request.environ[ + 'HTTP_IF_MODIFIED_SINCE'] = 'Tue, 24 Nov 2009 03:04:05 GMT' etag = '|foo' - lastModified = datetime.datetime(2010, 11, 24, 3, 4, 5, 6, dateutil.tz.tzutc()) + lastModified = datetime.datetime( + 2010, 11, 24, 3, 4, 5, 6, dateutil.tz.tzutc()) - self.assertEqual(True, isModified(request, etag=etag, lastModified=lastModified)) + self.assertEqual(True, isModified( + request, etag=etag, lastModified=lastModified)) def test_isModified_inm_match_not_modified(self): from plone.app.caching.operations.utils import isModified @@ -756,12 +795,15 @@ def test_isModified_inm_match_not_modified(self): request = HTTPRequest(StringIO(), environ, response) request.environ['HTTP_IF_NONE_MATCH'] = '"|foo"' - request.environ['HTTP_IF_MODIFIED_SINCE'] = 'Thu, 24 Nov 2011 03:04:05 GMT' + request.environ[ + 'HTTP_IF_MODIFIED_SINCE'] = 'Thu, 24 Nov 2011 03:04:05 GMT' etag = '|foo' - lastModified = datetime.datetime(2010, 11, 24, 3, 4, 5, 6, dateutil.tz.tzutc()) + lastModified = datetime.datetime( + 2010, 11, 24, 3, 4, 5, 6, dateutil.tz.tzutc()) - self.assertEqual(False, isModified(request, etag=etag, lastModified=lastModified)) + self.assertEqual(False, isModified( + request, etag=etag, lastModified=lastModified)) # visibleToRole() @@ -849,7 +891,8 @@ def test_formatDateTime_local(self): # We lose microseconds. Big whoop. p = dateutil.parser.parse(inGMT).astimezone(dateutil.tz.tzlocal()) - lofi = datetime.datetime(2010, 11, 24, 3, 4, 5, 0, dateutil.tz.tzlocal()) + lofi = datetime.datetime( + 2010, 11, 24, 3, 4, 5, 0, dateutil.tz.tzlocal()) self.assertEqual(p, lofi) def test_formatDateTime_naive(self): @@ -864,7 +907,8 @@ def test_formatDateTime_naive(self): # Can't compare offset aware and naive p = dateutil.parser.parse(inGMT).astimezone(dateutil.tz.tzlocal()) - self.assertEqual((2010, 11, 24, 3, 4, 5), (p.year, p.month, p.day, p.hour, p.minute, p.second)) + self.assertEqual((2010, 11, 24, 3, 4, 5), + (p.year, p.month, p.day, p.hour, p.minute, p.second)) # parseDateTime() @@ -883,7 +927,8 @@ def test_formatDateTime_no_timezone(self): from plone.app.caching.operations.utils import parseDateTime # parser will assume input was local time - dt = datetime.datetime(2010, 11, 23, 19, 4, 5, 0, dateutil.tz.tzlocal()) + dt = datetime.datetime(2010, 11, 23, 19, 4, 5, + 0, dateutil.tz.tzlocal()) self.assertEqual(dt, parseDateTime("'Tue, 23 Nov 2010 19:04:05'")) # getLastModified() @@ -929,7 +974,7 @@ def __call__(self): published = DummyPublished() self.assertEqual(datetime.datetime(2010, 11, 24, 3, 4, 5, 6, dateutil.tz.tzlocal()), - getLastModified(published)) + getLastModified(published)) def test_getLastModified_timezone(self): from plone.app.caching.operations.utils import getLastModified @@ -948,8 +993,7 @@ def __call__(self): published = DummyPublished() self.assertEqual(datetime.datetime(2010, 11, 24, 3, 4, 5, 6, dateutil.tz.tzutc()), - getLastModified(published)) - + getLastModified(published)) # getExpiration() @@ -988,7 +1032,6 @@ def test_getExpiration_future(self): # give the test two seconds' leeway self.assertTrue(difference >= datetime.timedelta(seconds=58)) - # getETag() def test_getETag_extra_only(self): @@ -999,7 +1042,8 @@ def test_getETag_extra_only(self): request = HTTPRequest(StringIO(), environ, response) published = DummyPublished() - self.assertEqual('|foo|bar;baz', getETag(published, request, extraTokens=('foo', 'bar,baz'))) + self.assertEqual('|foo|bar;baz', getETag( + published, request, extraTokens=('foo', 'bar,baz'))) def test_getETag_key_not_found(self): from plone.app.caching.operations.utils import getETag @@ -1009,7 +1053,8 @@ def test_getETag_key_not_found(self): request = HTTPRequest(StringIO(), environ, response) published = DummyPublished() - self.assertEqual(None, getETag(published, request, keys=('foo', 'bar',))) + self.assertEqual(None, getETag( + published, request, keys=('foo', 'bar',))) def test_getETag_adapter_returns_none(self): from plone.app.caching.operations.utils import getETag @@ -1046,7 +1091,8 @@ def __call__(self): request = HTTPRequest(StringIO(), environ, response) published = DummyPublished() - self.assertEqual('|foo|', getETag(published, request, keys=('foo', 'bar',))) + self.assertEqual('|foo|', getETag( + published, request, keys=('foo', 'bar',))) def test_getETag_full(self): from plone.app.caching.operations.utils import getETag @@ -1084,8 +1130,7 @@ def __call__(self): published = DummyPublished() self.assertEqual('|foo|bar|baz;qux', getETag(published, request, - keys=('foo', 'bar',), extraTokens=('baz,qux',))) - + keys=('foo', 'bar',), extraTokens=('baz,qux',))) # parseETags() @@ -1115,35 +1160,44 @@ def test_parseETags_single_quoted(self): def test_parseETags_multiple(self): from plone.app.caching.operations.utils import parseETags - self.assertEqual(['|foo|bar;baz', '1234'], parseETags('|foo|bar;baz, 1234')) + self.assertEqual(['|foo|bar;baz', '1234'], + parseETags('|foo|bar;baz, 1234')) def test_parseETags_multiple_quoted(self): from plone.app.caching.operations.utils import parseETags - self.assertEqual(['|foo|bar;baz', '1234'], parseETags('"|foo|bar;baz", "1234"')) + self.assertEqual(['|foo|bar;baz', '1234'], + parseETags('"|foo|bar;baz", "1234"')) def test_parseETags_multiple_nospace(self): from plone.app.caching.operations.utils import parseETags - self.assertEqual(['|foo|bar;baz', '1234'], parseETags('|foo|bar;baz,1234')) + self.assertEqual(['|foo|bar;baz', '1234'], + parseETags('|foo|bar;baz,1234')) def test_parseETags_multiple_quoted_nospace(self): from plone.app.caching.operations.utils import parseETags - self.assertEqual(['|foo|bar;baz', '1234'], parseETags('"|foo|bar;baz","1234"')) + self.assertEqual(['|foo|bar;baz', '1234'], + parseETags('"|foo|bar;baz","1234"')) def test_parseETags_multiple_weak(self): from plone.app.caching.operations.utils import parseETags - self.assertEqual(['|foo|bar;baz', '1234'], parseETags('|foo|bar;baz, W/1234')) + self.assertEqual(['|foo|bar;baz', '1234'], + parseETags('|foo|bar;baz, W/1234')) def test_parseETags_multiple_quoted_weak(self): from plone.app.caching.operations.utils import parseETags - self.assertEqual(['|foo|bar;baz', '1234'], parseETags('"|foo|bar;baz", W/"1234"')) + self.assertEqual(['|foo|bar;baz', '1234'], + parseETags('"|foo|bar;baz", W/"1234"')) def test_parseETags_multiple_weak_disallowed(self): from plone.app.caching.operations.utils import parseETags - self.assertEqual(['|foo|bar;baz'], parseETags('|foo|bar;baz, W/1234', allowWeak=False)) + self.assertEqual(['|foo|bar;baz'], parseETags( + '|foo|bar;baz, W/1234', allowWeak=False)) def test_parseETags_multiple_quoted_weak_disallowed(self): from plone.app.caching.operations.utils import parseETags - self.assertEqual(['|foo|bar;baz'], parseETags('"|foo|bar;baz", W/"1234"', allowWeak=False)) + self.assertEqual(['|foo|bar;baz'], parseETags( + '"|foo|bar;baz", W/"1234"', allowWeak=False)) + class RAMCacheTest(unittest.TestCase): @@ -1195,7 +1249,6 @@ def __call__(self, key): provideUtility(Chooser()) self.assertEqual(cache, getRAMCache()) - # getRAMCacheKey() def test_getRAMCacheKey_empty(self): @@ -1217,7 +1270,8 @@ def test_getRAMCacheKey_normal(self): request.environ['PATH_INFO'] = '/foo/bar' request.environ['QUERY_STRING'] = 'x=1&y=2' - self.assertEqual('http://example.com/foo/bar?x=1&y=2', getRAMCacheKey(request)) + self.assertEqual('http://example.com/foo/bar?x=1&y=2', + getRAMCacheKey(request)) def test_getRAMCacheKey_etag(self): from plone.app.caching.operations.utils import getRAMCacheKey @@ -1229,8 +1283,8 @@ def test_getRAMCacheKey_etag(self): request.environ['PATH_INFO'] = '/foo/bar' request.environ['QUERY_STRING'] = 'x=1&y=2' - self.assertEqual('||foo|bar||http://example.com/foo/bar?x=1&y=2', getRAMCacheKey(request, etag="|foo|bar")) - + self.assertEqual('||foo|bar||http://example.com/foo/bar?x=1&y=2', + getRAMCacheKey(request, etag="|foo|bar")) # storeResponseInRAMCache() @@ -1272,7 +1326,8 @@ def test_storeResponseInRAMCache_no_cache(self): result = u"Body" response.setHeader('X-Foo', 'bar') - IAnnotations(request)['plone.app.caching.operations.ramcache.key'] = 'foo' + IAnnotations(request)[ + 'plone.app.caching.operations.ramcache.key'] = 'foo' storeResponseInRAMCache(request, response, result) @@ -1300,7 +1355,8 @@ def __call__(self, key): result = u"Body" response.setHeader('X-Foo', 'bar') - IAnnotations(request)['plone.app.caching.operations.ramcache.key'] = 'foo' + IAnnotations(request)[ + 'plone.app.caching.operations.ramcache.key'] = 'foo' storeResponseInRAMCache(request, response, result) @@ -1335,7 +1391,8 @@ def __call__(self, key): result = u"Body" response.setHeader('X-Foo', 'bar') - IAnnotations(request)['plone.app.caching.operations.ramcache.key'] = 'foo' + IAnnotations(request)[ + 'plone.app.caching.operations.ramcache.key'] = 'foo' storeResponseInRAMCache(request, response, result) @@ -1369,7 +1426,8 @@ def __call__(self, key): IAnnotations(request)['annkey'] = 'foo' - storeResponseInRAMCache(request, response, result, globalKey='cachekey', annotationsKey='annkey') + storeResponseInRAMCache(request, response, result, + globalKey='cachekey', annotationsKey='annkey') self.assertEqual(1, len(cache)) cached = normalize_response_cache(cache['foo']) @@ -1442,9 +1500,11 @@ def __call__(self, key): request.environ['PATH_INFO'] = '/foo/bar' request.environ['QUERY_STRING'] = '' - cache['||a|b||http://example.com/foo/bar?'] = (200, {'x-foo': 'bar'}, u'Body') + cache[ + '||a|b||http://example.com/foo/bar?'] = (200, {'x-foo': 'bar'}, u'Body') - cached = normalize_response_cache(fetchFromRAMCache(request, etag="|a|b")) + cached = normalize_response_cache( + fetchFromRAMCache(request, etag="|a|b")) self.assertEqual((200, {'x-foo': 'bar'}, u'Body'), cached) def test_fetchFromRAMCache_custom_key(self): @@ -1473,7 +1533,8 @@ def __call__(self, key): cache['http://example.com/foo/bar?'] = (200, {'x-foo': 'bar'}, u'Body') - cached = normalize_response_cache(fetchFromRAMCache(request, globalKey='cachekey')) + cached = normalize_response_cache( + fetchFromRAMCache(request, globalKey='cachekey')) self.assertEqual((200, {'x-foo': 'bar'}, u'Body'), cached) def test_fetchFromRAMCache_miss(self): @@ -1502,7 +1563,8 @@ def __call__(self, key): cache['http://example.com/foo/bar?'] = (200, {'x-foo': 'bar'}, u'Body') - cached = normalize_response_cache(fetchFromRAMCache(request, etag='|foo')) + cached = normalize_response_cache( + fetchFromRAMCache(request, etag='|foo')) self.assertEqual(None, cached) def test_fetchFromRAMCache_miss_custom_default(self): @@ -1532,5 +1594,6 @@ def __call__(self, key): cache['http://example.com/foo/bar?'] = (200, {'x-foo': 'bar'}, u'Body') marker = object() - cached = normalize_response_cache(fetchFromRAMCache(request, etag='|foo', default=marker)) + cached = normalize_response_cache( + fetchFromRAMCache(request, etag='|foo', default=marker)) self.assertTrue(cached is marker) diff --git a/plone/app/caching/tests/test_profile_with_caching_proxy.py b/plone/app/caching/tests/test_profile_with_caching_proxy.py index 9143980..e027712 100644 --- a/plone/app/caching/tests/test_profile_with_caching_proxy.py +++ b/plone/app/caching/tests/test_profile_with_caching_proxy.py @@ -31,12 +31,14 @@ import urllib -TEST_FILE = pkg_resources.resource_filename('plone.app.caching.tests', 'test.gif') +TEST_FILE = pkg_resources.resource_filename( + 'plone.app.caching.tests', 'test.gif') def test_image(): from plone.namedfile.file import NamedBlobImage - filename = pkg_resources.resource_filename('plone.app.caching.tests', 'test.gif') + filename = pkg_resources.resource_filename( + 'plone.app.caching.tests', 'test.gif') filename = os.path.join(os.path.dirname(__file__), u'test.gif') return NamedBlobImage( data=open(filename, 'r').read(), @@ -65,7 +67,8 @@ def setUp(self): self.app = self.layer['app'] self.portal = self.layer['portal'] - test_css = FSFile('test.css', os.path.join(os.path.dirname(__file__), 'test.css')) + test_css = FSFile('test.css', os.path.join( + os.path.dirname(__file__), 'test.css')) self.portal.portal_skins.custom._setOb('test.css', test_css) setRequest(self.portal.REQUEST) @@ -75,8 +78,10 @@ def setUp(self): self.registry = getUtility(IRegistry) self.cacheSettings = self.registry.forInterface(ICacheSettings) - self.cachePurgingSettings = self.registry.forInterface(ICachePurgingSettings) - self.ploneCacheSettings = self.registry.forInterface(IPloneCacheSettings) + self.cachePurgingSettings = self.registry.forInterface( + ICachePurgingSettings) + self.ploneCacheSettings = self.registry.forInterface( + IPloneCacheSettings) self.cacheSettings.enabled = True @@ -114,7 +119,8 @@ def test_composite_views(self): # Publish the folder and page self.portal.portal_workflow.doActionFor(self.portal['f1'], 'publish') - self.portal.portal_workflow.doActionFor(self.portal['f1']['d1'], 'publish') + self.portal.portal_workflow.doActionFor( + self.portal['f1']['d1'], 'publish') # Should we set up the etag components? # - set member? No @@ -125,56 +131,78 @@ def test_composite_views(self): # - leave status unlocked # - import transaction; transaction.commit() + import transaction + transaction.commit() # Request the authenticated folder now = stable_now() browser = Browser(self.app) - browser.addHeader('Authorization', 'Basic %s:%s' % (TEST_USER_NAME, TEST_USER_PASSWORD,)) + browser.addHeader('Authorization', 'Basic %s:%s' % + (TEST_USER_NAME, TEST_USER_PASSWORD,)) browser.open(self.portal['f1'].absolute_url()) - self.assertEqual('plone.content.folderView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.folderView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # This should use cacheInBrowser - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertEqual('"|test_user_1_|%d|en|%s|0|0' % (catalog.getCounter(), skins_tool.default_skin), _normalize_etag(browser.headers['ETag'])) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + self.assertEqual('"|test_user_1_|%d|en|%s|0|0' % (catalog.getCounter( + ), skins_tool.default_skin), _normalize_etag(browser.headers['ETag'])) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Set the copy/cut cookie and then request the folder view again browser.cookies.create('__cp', 'xxx') browser.open(self.portal['f1'].absolute_url()) # The response should be the same as before except for the etag - self.assertEqual('plone.content.folderView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertEqual('"|test_user_1_|%d|en|%s|0|1' % (catalog.getCounter(), skins_tool.default_skin), _normalize_etag(browser.headers['ETag'])) + self.assertEqual('plone.content.folderView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + self.assertEqual('"|test_user_1_|%d|en|%s|0|1' % (catalog.getCounter( + ), skins_tool.default_skin), _normalize_etag(browser.headers['ETag'])) # Request the authenticated page now = stable_now() browser = Browser(self.app) - browser.addHeader('Authorization', 'Basic %s:%s' % (TEST_USER_NAME, TEST_USER_PASSWORD,)) + browser.addHeader('Authorization', 'Basic %s:%s' % + (TEST_USER_NAME, TEST_USER_PASSWORD,)) browser.open(self.portal['f1']['d1'].absolute_url()) self.assertTrue(testText in browser.contents) - self.assertEqual('plone.content.itemView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.itemView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # This should use cacheInBrowser - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertEqual('"|test_user_1_|%d|en|%s|0' % (catalog.getCounter(), skins_tool.default_skin), _normalize_etag(browser.headers['ETag'])) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + self.assertEqual('"|test_user_1_|%d|en|%s|0' % (catalog.getCounter( + ), skins_tool.default_skin), _normalize_etag(browser.headers['ETag'])) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Request the authenticated page again -- to test RAM cache. browser = Browser(self.app) - browser.addHeader('Authorization', 'Basic %s:%s' % (TEST_USER_NAME, TEST_USER_PASSWORD,)) + browser.addHeader('Authorization', 'Basic %s:%s' % + (TEST_USER_NAME, TEST_USER_PASSWORD,)) browser.open(self.portal['f1']['d1'].absolute_url()) - self.assertEqual('plone.content.itemView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.itemView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # Authenticated should NOT be RAM cached self.assertEqual(None, browser.headers.get('X-RAMCache')) - # Request the authenticated page again -- with an INM header to test 304 + # Request the authenticated page again -- with an INM header to test + # 304 etag = browser.headers['ETag'] browser = Browser(self.app) browser.raiseHttpErrors = False # we really do want to see the 304 - browser.addHeader('Authorization', 'Basic %s:%s' % (TEST_USER_NAME, TEST_USER_PASSWORD,)) + browser.addHeader('Authorization', 'Basic %s:%s' % + (TEST_USER_NAME, TEST_USER_PASSWORD,)) browser.addHeader('If-None-Match', etag) browser.open(self.portal['f1']['d1'].absolute_url()) # This should be a 304 response @@ -185,38 +213,54 @@ def test_composite_views(self): now = stable_now() browser = Browser(self.app) browser.open(self.portal['f1'].absolute_url()) - self.assertEqual('plone.content.folderView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.folderView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # This should use cacheInBrowser - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertEqual('"||%d|en|%s|0|0' % (catalog.getCounter(), skins_tool.default_skin), _normalize_etag(browser.headers['ETag'])) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + self.assertEqual('"||%d|en|%s|0|0' % (catalog.getCounter( + ), skins_tool.default_skin), _normalize_etag(browser.headers['ETag'])) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Request the anonymous page now = stable_now() browser = Browser(self.app) browser.open(self.portal['f1']['d1'].absolute_url()) - self.assertEqual('plone.content.itemView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.itemView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) self.assertTrue(testText in browser.contents) # This should use cacheInBrowser - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertEqual('"||%d|en|%s|0' % (catalog.getCounter(), skins_tool.default_skin), _normalize_etag(browser.headers['ETag'])) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + self.assertEqual('"||%d|en|%s|0' % (catalog.getCounter( + ), skins_tool.default_skin), _normalize_etag(browser.headers['ETag'])) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Request the anonymous page again -- to test RAM cache. # Anonymous should be RAM cached now = stable_now() browser = Browser(self.app) browser.open(self.portal['f1']['d1'].absolute_url()) - self.assertEqual('plone.content.itemView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.itemView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # This should come from RAM cache - self.assertEqual('plone.app.caching.operations.ramcache', browser.headers['X-RAMCache']) + self.assertEqual('plone.app.caching.operations.ramcache', + browser.headers['X-RAMCache']) self.assertTrue(testText in browser.contents) - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertEqual('"||%d|en|%s|0' % (catalog.getCounter(), skins_tool.default_skin), _normalize_etag(browser.headers['ETag'])) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + self.assertEqual('"||%d|en|%s|0' % (catalog.getCounter( + ), skins_tool.default_skin), _normalize_etag(browser.headers['ETag'])) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Request the anonymous page again -- with an INM header to test 304. etag = browser.headers['ETag'] @@ -224,8 +268,10 @@ def test_composite_views(self): browser.raiseHttpErrors = False browser.addHeader('If-None-Match', etag) browser.open(self.portal['f1']['d1'].absolute_url()) - self.assertEqual('plone.content.itemView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.itemView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # This should be a 304 response self.assertEqual('304 Not Modified', browser.headers['Status']) self.assertEqual('', browser.contents) @@ -239,15 +285,19 @@ def test_composite_views(self): ) self.portal['f1']['d1'].reindexObject() - import transaction; transaction.commit() + import transaction + transaction.commit() - # Request the anonymous page again -- to test expiration of 304 and RAM. + # Request the anonymous page again -- to test expiration of 304 and + # RAM. etag = browser.headers['ETag'] browser = Browser(self.app) browser.addHeader('If-None-Match', etag) browser.open(self.portal['f1']['d1'].absolute_url()) - self.assertEqual('plone.content.itemView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.itemView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # The etag has changed so we should get a fresh page. self.assertEqual(None, browser.headers.get('X-RAMCache')) self.assertEqual('200 Ok', browser.headers['Status']) @@ -263,18 +313,23 @@ def test_content_feeds(self): self.syndication.editProperties(isAllowed=True) self.syndication.enableSyndication(self.portal) - import transaction; transaction.commit() + import transaction + transaction.commit() # Request the rss feed now = stable_now() browser = Browser(self.app) browser.open(self.portal.absolute_url() + '/RSS') self.assertEqual('plone.content.feed', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.moderateCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.moderateCaching', + browser.headers['X-Cache-Operation']) # This should use cacheInProxy - self.assertEqual('max-age=0, s-maxage=86400, must-revalidate', browser.headers['Cache-Control']) - self.assertEqual('"||%d|en|%s"' % (catalog.getCounter(), skins_tool.default_skin), browser.headers['ETag']) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertEqual('max-age=0, s-maxage=86400, must-revalidate', + browser.headers['Cache-Control']) + self.assertEqual('"||%d|en|%s"' % (catalog.getCounter(), + skins_tool.default_skin), browser.headers['ETag']) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Request the rss feed again -- to test RAM cache now = stable_now() @@ -282,13 +337,18 @@ def test_content_feeds(self): browser = Browser(self.app) browser.open(self.portal.absolute_url() + '/RSS') self.assertEqual('plone.content.feed', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.moderateCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.moderateCaching', + browser.headers['X-Cache-Operation']) # This should come from the RAM cache - self.assertEqual('plone.app.caching.operations.ramcache', browser.headers['X-RAMCache']) + self.assertEqual('plone.app.caching.operations.ramcache', + browser.headers['X-RAMCache']) self.assertEqual(rssText, browser.contents) - self.assertEqual('max-age=0, s-maxage=86400, must-revalidate', browser.headers['Cache-Control']) - self.assertEqual('"||%d|en|%s"' % (catalog.getCounter(), skins_tool.default_skin), browser.headers['ETag']) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertEqual('max-age=0, s-maxage=86400, must-revalidate', + browser.headers['Cache-Control']) + self.assertEqual('"||%d|en|%s"' % (catalog.getCounter(), + skins_tool.default_skin), browser.headers['ETag']) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Request the rss feed again -- with an INM header to test 304. etag = browser.headers['ETag'] @@ -297,7 +357,8 @@ def test_content_feeds(self): browser.addHeader('If-None-Match', etag) browser.open(self.portal.absolute_url() + '/RSS') self.assertEqual('plone.content.feed', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.moderateCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.moderateCaching', + browser.headers['X-Cache-Operation']) # This should be a 304 response self.assertEqual('304 Not Modified', browser.headers['Status']) self.assertEqual('', browser.contents) @@ -305,21 +366,28 @@ def test_content_feeds(self): # Request the authenticated rss feed now = stable_now() browser = Browser(self.app) - browser.addHeader('Authorization', 'Basic %s:%s' % (TEST_USER_NAME, TEST_USER_PASSWORD,)) + browser.addHeader('Authorization', 'Basic %s:%s' % + (TEST_USER_NAME, TEST_USER_PASSWORD,)) browser.open(self.portal.absolute_url() + '/RSS') self.assertEqual('plone.content.feed', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.moderateCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.moderateCaching', + browser.headers['X-Cache-Operation']) # This should use cacheInBrowser - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertEqual('"|test_user_1_|%d|en|%s"' % (catalog.getCounter(), skins_tool.default_skin), browser.headers['ETag']) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + self.assertEqual('"|test_user_1_|%d|en|%s"' % ( + catalog.getCounter(), skins_tool.default_skin), browser.headers['ETag']) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Request the authenticated rss feed again -- to test RAM cache browser = Browser(self.app) - browser.addHeader('Authorization', 'Basic %s:%s' % (TEST_USER_NAME, TEST_USER_PASSWORD,)) + browser.addHeader('Authorization', 'Basic %s:%s' % + (TEST_USER_NAME, TEST_USER_PASSWORD,)) browser.open(self.portal.absolute_url() + '/RSS') self.assertEqual('plone.content.feed', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.moderateCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.moderateCaching', + browser.headers['X-Cache-Operation']) # Authenticated should NOT be RAM cached self.assertEqual(None, browser.headers.get('X-RAMCache')) @@ -339,52 +407,69 @@ def test_content_files(self): self.portal['f1']['i1'].image = test_image() self.portal['f1']['i1'].reindexObject() - import transaction; transaction.commit() + import transaction + transaction.commit() # Request the image with Manager role now = stable_now() browser = Browser(self.app) - browser.addHeader('Authorization', 'Basic %s:%s' % (SITE_OWNER_NAME, SITE_OWNER_PASSWORD,)) + browser.addHeader('Authorization', 'Basic %s:%s' % + (SITE_OWNER_NAME, SITE_OWNER_PASSWORD,)) browser.open(self.portal['f1']['i1'].absolute_url()) self.assertEqual('plone.content.file', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.moderateCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.moderateCaching', + browser.headers['X-Cache-Operation']) # Folder not published yet so image should not be cached in proxy # so this should use cacheInBrowser - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertFalse(None == browser.headers.get('Last-Modified')) # remove this when the next line works + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + # remove this when the next line works + self.assertFalse(None == browser.headers.get('Last-Modified')) #self.assertEqual('---lastmodified---', browser.headers['Last-Modified']) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Request an image scale with Manager role now = stable_now() browser = Browser(self.app) - browser.addHeader('Authorization', 'Basic %s:%s' % (SITE_OWNER_NAME, SITE_OWNER_PASSWORD,)) - browser.open(self.portal['f1']['i1'].absolute_url() + '/@@images/image/preview') + browser.addHeader('Authorization', 'Basic %s:%s' % + (SITE_OWNER_NAME, SITE_OWNER_PASSWORD,)) + browser.open(self.portal['f1'][ + 'i1'].absolute_url() + '/@@images/image/preview') self.assertEqual('plone.content.file', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.moderateCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.moderateCaching', + browser.headers['X-Cache-Operation']) # Folder not published yet so image scale should not be cached in proxy # so this should use cacheInBrowser - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertFalse(None == browser.headers.get('Last-Modified')) # remove this when the next line works + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + # remove this when the next line works + self.assertFalse(None == browser.headers.get('Last-Modified')) #self.assertEqual('---lastmodified---', browser.headers['Last-Modified']) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Publish the folder self.portal.portal_workflow.doActionFor(self.portal['f1'], 'publish') - import transaction; transaction.commit() + import transaction + transaction.commit() # Request the image now = stable_now() browser = Browser(self.app) browser.open(self.portal['f1']['i1'].absolute_url()) self.assertEqual('plone.content.file', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.moderateCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.moderateCaching', + browser.headers['X-Cache-Operation']) # Now visible to anonymous so this should use cacheInProxy - self.assertEqual('max-age=0, s-maxage=86400, must-revalidate', browser.headers['Cache-Control']) - self.assertFalse(None == browser.headers.get('Last-Modified')) # remove this when the next line works + self.assertEqual('max-age=0, s-maxage=86400, must-revalidate', + browser.headers['Cache-Control']) + # remove this when the next line works + self.assertFalse(None == browser.headers.get('Last-Modified')) #self.assertEqual('---lastmodified---', browser.headers['Last-Modified']) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Request the image again -- with an IMS header to test 304 lastmodified = browser.headers['Last-Modified'] @@ -393,7 +478,8 @@ def test_content_files(self): browser.addHeader('If-Modified-Since', lastmodified) browser.open(self.portal['f1']['i1'].absolute_url()) self.assertEqual('plone.content.file', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.moderateCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.moderateCaching', + browser.headers['X-Cache-Operation']) # This should be a 304 response self.assertEqual('304 Not Modified', browser.headers['Status']) self.assertEqual('', browser.contents) @@ -401,30 +487,39 @@ def test_content_files(self): # Request an image scale now = stable_now() browser = Browser(self.app) - browser.open(self.portal['f1']['i1'].absolute_url() + '/@@images/image/preview') + browser.open(self.portal['f1'][ + 'i1'].absolute_url() + '/@@images/image/preview') self.assertEqual('plone.content.file', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.moderateCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.moderateCaching', + browser.headers['X-Cache-Operation']) # Now visible to anonymous so this should use cacheInProxy - self.assertEqual('max-age=0, s-maxage=86400, must-revalidate', browser.headers['Cache-Control']) - self.assertFalse(None == browser.headers.get('Last-Modified')) # remove this when the next line works + self.assertEqual('max-age=0, s-maxage=86400, must-revalidate', + browser.headers['Cache-Control']) + # remove this when the next line works + self.assertFalse(None == browser.headers.get('Last-Modified')) #self.assertEqual('---lastmodified---', browser.headers['Last-Modified']) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) def test_resources(self): # This is a clone of the same test for 'without-caching-proxy' # Can we just call that test from this context? - import transaction; transaction.commit() + import transaction + transaction.commit() # Request a skin image now = stable_now() browser = Browser(self.app) browser.open(self.portal.absolute_url() + '/rss.png') self.assertEqual('plone.resource', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.strongCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.strongCaching', + browser.headers['X-Cache-Operation']) # This should use cacheInBrowserAndProxy - self.assertEqual('max-age=86400, proxy-revalidate, public', browser.headers['Cache-Control']) - self.assertFalse(None == browser.headers.get('Last-Modified')) # remove this when the next line works + self.assertEqual('max-age=86400, proxy-revalidate, public', + browser.headers['Cache-Control']) + # remove this when the next line works + self.assertFalse(None == browser.headers.get('Last-Modified')) #self.assertEqual('---lastmodified---', browser.headers['Last-Modified']) timedelta = dateutil.parser.parse(browser.headers['Expires']) - now self.assertTrue(timedelta > datetime.timedelta(seconds=86390)) @@ -436,7 +531,8 @@ def test_resources(self): browser.addHeader('If-Modified-Since', lastmodified) browser.open(self.portal.absolute_url() + '/rss.png') self.assertEqual('plone.resource', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.strongCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.strongCaching', + browser.headers['X-Cache-Operation']) # This should be a 304 response self.assertEqual('304 Not Modified', browser.headers['Status']) self.assertEqual('', browser.contents) @@ -448,17 +544,22 @@ def test_resources(self): # large Resource Registry cooked files, which all use the same # method to initiate a streamed response. s = "a" * (1 << 16) * 3 - self.portal.manage_addFile('bigfile', file=StringIO(s), content_type='application/octet-stream') + self.portal.manage_addFile('bigfile', file=StringIO( + s), content_type='application/octet-stream') - import transaction; transaction.commit() + import transaction + transaction.commit() browser = Browser(self.app) browser.open(self.portal['bigfile'].absolute_url()) self.assertEqual('plone.resource', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.strongCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.strongCaching', + browser.headers['X-Cache-Operation']) # This should use cacheInBrowserAndProxy - self.assertEqual('max-age=86400, proxy-revalidate, public', browser.headers['Cache-Control']) - self.assertFalse(None == browser.headers.get('Last-Modified')) # remove this when the next line works + self.assertEqual('max-age=86400, proxy-revalidate, public', + browser.headers['Cache-Control']) + # remove this when the next line works + self.assertFalse(None == browser.headers.get('Last-Modified')) #self.assertEqual('---lastmodified---', browser.headers['Last-Modified']) timedelta = dateutil.parser.parse(browser.headers['Expires']) - now self.assertTrue(timedelta > datetime.timedelta(seconds=86390)) diff --git a/plone/app/caching/tests/test_profile_without_caching_proxy.py b/plone/app/caching/tests/test_profile_without_caching_proxy.py index f8848a1..fedb42a 100644 --- a/plone/app/caching/tests/test_profile_without_caching_proxy.py +++ b/plone/app/caching/tests/test_profile_without_caching_proxy.py @@ -28,12 +28,14 @@ import urllib -TEST_FILE = pkg_resources.resource_filename('plone.app.caching.tests', 'test.gif') +TEST_FILE = pkg_resources.resource_filename( + 'plone.app.caching.tests', 'test.gif') def test_image(): from plone.namedfile.file import NamedBlobImage - filename = pkg_resources.resource_filename('plone.app.caching.tests', 'test.gif') + filename = pkg_resources.resource_filename( + 'plone.app.caching.tests', 'test.gif') filename = os.path.join(os.path.dirname(__file__), u'test.gif') return NamedBlobImage( data=open(filename, 'r').read(), @@ -62,7 +64,8 @@ def setUp(self): self.app = self.layer['app'] self.portal = self.layer['portal'] - test_css = FSFile('test.css', os.path.join(os.path.dirname(__file__), 'test.css')) + test_css = FSFile('test.css', os.path.join( + os.path.dirname(__file__), 'test.css')) self.portal.portal_skins.custom._setOb('test.css', test_css) setRequest(self.portal.REQUEST) @@ -72,8 +75,10 @@ def setUp(self): self.registry = getUtility(IRegistry) self.cacheSettings = self.registry.forInterface(ICacheSettings) - self.cachePurgingSettings = self.registry.forInterface(ICachePurgingSettings) - self.ploneCacheSettings = self.registry.forInterface(IPloneCacheSettings) + self.cachePurgingSettings = self.registry.forInterface( + ICachePurgingSettings) + self.ploneCacheSettings = self.registry.forInterface( + IPloneCacheSettings) self.cacheSettings.enabled = True @@ -106,7 +111,8 @@ def test_composite_views(self): # Publish the folder and page self.portal.portal_workflow.doActionFor(self.portal['f1'], 'publish') - self.portal.portal_workflow.doActionFor(self.portal['f1']['d1'], 'publish') + self.portal.portal_workflow.doActionFor( + self.portal['f1']['d1'], 'publish') # Should we set up the etag components? # - set member? No @@ -117,56 +123,78 @@ def test_composite_views(self): # - leave status unlocked # - set the mod date on the resource registries? Probably. - import transaction; transaction.commit() + import transaction + transaction.commit() # Request the quthenticated folder now = stable_now() browser = Browser(self.app) - browser.addHeader('Authorization', 'Basic %s:%s' % (TEST_USER_NAME, TEST_USER_PASSWORD,)) + browser.addHeader('Authorization', 'Basic %s:%s' % + (TEST_USER_NAME, TEST_USER_PASSWORD,)) browser.open(self.portal['f1'].absolute_url()) - self.assertEqual('plone.content.folderView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.folderView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # This should use cacheInBrowser - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertEqual('"|test_user_1_|%d|en|%s|0|0' % (catalog.getCounter(), default_skin), _normalize_etag(browser.headers['ETag'])) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + self.assertEqual('"|test_user_1_|%d|en|%s|0|0' % ( + catalog.getCounter(), default_skin), _normalize_etag(browser.headers['ETag'])) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Set the copy/cut cookie and then request the folder view again browser.cookies.create('__cp', 'xxx') browser.open(self.portal['f1'].absolute_url()) # The response should be the same as before except for the etag - self.assertEqual('plone.content.folderView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertEqual('"|test_user_1_|%d|en|%s|0|1' % (catalog.getCounter(), default_skin), _normalize_etag(browser.headers['ETag'])) + self.assertEqual('plone.content.folderView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + self.assertEqual('"|test_user_1_|%d|en|%s|0|1' % ( + catalog.getCounter(), default_skin), _normalize_etag(browser.headers['ETag'])) # Request the authenticated page now = stable_now() browser = Browser(self.app) - browser.addHeader('Authorization', 'Basic %s:%s' % (TEST_USER_NAME, TEST_USER_PASSWORD,)) + browser.addHeader('Authorization', 'Basic %s:%s' % + (TEST_USER_NAME, TEST_USER_PASSWORD,)) browser.open(self.portal['f1']['d1'].absolute_url()) self.assertTrue(testText in browser.contents) - self.assertEqual('plone.content.itemView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.itemView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # This should use cacheInBrowser - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertEqual('"|test_user_1_|%d|en|%s|0' % (catalog.getCounter(), default_skin), _normalize_etag(browser.headers['ETag'])) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + self.assertEqual('"|test_user_1_|%d|en|%s|0' % ( + catalog.getCounter(), default_skin), _normalize_etag(browser.headers['ETag'])) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Request the authenticated page again -- to test RAM cache. browser = Browser(self.app) - browser.addHeader('Authorization', 'Basic %s:%s' % (TEST_USER_NAME, TEST_USER_PASSWORD,)) + browser.addHeader('Authorization', 'Basic %s:%s' % + (TEST_USER_NAME, TEST_USER_PASSWORD,)) browser.open(self.portal['f1']['d1'].absolute_url()) - self.assertEqual('plone.content.itemView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.itemView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # Authenticated should NOT be RAM cached self.assertEqual(None, browser.headers.get('X-RAMCache')) - # Request the authenticated page again -- with an INM header to test 304 + # Request the authenticated page again -- with an INM header to test + # 304 etag = browser.headers['ETag'] browser = Browser(self.app) browser.raiseHttpErrors = False # we really do want to see the 304 - browser.addHeader('Authorization', 'Basic %s:%s' % (TEST_USER_NAME, TEST_USER_PASSWORD,)) + browser.addHeader('Authorization', 'Basic %s:%s' % + (TEST_USER_NAME, TEST_USER_PASSWORD,)) browser.addHeader('If-None-Match', etag) browser.open(self.portal['f1']['d1'].absolute_url()) # This should be a 304 response @@ -177,38 +205,54 @@ def test_composite_views(self): now = stable_now() browser = Browser(self.app) browser.open(self.portal['f1'].absolute_url()) - self.assertEqual('plone.content.folderView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.folderView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # This should use cacheInBrowser - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertEqual('"||%d|en|%s|0|0' % (catalog.getCounter(), default_skin), _normalize_etag(browser.headers['ETag'])) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + self.assertEqual('"||%d|en|%s|0|0' % ( + catalog.getCounter(), default_skin), _normalize_etag(browser.headers['ETag'])) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Request the anonymous page now = stable_now() browser = Browser(self.app) browser.open(self.portal['f1']['d1'].absolute_url()) - self.assertEqual('plone.content.itemView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.itemView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) self.assertTrue(testText in browser.contents) # This should use cacheInBrowser - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertEqual('"||%d|en|%s|0' % (catalog.getCounter(), default_skin), _normalize_etag(browser.headers['ETag'])) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + self.assertEqual('"||%d|en|%s|0' % ( + catalog.getCounter(), default_skin), _normalize_etag(browser.headers['ETag'])) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Request the anonymous page again -- to test RAM cache. # Anonymous should be RAM cached now = stable_now() browser = Browser(self.app) browser.open(self.portal['f1']['d1'].absolute_url()) - self.assertEqual('plone.content.itemView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.itemView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # This should come from RAM cache - self.assertEqual('plone.app.caching.operations.ramcache', browser.headers['X-RAMCache']) + self.assertEqual('plone.app.caching.operations.ramcache', + browser.headers['X-RAMCache']) self.assertTrue(testText in browser.contents) - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertEqual('"||%d|en|%s|0'% (catalog.getCounter(), default_skin), _normalize_etag(browser.headers['ETag'])) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + self.assertEqual('"||%d|en|%s|0' % ( + catalog.getCounter(), default_skin), _normalize_etag(browser.headers['ETag'])) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Request the anonymous page again -- with an INM header to test 304. etag = browser.headers['ETag'] @@ -216,8 +260,10 @@ def test_composite_views(self): browser.raiseHttpErrors = False browser.addHeader('If-None-Match', etag) browser.open(self.portal['f1']['d1'].absolute_url()) - self.assertEqual('plone.content.itemView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.itemView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # This should be a 304 response self.assertEqual('304 Not Modified', browser.headers['Status']) self.assertEqual('', browser.contents) @@ -233,13 +279,16 @@ def test_composite_views(self): transaction.commit() - # Request the anonymous page again -- to test expiration of 304 and RAM. + # Request the anonymous page again -- to test expiration of 304 and + # RAM. etag = browser.headers['ETag'] browser = Browser(self.app) browser.addHeader('If-None-Match', etag) browser.open(self.portal['f1']['d1'].absolute_url()) - self.assertEqual('plone.content.itemView', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.content.itemView', + browser.headers['X-Cache-Rule']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # The etag has changed so we should get a fresh page. self.assertEqual(None, browser.headers.get('X-RAMCache')) self.assertEqual('200 Ok', browser.headers['Status']) @@ -255,18 +304,23 @@ def test_content_feeds(self): self.syndication.editProperties(isAllowed=True) self.syndication.enableSyndication(self.portal) - import transaction; transaction.commit() + import transaction + transaction.commit() # Request the rss feed now = stable_now() browser = Browser(self.app) browser.open(self.portal.absolute_url() + '/RSS') self.assertEqual('plone.content.feed', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # This should use cacheInBrowser - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertEqual('"||%d|en|%s"' % (catalog.getCounter(), default_skin), browser.headers['ETag']) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + self.assertEqual('"||%d|en|%s"' % (catalog.getCounter(), + default_skin), browser.headers['ETag']) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Request the rss feed again -- to test RAM cache now = stable_now() @@ -274,13 +328,18 @@ def test_content_feeds(self): browser = Browser(self.app) browser.open(self.portal.absolute_url() + '/RSS') self.assertEqual('plone.content.feed', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # This should come from RAM cache - self.assertEqual('plone.app.caching.operations.ramcache', browser.headers['X-RAMCache']) + self.assertEqual('plone.app.caching.operations.ramcache', + browser.headers['X-RAMCache']) self.assertEqual(rssText, browser.contents) - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertEqual('"||%d|en|%s"' % (catalog.getCounter(), default_skin), browser.headers['ETag']) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + self.assertEqual('"||%d|en|%s"' % (catalog.getCounter(), + default_skin), browser.headers['ETag']) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Request the rss feed again -- with an INM header to test 304. etag = browser.headers['ETag'] @@ -289,7 +348,8 @@ def test_content_feeds(self): browser.addHeader('If-None-Match', etag) browser.open(self.portal.absolute_url() + '/RSS') self.assertEqual('plone.content.feed', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # This should be a 304 response self.assertEqual('304 Not Modified', browser.headers['Status']) self.assertEqual('', browser.contents) @@ -297,21 +357,28 @@ def test_content_feeds(self): # Request the authenticated rss feed now = stable_now() browser = Browser(self.app) - browser.addHeader('Authorization', 'Basic %s:%s' % (TEST_USER_NAME, TEST_USER_PASSWORD,)) + browser.addHeader('Authorization', 'Basic %s:%s' % + (TEST_USER_NAME, TEST_USER_PASSWORD,)) browser.open(self.portal.absolute_url() + '/RSS') self.assertEqual('plone.content.feed', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # This should use cacheInBrowser - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertEqual('"|test_user_1_|%d|en|%s"' % (catalog.getCounter(), default_skin), browser.headers['ETag']) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + self.assertEqual('"|test_user_1_|%d|en|%s"' % ( + catalog.getCounter(), default_skin), browser.headers['ETag']) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Request the authenticated rss feed again -- to test RAM cache browser = Browser(self.app) - browser.addHeader('Authorization', 'Basic %s:%s' % (TEST_USER_NAME, TEST_USER_PASSWORD,)) + browser.addHeader('Authorization', 'Basic %s:%s' % + (TEST_USER_NAME, TEST_USER_PASSWORD,)) browser.open(self.portal.absolute_url() + '/RSS') self.assertEqual('plone.content.feed', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # Authenticated should NOT be RAM cached self.assertEqual(None, browser.headers.get('X-RAMCache')) @@ -334,19 +401,24 @@ def test_content_files(self): # Publish the folder self.portal.portal_workflow.doActionFor(self.portal['f1'], 'publish') - import transaction; transaction.commit() + import transaction + transaction.commit() # Request the image now = stable_now() browser = Browser(self.app) browser.open(self.portal['f1']['i1'].absolute_url()) self.assertEqual('plone.content.file', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # This should use cacheInBrowser - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertFalse(None == browser.headers.get('Last-Modified')) # remove this when the next line works + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + # remove this when the next line works + self.assertFalse(None == browser.headers.get('Last-Modified')) #self.assertEqual('---lastmodified---', browser.headers['Last-Modified']) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) # Request the image again -- with an IMS header to test 304 lastmodified = browser.headers['Last-Modified'] @@ -355,7 +427,8 @@ def test_content_files(self): browser.addHeader('If-Modified-Since', lastmodified) browser.open(self.portal['f1']['i1'].absolute_url()) self.assertEqual('plone.content.file', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # This should be a 304 response self.assertEqual('304 Not Modified', browser.headers['Status']) self.assertEqual('', browser.contents) @@ -363,28 +436,37 @@ def test_content_files(self): # Request an image scale now = stable_now() browser = Browser(self.app) - browser.open(self.portal['f1']['i1'].absolute_url() + '/@@images/image/preview') + browser.open(self.portal['f1'][ + 'i1'].absolute_url() + '/@@images/image/preview') self.assertEqual('plone.content.file', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.weakCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.weakCaching', + browser.headers['X-Cache-Operation']) # This should use cacheInBrowser - self.assertEqual('max-age=0, must-revalidate, private', browser.headers['Cache-Control']) - self.assertFalse(None == browser.headers.get('Last-Modified')) # remove this when the next line works + self.assertEqual('max-age=0, must-revalidate, private', + browser.headers['Cache-Control']) + # remove this when the next line works + self.assertFalse(None == browser.headers.get('Last-Modified')) #self.assertEqual('---lastmodified---', browser.headers['Last-Modified']) - self.assertTrue(now > dateutil.parser.parse(browser.headers['Expires'])) + self.assertTrue(now > dateutil.parser.parse( + browser.headers['Expires'])) def test_resources(self): - import transaction; transaction.commit() + import transaction + transaction.commit() # Request a skin image now = stable_now() browser = Browser(self.app) browser.open(self.portal.absolute_url() + '/rss.png') self.assertEqual('plone.resource', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.strongCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.strongCaching', + browser.headers['X-Cache-Operation']) # This should use cacheInBrowserAndProxy - self.assertEqual('max-age=86400, proxy-revalidate, public', browser.headers['Cache-Control']) - self.assertFalse(None == browser.headers.get('Last-Modified')) # remove this when the next line works + self.assertEqual('max-age=86400, proxy-revalidate, public', + browser.headers['Cache-Control']) + # remove this when the next line works + self.assertFalse(None == browser.headers.get('Last-Modified')) #self.assertEqual('---lastmodified---', browser.headers['Last-Modified']) timedelta = dateutil.parser.parse(browser.headers['Expires']) - now self.assertTrue(timedelta > datetime.timedelta(seconds=86390)) @@ -396,7 +478,8 @@ def test_resources(self): browser.addHeader('If-Modified-Since', lastmodified) browser.open(self.portal.absolute_url() + '/rss.png') self.assertEqual('plone.resource', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.strongCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.strongCaching', + browser.headers['X-Cache-Operation']) # This should be a 304 response self.assertEqual('304 Not Modified', browser.headers['Status']) self.assertEqual('', browser.contents) @@ -408,17 +491,22 @@ def test_resources(self): # large Resource Registry cooked files, which all use the same # method to initiate a streamed response. s = "a" * (1 << 16) * 3 - self.portal.manage_addFile('bigfile', file=StringIO(s), content_type='application/octet-stream') + self.portal.manage_addFile('bigfile', file=StringIO( + s), content_type='application/octet-stream') - import transaction; transaction.commit() + import transaction + transaction.commit() browser = Browser(self.app) browser.open(self.portal['bigfile'].absolute_url()) self.assertEqual('plone.resource', browser.headers['X-Cache-Rule']) - self.assertEqual('plone.app.caching.strongCaching', browser.headers['X-Cache-Operation']) + self.assertEqual('plone.app.caching.strongCaching', + browser.headers['X-Cache-Operation']) # This should use cacheInBrowserAndProxy - self.assertEqual('max-age=86400, proxy-revalidate, public', browser.headers['Cache-Control']) - self.assertFalse(None == browser.headers.get('Last-Modified')) # remove this when the next line works + self.assertEqual('max-age=86400, proxy-revalidate, public', + browser.headers['Cache-Control']) + # remove this when the next line works + self.assertFalse(None == browser.headers.get('Last-Modified')) #self.assertEqual('---lastmodified---', browser.headers['Last-Modified']) timedelta = dateutil.parser.parse(browser.headers['Expires']) - now self.assertTrue(timedelta > datetime.timedelta(seconds=86390)) diff --git a/plone/app/caching/tests/test_purge.py b/plone/app/caching/tests/test_purge.py index 3d046bf..bac4b7a 100644 --- a/plone/app/caching/tests/test_purge.py +++ b/plone/app/caching/tests/test_purge.py @@ -45,9 +45,11 @@ def __init__(self): def handler(self, event): self.invocations.append(event) + class FauxRequest(dict): pass + @implementer(IContentish) class FauxNonContent(Explicit): @@ -68,6 +70,7 @@ def virtual_url_path(self): def getPhysicalPath(self): return ('', ) + @implementer(IBrowserDefault) class FauxContent(FauxNonContent): @@ -76,10 +79,12 @@ class FauxContent(FauxNonContent): def defaultView(self): return 'default-view' + @implementer(IDiscussionResponse) class FauxDiscussable(Explicit): pass + class TestPurgeRedispatch(unittest.TestCase): layer = UNIT_TESTING @@ -151,6 +156,7 @@ def test_removed(self): self.assertEqual(1, len(self.handler.invocations)) self.assertEqual(context, self.handler.invocations[0].object) + class TestContentPurgePaths(unittest.TestCase): layer = UNIT_TESTING @@ -188,6 +194,7 @@ def test_parent_default_view(self): list(purger.getRelativePaths())) self.assertEqual([], list(purger.getAbsolutePaths())) + class TestDiscussionItemPurgePaths(unittest.TestCase): layer = UNIT_TESTING @@ -281,6 +288,7 @@ def getDiscussionThread(self, item): self.assertEqual(['/app/foo'], list(purge.getRelativePaths())) self.assertEqual(['/purgeme'], list(purge.getAbsolutePaths())) + class TestObjectFieldPurgePaths(unittest.TestCase): maxDiff = None @@ -308,7 +316,8 @@ class ATMultipleFields(atapi.BaseContent): atapi.StringField('foo'), atapi.FileField('file1'), atapi.ImageField('image1'), - atapi.ImageField('image2', sizes={'mini': (50, 50), 'normal': (100, 100)}), + atapi.ImageField('image2', sizes={ + 'mini': (50, 50), 'normal': (100, 100)}), BlobField('blob1'), )) @@ -331,7 +340,8 @@ class ATMultipleFields(atapi.BaseContent): atapi.StringField('foo'), atapi.FileField('file1'), atapi.ImageField('image1'), - atapi.ImageField('image2', sizes={'mini': (50, 50), 'normal': (100, 100)}), + atapi.ImageField('image2', sizes={ + 'mini': (50, 50), 'normal': (100, 100)}), atapi.TextField('text'), )) diff --git a/plone/app/caching/tests/test_utils.py b/plone/app/caching/tests/test_utils.py index 925baec..6d0c3ec 100644 --- a/plone/app/caching/tests/test_utils.py +++ b/plone/app/caching/tests/test_utils.py @@ -41,16 +41,18 @@ def __init__(self, portal_type='testtype', defaultView='defaultView'): def defaultView(self): return self._defaultView + class DummyNotContent(Explicit): pass + class DummyFTI(object): def __init__(self, portal_type, viewAction=''): self.id = portal_type self._actions = { - 'object/view': {'url': viewAction}, - } + 'object/view': {'url': viewAction}, + } def getActionInfo(self, name): return self._actions[name] @@ -62,6 +64,7 @@ def queryMethodID(self, id, default=None, context=None): return '@@defaultView' return default + @implementer(IDynamicType) class DummyNotBrowserDefault(Explicit): @@ -72,6 +75,7 @@ def __init__(self, portal_type='testtype', viewAction=''): def getTypeInfo(self): return DummyFTI(self.portal_type, self._viewAction) + class TestIsPurged(unittest.TestCase): layer = UNIT_TESTING @@ -121,6 +125,7 @@ def test_listed(self): content = DummyContent() self.assertEqual(True, isPurged(content)) + class TestGetObjectDefaultPath(unittest.TestCase): layer = UNIT_TESTING @@ -138,7 +143,8 @@ def test_browserviewdefault(self): self.assertEqual('defaultView', getObjectDefaultView(context)) def test_not_IBrowserDefault_methodid(self): - context = DummyNotBrowserDefault('testtype', 'string:${object_url}/view') + context = DummyNotBrowserDefault( + 'testtype', 'string:${object_url}/view') self.assertEqual('defaultView', getObjectDefaultView(context)) def test_not_IBrowserDefault_default_method(self): @@ -146,5 +152,6 @@ def test_not_IBrowserDefault_default_method(self): self.assertEqual('defaultView', getObjectDefaultView(context)) def test_not_IBrowserDefault_actiononly(self): - context = DummyNotBrowserDefault('testtype', 'string:${object_url}/defaultView') + context = DummyNotBrowserDefault( + 'testtype', 'string:${object_url}/defaultView') self.assertEqual('defaultView', getObjectDefaultView(context)) diff --git a/plone/app/caching/utils.py b/plone/app/caching/utils.py index 9317c05..83a5a64 100644 --- a/plone/app/caching/utils.py +++ b/plone/app/caching/utils.py @@ -27,6 +27,7 @@ def isPurged(object): return (portal_type in settings.purgedContentTypes) + def stripLeadingCharacters(name): """Strip off leading / and/or @@ """ @@ -38,6 +39,7 @@ def stripLeadingCharacters(name): return name + def getObjectDefaultView(context): """Get the id of an object's default view """ @@ -59,7 +61,7 @@ def getObjectDefaultView(context): fti = context.getTypeInfo() try: # XXX: This isn't quite right since it assumes the action starts - #with ${object_url} + # with ${object_url} action = fti.getActionInfo('object/view')['url'].split('/')[-1] except ValueError: # If the action doesn't exist, stop @@ -67,9 +69,9 @@ def getObjectDefaultView(context): # Try resolving method aliases because we need a real template_id here if action: - action = fti.queryMethodID(action, default = action, context = context) + action = fti.queryMethodID(action, default=action, context=context) else: - action = fti.queryMethodID('(Default)', default = action, - context = context) + action = fti.queryMethodID('(Default)', default=action, + context=context) return stripLeadingCharacters(action)