Mailing-List: contact bloodhound-commits-help@incubator.apache.org;
 run by ezmlm
Precedence: bulk
Reply-To: bloodhound-dev@incubator.apache.org
Content-Type: text/plain; charset="utf-8"
MIME-Version: 1.0
Content-Transfer-Encoding: 7bit
Subject: svn commit: r1460332 - in
 /incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct:
 multiproduct/perm.py multiproduct/product_admin.py
 tests/admin/product_admin.py tests/env.py tests/perm.py
Date: Sun, 24 Mar 2013 12:24:30 -0000
To: bloodhound-commits@incubator.apache.org
From: jure@apache.org
Message-Id: <20130324122431.2F7612388978@eris.apache.org>

Author: jure
Date: Sun Mar 24 12:24:30 2013
New Revision: 1460332

URL: http://svn.apache.org/r1460332
Log:
#430, admin panels for PRODUCT_ADMIN based on white list, patch t430_r1457691_product_admin_whitelist.diff applied (from Olemis)


Added:
    incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/tests/admin/product_admin.py
Modified:
    incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/multiproduct/perm.py
    incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/multiproduct/product_admin.py
    incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/tests/env.py
    incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/tests/perm.py

Modified: incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/multiproduct/perm.py
URL: http://svn.apache.org/viewvc/incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/multiproduct/perm.py?rev=1460332&r1=1460331&r2=1460332&view=diff
==============================================================================
--- incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/multiproduct/perm.py (original)
+++ incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/multiproduct/perm.py Sun Mar 24 12:24:30 2013
@@ -1,3 +1,4 @@
+from functools import wraps
 
 #  Licensed to the Apache Software Foundation (ASF) under one
 #  or more contributor license agreements.  See the NOTICE file
@@ -21,7 +22,11 @@
 __all__ = 'ProductPermissionPolicy',
 
 from trac.core import Component, implements
-from trac.perm import IPermissionPolicy, PermissionSystem
+from trac.perm import IPermissionPolicy, PermissionSystem, PermissionError
+
+#--------------------------
+# Permission components
+#--------------------------
 
 class MultiproductPermissionPolicy(Component):
     """Apply product policy in product environments to deal with TRAC_ADMIN,
@@ -47,3 +52,153 @@ class MultiproductPermissionPolicy(Compo
                 return True if action in permsys.get_actions() and \
                                 action != 'TRAC_ADMIN' \
                             else None
+
+
+#--------------------------
+# Impersonation helpers
+#--------------------------
+
+class SudoPermissionContext(object):
+    """Allows a permitted user (by default `PRODUCT_ADMIN`) to execute
+    a command as if temporarily granted with `TRAC_ADMIN` or other specific
+    permission. There is also support to revoke some actions unconditionally.
+    
+    These objects will act as context managers wrapping the permissions cache
+    of the target request object. Entering the same context more than once
+    is not supported and will result in unexpected behavior.
+    """
+    def __init__(self, req, require=None, grant=None, revoke=None):
+        grant = frozenset(grant if grant is not None else ('TRAC_ADMIN',))
+        revoke = frozenset(revoke or [])
+        if grant & revoke:
+            raise ValueError('Impossible to grant and revoke (%s)' %
+                             ', '.join(sorted(grant & revoke)))
+
+        self.grant = grant
+        self.revoke = revoke
+        if req:
+            self._expand_perms(req.perm.env)
+        else:
+            self._expanded = False
+        self._perm = None
+        self.req = req
+        self.require_actions = frozenset(('PRODUCT_ADMIN',) if require is None 
+                                         else ([require] 
+                                               if isinstance(require, basestring)
+                                               else require))
+
+    @property
+    def perm(self):
+        return self._perm
+
+    @perm.setter
+    def perm(self, perm):
+        if perm and not self._expanded:
+            self._expand_perms(perm.env)
+        self._perm = perm
+
+    def __getattr__(self, attrnm):
+        # Actually PermissionCache.__slots__ but this will be faster
+        if attrnm in ('env', 'username', '_resource', '_cache'):
+            try:
+                return getattr(self.perm, attrnm)
+            except AttributeError:
+                pass
+        raise AttributeError("'%s' object has no attribute '%s'" %
+                             (self.__class__.__name__, attrnm))
+
+    def __enter__(self):
+        if self.req is None:
+            # e.g. instances returned by __call__
+            raise ValueError('Context manager not bound to request object')
+        req = self.req
+        for action in self.require_actions:
+            req.perm.require(action)
+        self.perm = req.perm
+        req.perm = self
+        return self
+
+    def __exit__(self, exc_type, exc_value, tb):
+        self.req.perm = self.perm
+        self.perm = None
+
+    # Internal methods
+
+    @property
+    def is_active(self):
+        """Determine whether this context is active
+        """
+        return self.req and self.perm
+
+    def _expand_perms(self, env):
+        permsys = PermissionSystem(env)
+        grant = frozenset(permsys.expand_actions(self.grant))
+        revoke = frozenset(permsys.expand_actions(self.revoke))
+        # Double check ambiguous action lists
+        if grant & revoke:
+            raise ValueError('Impossible to grant and revoke (%s)' %
+                             ', '.join(sorted(grant & revoke)))
+        self.grant = grant
+        self.revoke = revoke
+        self._expanded = True
+
+    def __assert_require(f):
+        @wraps(f)
+        def __require(self, *args, **kwargs):
+            # FIXME : No check ? Transform into assert statement ?
+            if not self.perm:
+                raise RuntimeError('Permission check out of context')
+            if not self.is_active:
+                for action in self.require_actions:
+                    self.perm.require(action)
+            return f(self, *args, **kwargs)
+
+        return __require
+
+    # PermissionCache methods
+    @__assert_require
+    def __call__(self, realm_or_resource, id=False, version=False):
+        newperm = self.perm(realm_or_resource, id, version)
+        if newperm is self.perm:
+            return self
+        else:
+            newctx = SudoPermissionContext(None, self.require_actions, self.grant,
+                                           self.revoke)
+            newctx.perm = newperm
+            return newctx
+
+    @__assert_require
+    def has_permission(self, action, realm_or_resource=None, id=False,
+                       version=False):
+        return action in self.grant or \
+               (action not in self.revoke and 
+                self.perm.has_permission(action, realm_or_resource, id, 
+                                         version))
+
+    __contains__ = has_permission
+
+    @__assert_require
+    def require(self, action, realm_or_resource=None, id=False, version=False):
+        if action in self.grant:
+            return
+        if action in self.revoke:
+            resource = self._normalize_resource(realm_or_resource, id, version)
+            raise PermissionError(action, resource, self.perm.env)
+        self.perm.require(action, realm_or_resource, id, version)
+
+    assert_permission = require
+
+    @__assert_require
+    def permissions(self):
+        """Deprecated (but still used by the HDF compatibility layer)
+        """
+        self.perm.env.log.warning("perm.permissions() is deprecated and "
+                             "is only present for HDF compatibility")
+        permsys = PermissionSystem(self.perm.env)
+        actions = permsys.get_user_permissions(self.perm.username)
+        return [action for action in actions if action in self]
+
+    del __assert_require
+
+
+sudo = SudoPermissionContext

Modified: incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/multiproduct/product_admin.py
URL: http://svn.apache.org/viewvc/incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/multiproduct/product_admin.py?rev=1460332&r1=1460331&r2=1460332&view=diff
==============================================================================
--- incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/multiproduct/product_admin.py (original)
+++ incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/multiproduct/product_admin.py Sun Mar 24 12:24:30 2013
@@ -18,17 +18,25 @@
 
 """Admin panels for product management"""
 
+from trac.admin.api import IAdminPanelProvider
+from trac.admin.web_ui import AdminModule
 from trac.core import *
 from trac.config import *
 from trac.perm import PermissionSystem
-from trac.admin.api import IAdminPanelProvider
-from trac.ticket.admin import TicketAdminPanel, _save_config
 from trac.resource import ResourceNotFound
-from model import Product
+from trac.ticket.admin import TicketAdminPanel, _save_config
+from trac.util import lazy
 from trac.util.translation import _, N_, gettext
+from trac.web.api import HTTPNotFound, IRequestFilter, IRequestHandler
 from trac.web.chrome import Chrome, add_notice, add_warning
+
 from multiproduct.env import ProductEnvironment
+from multiproduct.model import Product
+from multiproduct.perm import sudo
 
+#--------------------------
+# Product admin panel
+#--------------------------
 
 class ProductAdminPanel(TicketAdminPanel):
     """The Product Admin Panel"""
@@ -125,3 +133,131 @@ class ProductAdminPanel(TicketAdminPanel
             data['owners'] = None
         return 'admin_products.html', data
 
+#--------------------------
+# Advanced administration in product context
+#--------------------------
+
+class IProductAdminAclContributor(Interface):
+    """Interface implemented by components contributing with entries to the
+    access control white list in order to enable admin panels in product
+    context. 
+    
+    **Notice** that deny entries configured by users in the blacklist
+    (i.e. using TracIni `admin_blacklist` option in `multiproduct` section)
+    will override these entries.
+    """
+    def enable_product_admin_panels():
+        """Return a sequence of `(cat_id, panel_id)` tuples that will be
+        enabled in product context unless specified otherwise in configuration.
+        If `panel_id` is set to `'*'` then all panels in section `cat_id`
+        will have green light.
+        """
+
+
+class ProductAdminModule(Component):
+    """Leverage administration panels in product context based on the
+    combination of white list and black list.
+    """
+    implements(IRequestFilter, IRequestHandler)
+
+    acl_contributors = ExtensionPoint(IProductAdminAclContributor)
+
+    raw_blacklist = ListOption('multiproduct', 'admin_blacklist', 
+        doc="""Do not show any product admin panels in this list even if
+        allowed by white list. Value must be a comma-separated list of
+        `cat:id` strings respectively identifying the section and identifier
+        of target admin panel. Empty values of `cat` and `id` will be ignored
+        and warnings emitted if TracLogging is enabled. If `id` is set
+        to `*` then all panels in `cat` section will be added to blacklist
+        while in product context.""")
+
+    @lazy
+    def acl(self):
+        """Access control table based on blacklist and white list.
+        """
+        # FIXME : Use an immutable (mapping?) type
+        acl = {}
+        if isinstance(self.env, ProductEnvironment):
+            for acl_c in self.acl_contributors:
+                for cat_id, panel_id in acl_c.enable_product_admin_panels():
+                    if cat_id and panel_id:
+                        if panel_id == '*':
+                            acl[cat_id] = True
+                        else:
+                            acl[(cat_id, panel_id)] = True
+                    else:
+                        self.log.warning('Invalid panel %s in white list',
+                                         panel_id)
+    
+            # Blacklist entries will override those in white list
+            warnings = []
+            for panelref in self.raw_blacklist:
+                try:
+                    cat_id, panel_id = panelref.split(':')
+                except ValueError:
+                    cat_id = panel_id = ''
+                if cat_id and panel_id:
+                    if panel_id == '*':
+                        acl[cat_id] = False
+                    else:
+                        acl[(cat_id, panel_id)] = False
+                else:
+                    warnings.append(panelref)
+            if warnings:
+                self.log.warning("Invalid panel descriptors '%s' in blacklist",
+                                 ','.join(warnings))
+        return acl
+
+    # IRequestFilter methods
+    def pre_process_request(self, req, handler):
+        """Intercept admin requests in product context if `TRAC_ADMIN`
+        expectations are not met.
+        """
+        if isinstance(self.env, ProductEnvironment) and \
+                handler is AdminModule(self.env) and \
+                not req.perm.has_permission('TRAC_ADMIN') and \
+                req.perm.has_permission('PRODUCT_ADMIN'):
+            # Intercept admin request
+            return self
+        return handler
+
+    def post_process_request(self, req, template, data, content_type):
+        return template, data, content_type
+
+    # IRequestHandler methods
+    def match_request(self, req):
+        """Never match a request"""
+
+    def process_request(self, req):
+        """Anticipate permission error to hijack admin panel dispatching
+        process in product context if `TRAC_ADMIN` expectations are not met.
+        """
+        # TODO: Verify `isinstance(self.env, ProductEnvironment)` once again ?
+        cat_id = req.args.get('cat_id')
+        panel_id = req.args.get('panel_id')
+        if self._check_panel(cat_id, panel_id):
+            with sudo(req):
+                return self.global_process_request(req)
+        else:
+            raise HTTPNotFound(_('Unknown administration panel'))
+
+    global_process_request = AdminModule.process_request.im_func
+
+    # Internal methods
+    def _get_panels(self, req):
+        if isinstance(self.env, ProductEnvironment):
+            panels, providers = AdminModule(self.env)._get_panels(req)
+            # Filter based on ACLs
+            panels = [p for p in panels if self._check_panel(p[0], p[2])]
+#            providers = dict([k, p] for k, p in providers.iteritems()
+#                                    if self._check_panel(*k))
+            return panels, providers
+        else:
+            return [], []
+
+    def _check_panel(self, cat_id, panel_id):
+        cat_allow = self.acl.get(cat_id)
+        panel_allow = self.acl.get((cat_id, panel_id))
+        return cat_allow is not False and panel_allow is not False \
+               and (cat_allow, panel_allow) != (None, None) \
+               and (cat_id, panel_id) != ('general', 'plugin') # double-check !

Added: incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/tests/admin/product_admin.py
URL: http://svn.apache.org/viewvc/incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/tests/admin/product_admin.py?rev=1460332&view=auto
==============================================================================
--- incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/tests/admin/product_admin.py (added)
+++ incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/tests/admin/product_admin.py Sun Mar 24 12:24:30 2013
@@ -0,0 +1,505 @@
+
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing,
+#  software distributed under the License is distributed on an
+#  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+#  KIND, either express or implied.  See the License for the
+#  specific language governing permissions and limitations
+#  under the License.
+
+"""Tests for Apache(TM) Bloodhound's product admin"""
+
+import sys
+import unittest
+from wsgiref.util import setup_testing_defaults
+
+from trac.admin.api import IAdminPanelProvider
+from trac.admin.web_ui import AdminModule, PluginAdminPanel
+from trac.core import Component, implements
+from trac.perm import DefaultPermissionPolicy, DefaultPermissionStore, \
+                      PermissionCache, PermissionSystem
+from trac.tests.perm import TestPermissionRequestor
+from trac.web.api import HTTP_STATUS, HTTPForbidden, HTTPNotFound, \
+                         IRequestFilter, RequestDone, Request
+from trac.web.main import RequestDispatcher
+
+from multiproduct import api, product_admin
+from multiproduct.env import ProductEnvironment
+from multiproduct.product_admin import IProductAdminAclContributor, \
+                                       ProductAdminModule
+from tests.env import MultiproductTestCase
+
+class TestAdminHandledException(Exception):
+    product = None
+    category = None
+    page = None
+    path_info = None
+    admin_panels = None
+
+class TestAdminPanel(Component):
+    implements(IAdminPanelProvider, IRequestFilter)
+
+    # IAdminPanelProvider methods
+    def get_admin_panels(self, req):
+        if 'TRAC_ADMIN' in req.perm:
+            yield 'testcat1', 'Test category 1', 'panel1', 'Test panel 1'
+            yield 'testcat1', 'Test category 1', 'panel2', 'Test panel 2'
+            yield 'testcat1', 'Test category 1', 'panel3', 'Test panel 3'
+    
+            yield 'testcat2', 'Test category 2', 'panel1', 'Test panel 1'
+            yield 'testcat2', 'Test category 2', 'panel_2', 'Test panel 2'
+            yield 'testcat2', 'Test category 2', 'panel-3', 'Test panel 3'
+    
+            yield 'testcat3', 'Test category 3', 'panel1', 'Test panel 1'
+            yield 'testcat3', 'Test category 3', 'panel2', 'Test panel 2'
+
+    def render_admin_panel(self, req, category, page, path_info):
+        req.perm.require('TRAC_ADMIN')
+        return 'test.html', {'path_info' : path_info}
+
+    def pre_process_request(self, req, handler):
+        return handler
+
+    def post_process_request(self, req, template, data, content_type):
+        if sys.exc_info() == (None, None, None):
+            exc = TestAdminHandledException()
+            exc.product = self.env.product.prefix \
+                     if isinstance(self.env, ProductEnvironment) \
+                     else ''
+            exc.category = data.get('active_cat')
+            exc.page = data.get('active_panel')
+            exc.path_info = data.get('path_info')
+            exc.admin_panels = data.get('panels')
+            raise exc
+        else:
+            return template, data, content_type
+
+
+class PanelsWhitelist(Component):
+    implements(product_admin.IProductAdminAclContributor)
+
+    # IProductAdminAclContributor methods
+    def enable_product_admin_panels(self):
+        yield 'testcat1', 'panel1'
+        yield 'testcat1', 'panel3'
+        yield 'testcat2', 'panel3'
+        yield 'general', 'plugin'
+
+
+class SectionWhitelist(Component):
+    implements(product_admin.IProductAdminAclContributor)
+
+    # IProductAdminAclContributor methods
+    def enable_product_admin_panels(self):
+        yield 'testcat3', '*'
+
+class BaseProductAdminPanelTestCase(MultiproductTestCase):
+    def setUp(self):
+        self._mp_setup(enable=[AdminModule, DefaultPermissionPolicy,
+                               DefaultPermissionStore, PermissionSystem,
+                               PluginAdminPanel, RequestDispatcher, 
+                               api.MultiProductSystem,
+                               product_admin.ProductAdminModule,
+                               PanelsWhitelist, SectionWhitelist, 
+                               TestAdminPanel, TestPermissionRequestor])
+        self.global_env = self.env
+        self.env = ProductEnvironment(self.global_env, self.default_product)
+
+        ProductAdminModule = product_admin.ProductAdminModule
+        self.global_product_admin = ProductAdminModule(self.global_env)
+        self.product_admin = ProductAdminModule(self.env)
+
+    def tearDown(self):
+        self.global_env.reset_db()
+        self.env = self.global_env = None
+        self.product_admin = self.global_product_admin = None
+
+
+class ProductAdminSetupTestCase(BaseProductAdminPanelTestCase):
+    ALL_PANELS = [('testcat1', 'panel1'), ('testcat1', 'panel2'), 
+                  ('testcat1', 'panel3'), ('testcat2', 'panel_1'), 
+                  ('testcat2', 'panel-2'), ('testcat2', 'panel3'), 
+                  ('testcat3', 'panel1'), ('testcat3', 'panel2'), 
+                  ('general', 'plugin'), ]
+
+    def test_init_whitelist(self):
+        self.assertEqual({}, self.global_product_admin.acl)
+        self.assertEqual({'testcat3' : True,
+                          ('testcat1', 'panel1') : True,
+                          ('testcat1', 'panel3'): True,
+                          ('testcat2', 'panel3'): True,
+                          ('general', 'plugin') : True,}, 
+                         self.product_admin.acl)
+        self.assertTrue(all(not self.global_product_admin._check_panel(c, p)
+                            for c, p in self.ALL_PANELS))
+        self.assertTrue(self.product_admin._check_panel('testcat1', 'panel1'))
+        self.assertFalse(self.product_admin._check_panel('testcat1', 'panel2'))
+        self.assertTrue(self.product_admin._check_panel('testcat1', 'panel3'))
+        self.assertFalse(self.product_admin._check_panel('testcat2', 'panel_1'))
+        self.assertFalse(self.product_admin._check_panel('testcat2', 'panel-2'))
+        self.assertTrue(self.product_admin._check_panel('testcat2', 'panel3'))
+        self.assertTrue(self.product_admin._check_panel('testcat3', 'panel1'))
+        self.assertTrue(self.product_admin._check_panel('testcat3', 'panel2'))
+        self.assertFalse(self.product_admin._check_panel('general', 'plugin'))
+        self.assertFalse(self.product_admin._check_panel('other', 'panel'))
+
+    def test_init_blacklist(self):
+        self.global_env.config.set('multiproduct', 'admin_blacklist', 
+                                   'testcat1:panel1,testcat3:panel2')
+        self.env.config.set('multiproduct', 'admin_blacklist', 
+                            'testcat1:panel3,testcat3:panel1,testcat2:*')
+
+        self.assertEqual(['testcat1:panel1','testcat3:panel2'],
+                          self.global_product_admin.raw_blacklist)
+        self.assertEqual(['testcat1:panel3','testcat3:panel1','testcat2:*'],
+                         self.product_admin.raw_blacklist)
+
+        self.assertEqual({}, self.global_product_admin.acl)
+        self.assertEqual({'testcat3' : True,
+                          'testcat2' : False,
+                          ('testcat1', 'panel1') : True,
+                          ('testcat1', 'panel3'): False,
+                          ('testcat2', 'panel3'): True,
+                          ('testcat3', 'panel1'): False,
+                          ('general', 'plugin'): True,}, 
+                         self.product_admin.acl)
+
+        self.assertTrue(all(not self.global_product_admin._check_panel(c, p)
+                            for c, p in self.ALL_PANELS))
+        self.assertTrue(self.product_admin._check_panel('testcat1', 'panel1'))
+        self.assertFalse(self.product_admin._check_panel('testcat1', 'panel2'))
+        self.assertFalse(self.product_admin._check_panel('testcat1', 'panel3'))
+        self.assertFalse(self.product_admin._check_panel('testcat2', 'panel_1'))
+        self.assertFalse(self.product_admin._check_panel('testcat2', 'panel-2'))
+        self.assertFalse(self.product_admin._check_panel('testcat2', 'panel3'))
+        self.assertFalse(self.product_admin._check_panel('testcat3', 'panel1'))
+        self.assertTrue(self.product_admin._check_panel('testcat3', 'panel2'))
+        self.assertFalse(self.product_admin._check_panel('general', 'plugin'))
+        self.assertFalse(self.product_admin._check_panel('other', 'panel'))
+
+
+class ProductAdminDispatchTestCase(BaseProductAdminPanelTestCase):
+    maxDiff = None
+
+    def setUp(self):
+        BaseProductAdminPanelTestCase.setUp(self)
+        self.global_env.config.set('multiproduct', 'admin_blacklist', 
+                                   'testcat1:panel1,testcat3:panel2')
+        self.env.config.set('multiproduct', 'admin_blacklist', 
+                            'testcat1:panel3,testcat3:panel1,testcat2:*')
+        global_permsys = PermissionSystem(self.global_env)
+        permsys = PermissionSystem(self.env)
+
+        global_permsys.grant_permission('adminuser', 'TRAC_ADMIN')
+        global_permsys.grant_permission('prodadmin', 'PRODUCT_ADMIN')
+        global_permsys.grant_permission('testuser', 'TEST_ADMIN')
+        permsys.grant_permission('prodadmin', 'PRODUCT_ADMIN')
+        permsys.grant_permission('testuser', 'TEST_ADMIN')
+
+        self.req = self._get_request_obj()
+
+    def tearDown(self):
+        BaseProductAdminPanelTestCase.tearDown(self)
+        self.req = None
+
+    def _get_request_obj(self):
+        environ = {}
+        setup_testing_defaults(environ)
+
+        def start_response(status, headers):
+            return lambda body: None
+
+        req = Request(environ, start_response)
+        return req
+
+    def _dispatch(self, req, env):
+        req.perm = PermissionCache(env, req.authname)
+        return RequestDispatcher(env).dispatch(req)
+
+    GLOBAL_PANELS = [
+            {'category': {'id': 'general', 'label': 'General'},
+             'panel': {'id': 'plugin', 'label': 'Plugins'}},
+            {'category': {'id': 'testcat1', 'label': 'Test category 1'},
+             'panel': {'id': 'panel1', 'label': 'Test panel 1'}},
+            {'category': {'id': 'testcat1', 'label': 'Test category 1'},
+             'panel': {'id': 'panel2', 'label': 'Test panel 2'}},
+            {'category': {'id': 'testcat1', 'label': 'Test category 1'},
+             'panel': {'id': 'panel3', 'label': 'Test panel 3'}},
+            {'category': {'id': 'testcat2', 'label': 'Test category 2'},
+             'panel': {'id': 'panel-3', 'label': 'Test panel 3'}},
+            {'category': {'id': 'testcat2', 'label': 'Test category 2'},
+             'panel': {'id': 'panel1', 'label': 'Test panel 1'}},
+            {'category': {'id': 'testcat2', 'label': 'Test category 2'},
+             'panel': {'id': 'panel_2', 'label': 'Test panel 2'}},
+            {'category': {'id': 'testcat3', 'label': 'Test category 3'},
+             'panel': {'id': 'panel1', 'label': 'Test panel 1'}},
+            {'category': {'id': 'testcat3', 'label': 'Test category 3'},
+             'panel': {'id': 'panel2', 'label': 'Test panel 2'}}]
+    PRODUCT_PANELS_ALL = [
+            {'category': {'id': 'testcat1', 'label': 'Test category 1'},
+             'panel': {'id': 'panel1', 'label': 'Test panel 1'}},
+            {'category': {'id': 'testcat1', 'label': 'Test category 1'},
+             'panel': {'id': 'panel2', 'label': 'Test panel 2'}},
+            {'category': {'id': 'testcat1', 'label': 'Test category 1'},
+             'panel': {'id': 'panel3', 'label': 'Test panel 3'}},
+            {'category': {'id': 'testcat2', 'label': 'Test category 2'},
+             'panel': {'id': 'panel-3', 'label': 'Test panel 3'}},
+            {'category': {'id': 'testcat2', 'label': 'Test category 2'},
+             'panel': {'id': 'panel1', 'label': 'Test panel 1'}},
+            {'category': {'id': 'testcat2', 'label': 'Test category 2'},
+             'panel': {'id': 'panel_2', 'label': 'Test panel 2'}},
+            {'category': {'id': 'testcat3', 'label': 'Test category 3'},
+             'panel': {'id': 'panel1', 'label': 'Test panel 1'}},
+            {'category': {'id': 'testcat3', 'label': 'Test category 3'},
+             'panel': {'id': 'panel2', 'label': 'Test panel 2'}}]
+    PRODUCT_PANELS_ALLOWED = [
+            {'category': {'id': 'testcat1', 'label': 'Test category 1'},
+             'panel': {'id': 'panel1', 'label': 'Test panel 1'}},
+            {'category': {'id': 'testcat3', 'label': 'Test category 3'},
+             'panel': {'id': 'panel2', 'label': 'Test panel 2'}}]
+
+    # TRAC_ADMIN
+    def test_tracadmin_global_panel(self):
+        """Test admin panel with TRAC_ADMIN in global env
+        """
+        req = self.req
+        req.authname = 'adminuser'
+        req.environ['PATH_INFO'] = '/admin/testcat1/panel1/some/path'
+        with self.assertRaises(TestAdminHandledException) as test_cm:
+            self._dispatch(req, self.global_env)
+
+        exc = test_cm.exception
+        self.assertEqual('', exc.product)
+        self.assertEqual('testcat1', exc.category)
+        self.assertEqual('panel1', exc.page)
+        self.assertEqual('some/path', exc.path_info)
+        self.assertEqual(self.GLOBAL_PANELS, exc.admin_panels)
+
+    def test_tracadmin_global_plugins(self):
+        """Plugin admin panel with TRAC_ADMIN in global env
+        """
+        req = self.req
+        req.authname = 'adminuser'
+        req.environ['PATH_INFO'] = '/admin/general/plugin'
+        # Plugin admin panel looked up but disabled
+        with self.assertRaises(TestAdminHandledException) as test_cm:
+            self._dispatch(req, self.global_env)
+
+        exc = test_cm.exception
+        self.assertEqual(self.GLOBAL_PANELS, exc.admin_panels)
+
+    def test_tracadmin_product_panel_blacklist(self):
+        """Test blacklisted admin panel with TRAC_ADMIN in product env
+        """
+        req = self.req
+        req.authname = 'adminuser'
+        req.environ['PATH_INFO'] = '/admin/testcat3/panel1/some/path'
+        with self.assertRaises(TestAdminHandledException) as test_cm:
+            self._dispatch(req, self.env)
+
+        exc = test_cm.exception
+        self.assertEqual(self.default_product, exc.product)
+        self.assertEqual('testcat3', exc.category)
+        self.assertEqual('panel1', exc.page)
+        self.assertEqual('some/path', exc.path_info)
+        self.assertEqual(self.PRODUCT_PANELS_ALL, exc.admin_panels)
+
+    def test_tracadmin_product_panel_whitelist(self):
+        """Test whitelisted admin panel with TRAC_ADMIN in product env
+        """
+        req = self.req
+        req.authname = 'adminuser'
+        req.environ['PATH_INFO'] = '/admin/testcat1/panel1/some/path'
+        with self.assertRaises(TestAdminHandledException) as test_cm:
+            self._dispatch(req, self.env)
+
+        exc = test_cm.exception
+        self.assertEqual(self.default_product, exc.product)
+        self.assertEqual('testcat1', exc.category)
+        self.assertEqual('panel1', exc.page)
+        self.assertEqual('some/path', exc.path_info)
+        self.assertEqual(self.PRODUCT_PANELS_ALL, exc.admin_panels)
+
+    def test_tracadmin_product_plugins(self):
+        """Plugin admin panel with TRAC_ADMIN in global env
+        """
+        req = self.req
+        req.authname = 'adminuser'
+        req.environ['PATH_INFO'] = '/admin/general/plugin'
+        # Plugin admin panel not available in product context
+        with self.assertRaises(HTTPNotFound):
+            self._dispatch(req, self.env)
+
+    # PRODUCT_ADMIN
+    def test_productadmin_global_panel_whitelist(self):
+        """Test whitelisted admin panel with PRODUCT_ADMIN in product env
+        """
+        req = self.req
+        req.authname = 'prodadmin'
+        req.environ['PATH_INFO'] = '/admin/testcat1/panel1/some/path'
+        with self.assertRaises(HTTPNotFound):
+            self._dispatch(req, self.global_env)
+
+    def test_productadmin_global_panel_blacklist(self):
+        """Test blacklisted admin panel with PRODUCT_ADMIN in product env
+        """
+        req = self.req
+        req.authname = 'prodadmin'
+        req.environ['PATH_INFO'] = '/admin/testcat3/panel1/some/path'
+        with self.assertRaises(HTTPNotFound):
+            self._dispatch(req, self.global_env)
+
+    def test_productadmin_global_panel_norules(self):
+        """Test unspecified admin panel with PRODUCT_ADMIN in product env
+        """
+        req = self.req
+        req.authname = 'prodadmin'
+        req.environ['PATH_INFO'] = '/admin/testcat1/panel2/some/path'
+        with self.assertRaises(HTTPNotFound):
+            self._dispatch(req, self.global_env)
+
+    def test_productadmin_global_plugins(self):
+        """Plugin admin panel with PRODUCT_ADMIN in global env
+        """
+        req = self.req
+        req.authname = 'prodadmin'
+        req.environ['PATH_INFO'] = '/admin/general/plugin'
+        with self.assertRaises(HTTPNotFound):
+            self._dispatch(req, self.global_env)
+
+    def test_productadmin_product_panel_whitelist(self):
+        """Test whitelisted admin panel with PRODUCT_ADMIN in product env
+        """
+        req = self.req
+        req.authname = 'prodadmin'
+        req.environ['PATH_INFO'] = '/admin/testcat1/panel1/some/path'
+        with self.assertRaises(TestAdminHandledException) as test_cm:
+            self._dispatch(req, self.env)
+
+        exc = test_cm.exception
+        self.assertEqual(self.default_product, exc.product)
+        self.assertEqual('testcat1', exc.category)
+        self.assertEqual('panel1', exc.page)
+        self.assertEqual('some/path', exc.path_info)
+        self.assertEqual(self.PRODUCT_PANELS_ALLOWED, exc.admin_panels)
+
+    def test_productadmin_product_panel_blacklist(self):
+        """Test blacklisted admin panel with PRODUCT_ADMIN in product env
+        """
+        req = self.req
+        req.authname = 'prodadmin'
+        req.environ['PATH_INFO'] = '/admin/testcat3/panel1/some/path'
+        with self.assertRaises(HTTPNotFound):
+            self._dispatch(req, self.env)
+
+    def test_productadmin_product_panel_norules(self):
+        """Test unspecified admin panel with PRODUCT_ADMIN in product env
+        """
+        req = self.req
+        req.authname = 'prodadmin'
+        req.environ['PATH_INFO'] = '/admin/testcat1/panel2/some/path'
+        with self.assertRaises(HTTPNotFound):
+            self._dispatch(req, self.env)
+
+    def test_productadmin_product_plugins(self):
+        """Plugin admin panel with PRODUCT_ADMIN in product env
+        """
+        req = self.req
+        req.authname = 'prodadmin'
+        req.environ['PATH_INFO'] = '/admin/general/plugin'
+        with self.assertRaises(HTTPNotFound):
+            self._dispatch(req, self.env)
+
+    # Without meta-permissions
+    def test_user_global_panel_whitelist(self):
+        """Test whitelisted admin panel without meta-perm in product env
+        """
+        req = self.req
+        req.authname = 'testuser'
+        req.environ['PATH_INFO'] = '/admin/testcat1/panel1/some/path'
+        with self.assertRaises(HTTPNotFound):
+            self._dispatch(req, self.global_env)
+
+    def test_user_global_panel_blacklist(self):
+        """Test blacklisted admin panel without meta-perm in product env
+        """
+        req = self.req
+        req.authname = 'testuser'
+        req.environ['PATH_INFO'] = '/admin/testcat3/panel1/some/path'
+        with self.assertRaises(HTTPNotFound):
+            self._dispatch(req, self.global_env)
+
+    def test_user_global_panel_norules(self):
+        """Test unspecified admin panel without meta-perm in product env
+        """
+        req = self.req
+        req.authname = 'testuser'
+        req.environ['PATH_INFO'] = '/admin/testcat1/panel2/some/path'
+        with self.assertRaises(HTTPNotFound):
+            self._dispatch(req, self.global_env)
+
+    def test_user_global_plugins(self):
+        """Plugin admin panel without meta-perm in global env
+        """
+        req = self.req
+        req.authname = 'testuser'
+        req.environ['PATH_INFO'] = '/admin/general/plugin'
+        with self.assertRaises(HTTPNotFound):
+            self._dispatch(req, self.global_env)
+
+    def test_user_product_panel_whitelist(self):
+        """Test whitelisted admin panel without meta-perm in product env
+        """
+        req = self.req
+        req.authname = 'testuser'
+        req.environ['PATH_INFO'] = '/admin/testcat1/panel1/some/path'
+        with self.assertRaises(HTTPNotFound):
+            self._dispatch(req, self.env)
+
+    def test_user_product_panel_blacklist(self):
+        """Test blacklisted admin panel without meta-perm in product env
+        """
+        req = self.req
+        req.authname = 'testuser'
+        req.environ['PATH_INFO'] = '/admin/testcat3/panel1/some/path'
+        with self.assertRaises(HTTPNotFound):
+            self._dispatch(req, self.env)
+
+    def test_user_product_panel_norules(self):
+        """Test unspecified admin panel without meta-perm in product env
+        """
+        req = self.req
+        req.authname = 'testuser'
+        req.environ['PATH_INFO'] = '/admin/testcat1/panel2/some/path'
+        with self.assertRaises(HTTPNotFound):
+            self._dispatch(req, self.env)
+
+    def test_user_product_plugins(self):
+        """Plugin admin panel without meta-perm in product env
+        """
+        req = self.req
+        req.authname = 'testuser'
+        req.environ['PATH_INFO'] = '/admin/general/plugin'
+        with self.assertRaises(HTTPNotFound):
+            self._dispatch(req, self.env)
+
+
+
+def test_suite():
+    return unittest.TestSuite([
+            unittest.makeSuite(ProductAdminSetupTestCase,'test'),
+            unittest.makeSuite(ProductAdminDispatchTestCase,'test'),
+        ])
+
+if __name__ == '__main__':
+    unittest.main(defaultTest='test_suite')
+

Modified: incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/tests/env.py
URL: http://svn.apache.org/viewvc/incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/tests/env.py?rev=1460332&r1=1460331&r2=1460332&view=diff
==============================================================================
--- incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/tests/env.py (original)
+++ incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/tests/env.py Sun Mar 24 12:24:30 2013
@@ -251,10 +251,10 @@ class MultiproductTestCase(unittest.Test
                             vals)
         env.log.debug('Loaded default data')
 
-    def _mp_setup(self):
+    def _mp_setup(self, **kwargs):
         """Shortcut for quick product-aware environment setup.
         """
-        self.env = self._setup_test_env()
+        self.env = self._setup_test_env(**kwargs)
         self._upgrade_mp(self.env)
         self._setup_test_log(self.env)
         self._load_product_from_data(self.env, self.default_product)

Modified: incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/tests/perm.py
URL: http://svn.apache.org/viewvc/incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/tests/perm.py?rev=1460332&r1=1460331&r2=1460332&view=diff
==============================================================================
--- incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/tests/perm.py (original)
+++ incubator/bloodhound/branches/bep_0003_multiproduct/bloodhound_multiproduct/tests/perm.py Sun Mar 24 12:24:30 2013
@@ -23,12 +23,13 @@ import unittest
 
 from trac.admin.api import AdminCommandError
 from trac import perm
+from trac.test import Mock
 from trac.tests.perm import DefaultPermissionStoreTestCase,\
         PermissionSystemTestCase, PermissionCacheTestCase,\
         PermissionPolicyTestCase, TestPermissionPolicy, TestPermissionRequestor
 
 from multiproduct.env import ProductEnvironment
-from multiproduct.perm import MultiproductPermissionPolicy
+from multiproduct.perm import MultiproductPermissionPolicy, sudo
 from tests.env import MultiproductTestCase
 
 
@@ -138,6 +139,98 @@ class ProductPermissionCacheTestCase(Per
         pass
 
 
+class SudoTestCase(ProductPermissionCacheTestCase):
+    loader = unittest.defaultTestLoader
+    tcnames = loader.getTestCaseNames(ProductPermissionCacheTestCase)
+    _gen_tests = {}
+
+    def test_sudo_wrong_context(self):
+        sudoperm = sudo(None, 'EMAIL_VIEW', ['TEST_ADMIN'])
+
+        with self.assertRaises(RuntimeError) as test_cm:
+            sudoperm.has_permission('TEST_MODIFY')
+        self.assertEqual('Permission check out of context', 
+                         str(test_cm.exception))
+
+        with self.assertRaises(ValueError) as test_cm:
+            with sudoperm:
+                pass
+        self.assertEquals('Context manager not bound to request object',
+                          str(test_cm.exception))
+
+    def test_sudo_fail_require(self):
+        sudoperm = sudo(None, 'EMAIL_VIEW', ['TEST_ADMIN'])
+
+        sudoperm.perm = self.perm
+        with self.assertRaises(perm.PermissionError) as test_cm:
+            sudoperm.require('TRAC_ADMIN')
+        self.assertEqual('EMAIL_VIEW', test_cm.exception.action)
+
+    def test_sudo_grant_meta_perm(self):
+        self.env.parent.enable_component(perm.PermissionSystem)
+        self.env.enable_component(perm.PermissionSystem)
+        del self.env.parent.enabled[perm.PermissionSystem]
+        del self.env.enabled[perm.PermissionSystem]
+
+        sudoperm = sudo(None, 'TEST_CREATE', ['TRAC_ADMIN'])
+        sudoperm.perm = self.perm
+        
+        self.assertTrue(sudoperm.has_permission('EMAIL_VIEW'))
+
+    def test_sudo_ambiguous(self):
+        with self.assertRaises(ValueError) as test_cm:
+            sudo(None, 'TEST_MODIFY', ['TEST_MODIFY', 'TEST_DELETE'], 
+                 ['TEST_MODIFY', 'TEST_CREATE'])
+        self.assertEquals('Impossible to grant and revoke (TEST_MODIFY)', 
+                          str(test_cm.exception))
+
+        with self.assertRaises(ValueError) as test_cm:
+            sudoperm = sudo(None, 'TEST_MODIFY', ['TEST_ADMIN'], 
+                 ['TEST_MODIFY', 'TEST_CREATE'])
+            sudoperm.perm = self.perm
+        self.assertEquals('Impossible to grant and revoke '
+                          '(TEST_CREATE, TEST_MODIFY)', 
+                          str(test_cm.exception))
+
+        with self.assertRaises(ValueError) as test_cm:
+            req = Mock(perm=self.perm)
+            sudo(req, 'TEST_MODIFY', ['TEST_ADMIN'], 
+                 ['TEST_MODIFY', 'TEST_CREATE'])
+        self.assertEquals('Impossible to grant and revoke '
+                          '(TEST_CREATE, TEST_MODIFY)', 
+                          str(test_cm.exception))
+
+    # Sudo permission context equivalent to  permissions cache
+    # if there's no action to require, allow or deny.
+    def _test_with_sudo_rules(tcnm, prefix, grant):
+        target = getattr(ProductPermissionCacheTestCase, tcnm)
+
+        def _sudo_eq_checker(self):
+            for action in grant:
+                self.perm_system.revoke_permission('testuser', action)
+            realperm = self.perm
+            self.perm = sudo(None, [], grant, [])
+            self.perm.perm = realperm
+            target(self)
+
+        _sudo_eq_checker.func_name = prefix + tcnm
+        return _sudo_eq_checker
+
+    for tcnm in tcnames:
+        f1 = _test_with_sudo_rules(tcnm, '', [])
+        f2 = _test_with_sudo_rules(tcnm, 'test_sudo_partial_', 
+                                   ['TEST_MODIFY'])
+        f3 = _test_with_sudo_rules(tcnm, 'test_sudo_full_', 
+                                   ['TEST_MODIFY', 'TEST_ADMIN'])
+        for f in (f1, f2, f3):
+            _gen_tests[f.func_name] = f
+
+    del loader, tcnames, tcnm, f1, f2, f3
+
+list(setattr(SudoTestCase, tcnm, f)
+     for tcnm, f in SudoTestCase._gen_tests.iteritems())
+
+
 class ProductPermissionPolicyTestCase(PermissionPolicyTestCase, 
                                            MultiproductTestCase):
     @property
@@ -249,6 +342,8 @@ def test_suite():
     suite.addTest(unittest.makeSuite(ProductPermissionSystemTestCase, 'test'))
     suite.addTest(unittest.makeSuite(ProductPermissionCacheTestCase, 'test'))
     suite.addTest(unittest.makeSuite(ProductPermissionPolicyTestCase, 'test'))
+
+    suite.addTest(unittest.makeSuite(SudoTestCase, 'test'))
     return suite
 
 if __name__ == '__main__':