def testGetURL_catalog_as_utility(self):
            request = DummyRequest()
            b = self._makeBrain(0)

            setRequest(request)
            self.assertEqual(b.getURL(), 'http://superbad.com/conflicter')
            clearRequest()

    def __call__(self):
        celery = getCelery()
        if celery.conf.task_always_eager:
            self.eager = True
            # dive out of setup, this is not run in a celery task runner
            self.app = getApp()
            return self._run()

        self.app = makerequest(getApp())
        self.app.REQUEST['PARENTS'] = [self.app]
        setRequest(self.app.REQUEST)

        transaction.begin()
        try:
            try:
                result = self._run()
                # commit transaction
                transaction.commit()
                return result
            except ConflictError as e:
                # On ZODB conflicts, retry using celery's mechanism
                transaction.abort()
                raise Retry(exc=e)
            except Exception:
                logger.warn('Error running task: %s' % traceback.format_exc())
                transaction.abort()
                raise
        finally:
            noSecurityManager()
            setSite(None)
            self.app._p_jar.close()
            clearRequest()

    def testSetUp(self):
        import Zope2
        import transaction

        # Open a new app and save it as the resource ``app``.

        environ = {
            'SERVER_NAME': self['host'],
            'SERVER_PORT': str(self['port']),
        }

        app = addRequestContainer(Zope2.app(), environ=environ)
        request = app.REQUEST
        request['PARENTS'] = [app]
        
        # Make sure we have a zope.globalrequest request
        try:
            from zope.globalrequest import setRequest
            setRequest(request)
        except ImportError:
            pass

        # Start a transaction
        transaction.begin()

        # Save resources for tests to access
        self['app'] = app
        self['request'] = request

    def test_basic_tile_purge_cache(self):
        provideHandler(queuePurge)

        request = self.request
        alsoProvides(request, IAttributeAnnotatable)
        setRequest(request)

        registry = queryUtility(IRegistry)
        registry.registerInterface(ICachePurgingSettings)
        provideUtility(registry, IRegistry)

        settings = registry.forInterface(ICachePurgingSettings)
        settings.enabled = True
        settings.cachingProxies = ('http://localhost:1234',)

        obj = self.portal['my-image']
        data = self.tile.data
        scales = api.content.get_view(u'images', obj, self.request)
        self.tile.data['image'] = NamedBlobImage(str(scales.scale('image').data))
        data_mgr = ITileDataManager(self.tile)
        data_mgr.set(data)

        self.assertEqual(
            set([
                '/c1/@@collective.cover.basic/test',
                '/c1/@@collective.cover.basic/test/@@images/image',
                '/c1/@@collective.cover.basic/test/@@images/icon',
                '/c1/@@collective.cover.basic/test/@@images/mini',
                '/c1/@@collective.cover.basic/test/@@images/large',
                '/c1/@@collective.cover.basic/test/@@images/listing',
                '/c1/@@collective.cover.basic/test/@@images/thumb',
                '/c1/@@collective.cover.basic/test/@@images/preview',
                '/c1/@@collective.cover.basic/test/@@images/tile']),
            IAnnotations(request)['plone.cachepurging.urls'])

    def get(self, path):
        if path == '/logout':
            if users.get_current_user():
                self.redirect(users.create_logout_url('/'))
            else:
                self.redirect(self.request.relative_url('/'))
            return
        setRequest(self.request)
        app = createObject('Application', self.request, self.response)
        setSite(app)
        traverser = ITraverser(app)
        stack = path.split('/')
        if stack[-1] == '':
            stack = stack[:-1]
        if stack[0] == '':
            stack = stack[1:]
        obj = traverser.traverse(self.request, self.response, stack)

        requires = getattr(obj, 'permission_required', None)
        if requires:
            user = users.get_current_user()
            if not checkPermission(obj, user, requires):
                self.redirect(users.create_login_url(self.request.uri))
                return

        if getattr(obj, 'render', None):
            self.response.out.write(obj.render())
        else:
            self.response.headers['Content-Type'] = 'text/plain'
            self.response.out.write(obj)
        getUtility(ISession).commit()
        clearRequest()

def setup_plone(plone, options=None):
    """Takes care of setting up a request, manager security context and
    setting up the site manager for the given Plone site.
    Returns the Plone site root object.

    This is based on the setup code found in p.recipe.zope2instance.ctl:
    https://github.com/plone/plone.recipe.zope2instance/blob/
    02bfb735/src/plone/recipe/zope2instance/ctl.py#L672-L680
    """
    app = plone.restrictedTraverse('/')

    # Set up request for debug / bin/instance run mode.
    app = makerequest(app)
    app.REQUEST['PARENTS'] = [app]
    setRequest(app.REQUEST)

    # Get a reference to the Plone site *inside* the request-wrapped app
    plone = app.restrictedTraverse(plone.id)

    # Set up Manager security context
    user = AccessControl.SecurityManagement.SpecialUsers.system
    user = user.__of__(app.acl_users)
    newSecurityManager(app, user)

    # Set up site to make component registry work
    setSite(plone)
    return plone

    def upgrade(
            self, upgrade_portal=True,
            upgrade_all_profiles=True, upgrade_profiles=(), **kw):
        hooks.setSite(self.context)
        setRequest(self.context.REQUEST)
        # initialize portal_skins
        self.context.setupCurrentSkin(self.context.REQUEST)
        # setup language for plone 4: see https://docs.plone.org/manage/upgrading/version_specific_migration/p4x_to_p5x_upgrade.html#portal-languages-is-now-a-utility  # noqa
        if getattr(self.context, 'portal_languages', None):
            self.context.portal_languages(self.context, self.context.REQUEST)
        self.setup = getToolByName(self.context, 'portal_setup')
        self.log('Upgrading {0}'.format(self.context))
        # setup BrowserLayer, see: https://dev.plone.org/ticket/11673
        notify(BeforeTraverseEvent(self.context, self.context.REQUEST))

        baseline = self.setup.getBaselineContextID()
        prof_type, profile_id = baseline.split('-', 1)
        self.base_profile = profile_id
        if upgrade_portal:
            # Do the baseline profile upgrade first
            self.upgradeProfile(profile_id, **kw)

        # Upgrade extension profiles
        if upgrade_all_profiles:
            if upgrade_profiles:
                raise ValueError(
                    'upgrade_profiles conflicts with upgrade_all_profiles')
            upgrade_profiles = self.setup.listProfilesWithPendingUpgrades()
        if upgrade_profiles:
            self.upgradeExtensions(upgrade_profiles, **kw)

        self.log('Upgraded {0}'.format(self.context))

def get_plone_site():
    import Zope2
    app = Zope2.app()
    from Testing.ZopeTestCase import utils
    utils._Z2HOST = HOST

    path = PLONE.split('/')

    app = utils.makerequest(app)
    app.REQUEST['PARENTS'] = [app]
    app.REQUEST.other['VirtualRootPhysicalPath'] = path
    from zope.globalrequest import setRequest
    setRequest(app.REQUEST)

    from AccessControl.SpecialUsers import system as user
    from AccessControl.SecurityManagement import newSecurityManager
    newSecurityManager(None, user)

    _site = app[path[-1]]
    site = _site.__of__(app)

    from zope.site.hooks import setSite
    setSite(site)

    return site

    def test_enabled(self):
        context = FauxContext()
        
        request = FauxRequest()
        alsoProvides(request, IAttributeAnnotatable)
        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")
        
        notify(Purge(context))
        
        self.assertEquals({'zojax.cachepurging.urls': set(['/foo', '/bar'])},
                          dict(IAnnotations(request)))

    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 testSetUp(self):
     """Do not mess up here with another stacked
     demo storage!!!"""
     import Zope2
     environ = {
         'SERVER_NAME': 'localhost',
         'SERVER_PORT': str(self['port']),
     }
     app = z2.addRequestContainer(
         Zope2.app(), environ=environ)
     request = app.REQUEST
     request['PARENTS'] = [app]
     # Make sure we have a zope.globalrequest request
     try:
         from zope.globalrequest import setRequest
         setRequest(request)
     except ImportError:
         pass
     self['app'] = app
     self['request'] = request
     self['portal'] = portal = self['app'][PLONE_SITE_ID]
     transaction.commit()
     self.setUpEnvironment(portal)
     transaction.commit()
     LayerMixin.testSetUp(self)
     transaction.commit()

    def setUp(self):
        self.settings = getUtility(IRegistry).forInterface(IThemeSettings)

        self.settings.enabled = True
        self.settings.rules = u'python://plone.app.theming/tests/rules.xml'
        self.settings.parameterExpressions = {
            'stringParam': 'string:string param value',
            'boolParam': 'python:False',
            'contextParam': 'context/absolute_url | string:no context',
            'requestParam': 'request/useother | string:off',
        }

        self.portal = self.layer['portal']
        setRequest(self.portal.REQUEST)

        applyProfile(self.portal, 'plone.app.caching:without-caching-proxy')

        self.cacheSettings = getUtility(IRegistry).forInterface(ICacheSettings)
        self.cacheSettings.enabled = True
        self.cacheSettings.operationMapping = {
            'plone.content.folderView': 'plone.app.caching.weakCaching'}
        registry = getUtility(IRegistry)
        registry['plone.app.caching.weakCaching.ramCache'] = True

        import transaction
        transaction.commit()

    def test_isolate_globalrequest(self):
        setRequest(self.layer['request'])

        with isolate_globalrequest():
            self.assertIsNone(None, getRequest())
            setRequest('bar')

        self.assertEquals(self.layer['request'], getRequest())

    def setUp(self):
        self.app = self.layer['app']
        self.portal = self.layer['portal']

        setRequest(self.portal.REQUEST)

        self.registry = getUtility(IRegistry)
        self.cacheSettings = self.registry.forInterface(ICacheSettings)
        self.cacheSettings.enabled = True

        def testGetObjectHappy_catalog_as_utility(self):
            request = DummyRequest()
            b = self._makeBrain(1)

            setRequest(request)
            self.assertEqual(b.getPath(), '/happy')
            self.assertEqual(b.getObject().REQUEST, request)
            self.assertTrue(aq_base(b.getObject()) is
                            aq_base(self.cat.getobject(1)))
            clearRequest()

    def test_decorator_isolates_globalrequest(self):
        setRequest(self.layer['request'])

        @isolated
        def foo():
            self.assertIsNone(None, getRequest())
            setRequest('bar')
            return 'Foo'

        self.assertEquals('Foo', foo())
        self.assertEquals(self.layer['request'], getRequest())

 def setUp(self):
     self.request = TestRequest(environ={'HTTP_ACCEPT_LANGUAGE': 'en'})
     setRequest(self.request)
     self.portal = self.layer['portal']
     login(self.portal, TEST_USER_NAME)
     setRoles(self.portal, TEST_USER_ID, ['Manager'])
     provideUtility(ExampleVocabulary(), name=u'vocab_class')
     provideUtility(ExampleFunctionVocabulary, name=u'vocab_function')
     vocabulary._permissions.update({
         'vocab_class': 'Modify portal content',
         'vocab_function': 'Modify portal content',
     })

    def test_no_tool(self):
        root = FauxContent('')
        content = FauxContent('foo').__of__(root)
        discussable = FauxDiscussable().__of__(content)

        request = FauxRequest()
        setRequest(request)

        purge = DiscussionItemPurgePaths(discussable)

        self.assertEqual([], list(purge.getRelativePaths()))
        self.assertEqual([], list(purge.getAbsolutePaths()))

    def test_no_registry(self):
        provideAdapter(DefaultRulesetLookup)

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

        modifyStreamingResponse(DummyStreamingEvent(response))

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

    def setUpPloneSite(self, portal):
        applyProfile(portal, 'xtgl.policy:default')
#         applyProfile(portal, 'my315ok.socialorgnization:default')
#         applyProfile(portal, 'dexterity.membrane:default')
#        applyProfile(portal, 'dexterity.membrane.content:example')

#        portal = self.layer['portal']
        #make global request work
        from zope.globalrequest import setRequest
        setRequest(portal.REQUEST)
        # login doesn't work so we need to call z2.login directly
        z2.login(portal.__parent__.acl_users, SITE_OWNER_NAME)