def test_dont_swallow_conflict_error(self):

        @implementer(IRulesetLookup)
        @adapter(Interface, Interface)
        class DummyRulesetLookup(object):

            def __init__(self, published, request):
                self.published = published
                self.request = request

            def __call__(self):
                raise ConflictError()

        provideAdapter(DummyRulesetLookup)

        provideUtility(Registry(), IRegistry)
        registry = getUtility(IRegistry)
        registry.registerInterface(ICacheSettings)
        settings = registry.forInterface(ICacheSettings)
        settings.enabled = True
        settings.operationMapping = {'foo': 'bar'}

        view = DummyView()
        request = DummyRequest(view, DummyResponse())
        self.assertRaises(ConflictError, intercept, DummyEvent(request))

    def test_locale_semantics(self):
        from zope.component import provideAdapter
        from zope.publisher.browser import BrowserLanguages
        from zope.publisher.interfaces.http import IHTTPRequest
        from zope.i18n.interfaces import IUserPreferredLanguages
        from zope.i18n.interfaces.locales import ILocale

        provideAdapter(BrowserLanguages, [IHTTPRequest],
                       IUserPreferredLanguages)
        env_ = {'HTTP_ACCEPT_LANGUAGE': 'en'}

        # we should also test the correct semantics of the locale
        for httplang in ('it', 'it-ch', 'it-CH', 'IT', 'IT-CH', 'IT-ch'):
            env = env_.copy()
            env['HTTP_ACCEPT_LANGUAGE'] = httplang
            request = self._makeOne(environ=env)
            locale = request.locale
            self.assert_(ILocale.providedBy(locale))
            parts = httplang.split('-')
            lang = parts.pop(0).lower()
            territory = variant = None
            if parts:
                territory = parts.pop(0).upper()
            if parts:
                variant = parts.pop(0).upper()
            self.assertEqual(locale.id.language, lang)
            self.assertEqual(locale.id.territory, territory)
            self.assertEqual(locale.id.variant, variant)

    def test_anonymous_can_not_add_comments_if_discussion_is_not_allowed(self):
        """Make sure that anonymous users can't post comments if anonymous
           comments are disabled.
        """

        # Anonymous comments are disabled by default

        logout()

        def make_request(form={}):
            request = TestRequest()
            request.form.update(form)
            alsoProvides(request, IFormLayer)
            alsoProvides(request, IAttributeAnnotatable)
            return request

        provideAdapter(adapts=(Interface, IBrowserRequest),
                       provides=Interface,
                       factory=CommentForm,
                       name=u"comment-form")

        request = make_request(form={'form.widgets.text': u'bar'})

        commentForm = getMultiAdapter((self.context, request),
                                      name=u"comment-form")
        commentForm.update()
        data, errors = commentForm.extractData()  # pylint: disable-msg=W0612

        self.assertEqual(len(errors), 0)
        self.assertRaises(
            Unauthorized,
            commentForm.handleComment,
            commentForm,
            "foo"
        )

    def test_LastModified(self):
        from plone.app.caching.operations.etags import LastModified

        mod = datetime(2010, 1, 2, 3, 4, 5, 6, tzlocal())
        utcStamp = time.mktime(mod.utctimetuple())

        class DummyLastModified(object):
            implements(ILastModified)
            adapts(DummyPublished)

            def __init__(self, context):
                self.context = context

            def __call__(self):
                return mod

        provideAdapter(DummyLastModified)

        environ = {'SERVER_NAME': 'example.com', 'SERVER_PORT': '80'}
        response = HTTPResponse()
        request = HTTPRequest(StringIO(), environ, response)
        published = DummyPublished(DummyContext())

        etag = LastModified(published, request)
        self.assertEqual(str(utcStamp), etag())

    def test_UserID_member(self):
        from plone.app.caching.operations.etags import UserID

        class DummyMember(object):

            def getId(self):
                return 'bob'

        class DummyPortalState(object):
            implements(Interface)
            adapts(DummyContext, Interface)

            def __init__(self, context, request):
                pass

            def member(self):
                return DummyMember()

        provideAdapter(DummyPortalState, name=u"plone_portal_state")

        environ = {'SERVER_NAME': 'example.com', 'SERVER_PORT': '80'}
        response = HTTPResponse()
        request = HTTPRequest(StringIO(), environ, response)
        published = DummyPublished(DummyContext())

        etag = UserID(published, request)

        self.assertEqual('bob', etag())

def registerPublisherForFTI(fti):
    # This registers a publisher that will allow to traverse to each sql item
    if not getattr(fti, 'sql_table', None):
        return
    name = getattr(fti, 'sql_folder_id', fti.id)
    has_folder = False
    if name and IRelationValue.providedBy(name):
        obj = name.to_object
        if obj:
            has_folder = True
            name = obj.getId()
        else:
            name = fti.context.getId()
    elif name and name.startswith('/'):
        portal = getToolByName(getSite(), "portal_url").getPortalObject()
        try:
            folder = portal.restrictedTraverse(name)
            has_folder = True
        except:
            pass
    elif not name:
        name = fti.context.getId()
    if not has_folder:
        view = queryMultiAdapter((None, ICollectiveBehaviorSQLLayer), IBrowserView, name='data-'+name, default=None)
        if view != None:
            return view
        publisher = SQLItemPublisher
        provideAdapter(
            factory=publisher,
            adapts=(None, ICollectiveBehaviorSQLLayer),
            provides=IBrowserView,
            name='data-'+name)

        LOG.info('Publisher registered for data-'+name)

def _setUpDefaultTraversable():
    from zope.interface import Interface
    from zope.component import provideAdapter
    from zope.traversing.interfaces import ITraversable
    from zope.traversing.adapters import DefaultTraversable

    provideAdapter(DefaultTraversable, (Interface,), ITraversable)

def pdfSetUp(test=None):
    setup.placefulSetUp()
    sbsetup.setUpCalendaring()
    app = ApplicationStub()
    provideAdapter(lambda x: app, (None,), ISchoolToolApplication)
    provideAdapter(getApplicationPreferences,
                   (ISchoolToolApplication,), IApplicationPreferences)

    def setUp(self):
        super(TestInvitationFormEmailing, self).setUp()
        self.login_as_portal_owner()
        provideAdapter(adapts=(Interface, IBrowserRequest),
                       provides=Interface,
                       factory=InviteForm,
                       name="invite")

        # create a workspace
        self.ws = api.content.create(
            self.portal,
            "ploneintranet.workspace.workspacefolder",
            "alejandro-workspace",
            title="Alejandro workspace")

        self.app = self.layer['app']
        self.portal = self.layer['portal']
        self.request = self.layer['request']
        # don't import this at the top, because it screws up the setup
        # and none of the tests can run
        from Products.CMFPlone.tests.utils import MockMailHost
    #    # Mock the mail host so we can test sending the email
        mockmailhost = MockMailHost('MailHost')

        if not hasattr(mockmailhost, 'smtp_host'):
            mockmailhost.smtp_host = 'localhost'

        self.portal.MailHost = mockmailhost
        sm = self.portal.getSiteManager()
        sm.registerUtility(component=mockmailhost, provided=IMailHost)

        self.mailhost = api.portal.get_tool('MailHost')

        self.portal._updateProperty('email_from_name', 'Portal Owner')
        self.portal._updateProperty('email_from_address', '[email protected]')

    def test_match_abort(self):
        provideAdapter(DefaultRulesetLookup)
        provideUtility(Registry(), IRegistry)
        registry = getUtility(IRegistry)
        registry.registerInterface(ICacheSettings)
        settings = registry.forInterface(ICacheSettings)
        settings.enabled = True

        z3c.caching.registry.register(DummyView, 'testrule')
        settings.operationMapping = {'testrule': 'op1'}

        @implementer(ICachingOperation)
        @adapter(Interface, Interface)
        class DummyOperation(object):

            def __init__(self, published, request):
                self.published = published
                self.request = request

            def modifyResponse(self, rulename, response):
                pass

            def interceptResponse(self, rulename, response):
                response.addHeader('X-Cache-Foo', 'test')
                return None

        provideAdapter(DummyOperation, name='op1')

        view = DummyView()
        request = DummyRequest(view, DummyResponse())

        intercept(DummyEvent(request))
        self.assertEqual({'PUBLISHED': view}, dict(request))
        self.assertEqual({'X-Cache-Rule': ['testrule'],
                          'X-Cache-Foo': ['test']}, dict(request.response))

    def test_off_switch(self):
        provideAdapter(DefaultRulesetLookup)
        provideUtility(Registry(), IRegistry)
        registry = getUtility(IRegistry)
        registry.registerInterface(ICacheSettings)
        settings = registry.forInterface(ICacheSettings)
        settings.enabled = False

        z3c.caching.registry.register(DummyView, 'testrule')
        settings.operationMapping = {'testrule': 'op1'}

        @implementer(ICachingOperation)
        @adapter(Interface, Interface)
        class DummyOperation(object):

            def __init__(self, published, request):
                self.published = published
                self.request = request

            def interceptResponse(self, rulename, response):
                return None

            def modifyResponse(self, rulename, response):
                response['X-Mutated'] = rulename

        provideAdapter(DummyOperation, name='op1')

        view = DummyView()
        request = DummyRequest(view, DummyResponse())

        MutatorTransform(view, request).transformUnicode(u'', 'utf-8')

        self.assertEqual({'PUBLISHED': view}, dict(request))
        self.assertEqual({}, dict(request.response))

    def test_swallow_other_error(self):

        @implementer(IRulesetLookup)
        @adapter(Interface, Interface)
        class DummyRulesetLookup(object):

            def __init__(self, published, request):
                self.published = published
                self.request = request

            def __call__(self):
                raise AttributeError('Should be swallowed and logged')

        provideAdapter(DummyRulesetLookup)

        provideUtility(Registry(), IRegistry)
        registry = getUtility(IRegistry)
        registry.registerInterface(ICacheSettings)
        settings = registry.forInterface(ICacheSettings)
        settings.enabled = True
        settings.operationMapping = {'foo': 'bar'}

        view = DummyView()
        request = DummyRequest(view, DummyResponse())

        try:
            intercept(DummyEvent(request))
        except Exception:
            self.fail('Intercept should not raise')

    def test_request_not_annotatable(self):
        context = FauxContext()
        
        request = FauxRequest()
        setRequest(request)

        configlet= CachePurgingConfiglet()
        provideUtility(configlet, ICachePurgingConfiglet)

        settings = getUtility(ICachePurgingConfiglet)
        settings.enabled = True
        settings.cachingProxies = ('http://localhost:1234',)
        
        class FauxPurgePaths(object):
            implements(IPurgePaths)
            adapts(FauxContext)
        
            def __init__(self, context):
                self.context = context
        
            def getRelativePaths(self):
                return ['/foo', '/bar']
        
            def getAbsolutePaths(self):
                return []
        
        provideAdapter(FauxPurgePaths, name="test1")
        
        try:
            notify(Purge(context))
        except:
            self.fail()

    def test_retry_keeps_everything(self):
        """lowlevel test for retry (see #98440)"""
        from zope.publisher.browser import TestRequest
        from zope.publisher.skinnable import setDefaultSkin
        from zope.publisher.interfaces import IDefaultSkin
        from zope.publisher.interfaces.browser import IBrowserRequest
        from zope.publisher.interfaces.browser import IBrowserSkinType
        # create a retryable request
        request = TestRequest()
        self.assertTrue(request.supportsRetry())
        # create a skin and register it as the default skin
        class ISomeSkin(Interface):
            pass
        alsoProvides(ISomeSkin, IBrowserSkinType)
        provideAdapter(ISomeSkin, (IBrowserRequest,), IDefaultSkin)
        # set the default skin for the request
        setDefaultSkin(request)

        # create a retried request
        retried = request.retry()

        # the requests are not the same
        self.assertTrue(request is not retried)
        # the requests both provide the default skin
        self.assertTrue(ISomeSkin.providedBy(request))
        self.assertTrue(ISomeSkin.providedBy(retried))

    def test_Roles_member(self):
        from plone.app.caching.operations.etags import Roles

        class DummyMember(object):

            def getRolesInContext(self, context):
                return ['Member', 'Manager']

        class DummyPortalState(object):
            implements(Interface)
            adapts(DummyContext, Interface)

            def __init__(self, context, request):
                pass

            def anonymous(self):
                return False

            def member(self):
                return DummyMember()

        provideAdapter(DummyPortalState, name=u"plone_portal_state")

        environ = {'SERVER_NAME': 'example.com', 'SERVER_PORT': '80'}
        response = HTTPResponse()
        request = HTTPRequest(StringIO(), environ, response)
        published = DummyPublished(DummyContext())

        etag = Roles(published, request)

        self.assertEqual('Manager;Member', etag())

    def test_form_create(self):

        # Context and request
        context = Container(u"container")
        request = TestRequest()

        # FTI - returns dummy factory name

        fti_mock = DexterityFTI(u"testtype")
        fti_mock.factory = u'testfactory'
        self.mock_utility(fti_mock, IDexterityFTI, name=u"testtype")

        # The form we're testing
        form = DefaultAddForm(context, request)
        form.portal_type = u"testtype"

        class ISchema(Interface):
            foo = schema.TextLine()
        form.fields = Fields(ISchema)

        # createObject and applyChanges

        obj_dummy = Item(id="dummy")
        alsoProvides(obj_dummy, ISchema)
        data_dummy = {u"foo": u"bar"}

        from zope.component import createObject
        self.patch_global(createObject, return_value=obj_dummy)

        provideAdapter(AttributeField)

        self.assertEqual(obj_dummy, form.create(data_dummy))
        self.assertEqual("testtype", obj_dummy.portal_type)

    def test_CatalogCounter(self):
        from plone.app.caching.operations.etags import CatalogCounter

        class DummyCatalog(object):

            def getCounter(self):
                return 10

        class DummyTools(object):
            implements(Interface)
            adapts(DummyContext, Interface)

            def __init__(self, context, request):
                pass

            def catalog(self):
                return DummyCatalog()

        provideAdapter(DummyTools, name=u"plone_tools")

        environ = {'SERVER_NAME': 'example.com', 'SERVER_PORT': '80'}
        response = HTTPResponse()
        request = HTTPRequest(StringIO(), environ, response)
        published = DummyPublished(DummyContext())

        etag = CatalogCounter(published, request)

        self.assertEqual('10', etag())

    def test_off_switch(self):

        class Transform1(FauxTransformBase):

            order = 0

            def transformBytes(self, result, encoding):
                return result + ' transformed'

            def transformUnicode(self, result, encoding):
                return result + u' transformed'

            def transformIterable(self, result, encoding):
                return ''.join(result) + ' transformed'

        provideAdapter(Transform1, name=u'test.one')

        published = FauxPublished()
        request = FauxRequest(published)
        request.environ['plone.transformchain.disable'] = True

        result = ['Blah']
        encoding = 'utf-8'

        new_result = self.t(request, result, encoding)
        self.assertEqual(None, new_result)

 def setUp(self):
     from Products.Reflecto.catalog import \
         FileProxyIndexableContentAdapter
     self.reflector = MockReflector()
     provideAdapter(FileProxyIndexableContentAdapter) 
     getGlobalSiteManager().registerUtility(FakeConverter, IConverter,
             'image/jpeg')

def registerPublisherForFTI(fti):
    # This registers a publisher that will allow to traverse to each sql item
    if not getattr(fti, 'sql_table', None):
        return
    name = getattr(fti, 'sql_folder_id', fti.getId())
    has_folder = False
    if name and IRelationValue.providedBy(name):
        obj = name.to_object
        if obj:
            has_folder = True
            name = obj.getId()
        else:
            name = fti.getId()
    elif not name:
        name = fti.getId()
    if not has_folder:
        view = queryMultiAdapter((None, ICollectiveBehaviorSQLLayer), IBrowserView, name='data-'+name, default=None)
        if view != None:
            return view
        publisher = SQLItemPublisher
        provideAdapter(
            factory=publisher,
            adapts=(None, ICollectiveBehaviorSQLLayer),
            provides=IBrowserView,
            name='data-'+name)

        LOG.info('Publisher registered for data-'+name)