Add helpers to handle mount points

- A new Base.getMountedObject method allow to access to parent object in the
  mounted storage so that it is possible to migrate their class persistently.
  A common pattern for modules in different Data.fs would be:
    real_module = portal.mounted_module.getMountedObject()
    real_module.getParentValue().migrateToPortalTypeClass()

- Make it possible to manage mount points in unit tests.

git-svn-id: https://svn.erp5.org/repos/public/erp5/trunk@44781 20353a03-c40f-0410-a6d1-a30d3c3de9de

product/ERP5Type/Base.py

@@ -1529,6 +1529,17 @@ class Base( CopyContainer,
     """
     return self
 
+  def getMountedObject(self):
+      """
+      If self is a mount-point, return the mounted object in its own storage
+      """
+      from Products.ZODBMountPoint.MountedObject import getMountPoint
+      mount_point = getMountPoint(self)
+      if mount_point is not None:
+        connection = self._p_jar
+        assert mount_point._getMountedConnection(connection) is connection
+        return mount_point._traverseToMountedRoot(connection.root(), None)
+
   def asSQLExpression(self, strict_membership=0, table='category', base_category = None):
     """
       Any document can be used as a Category. It can therefore

product/ERP5Type/tests/ERP5TypeTestCase.py

@@ -100,6 +100,7 @@ ZopeTestCase.installProduct('PageTemplates', quiet=install_product_quiet)
 ZopeTestCase.installProduct('PythonScripts', quiet=install_product_quiet)
 ZopeTestCase.installProduct('ExternalMethod', quiet=install_product_quiet)
 ZopeTestCase.installProduct('Sessions', quiet=install_product_quiet)
+ZopeTestCase.installProduct('ZODBMountPoint', quiet=install_product_quiet)
 
 try:
   from Testing.ZopeTestCase.layer import onsetup

product/ERP5Type/tests/runUnitTest.py

@@ -363,6 +363,11 @@ def runUnitTestList(test_list, verbosity=1, debug=0, run_only=None):
   cfg = App.config.getConfiguration()
   cfg.testinghome = instance_home
   cfg.instancehome = instance_home
+  try:
+    from Zope2.Startup.datatypes import DBTab
+    cfg.dbtab = DBTab({}, {})
+  except ImportError: # Zope 2.8
+    pass
   App.config.setConfiguration(cfg)
 
   if WIN:
@@ -457,6 +462,10 @@ def runUnitTestList(test_list, verbosity=1, debug=0, run_only=None):
       assert False
     layer.onsetup = assertFalse
 
+    from Products.ERP5Type.tests.utils import DbFactory
+    root_db_name, = cfg.dbtab.databases.keys()
+    DbFactory(root_db_name).addMountPoint('/')
+
   TestRunner = backportUnittest.TextTestRunner
 
   import Lifetime

product/ERP5Type/tests/utils.py

@@ -35,7 +35,11 @@ import random
 import socket
 import sys
 import transaction
+import ZODB
 import zLOG
+from App.config import getConfiguration
+from ZConfig.matcher import SectionValue
+from Zope2.Startup.datatypes import ZopeDatabase
 import Products.ERP5Type
 from Products.MailHost.MailHost import MailHost
 from email import message_from_string
@@ -337,6 +341,34 @@ def createZServer(log=os.devnull, zserver_type='http'):
         raise
       hs.close()
 
+class DbFactory(ZopeDatabase):
+
+  def __init__(self, name, storage=None, **kw):
+    ZopeDatabase.__init__(self, SectionValue({'container_class': None,
+                                              'mount_points': [],
+                                             }, name, None))
+    if storage is not None:
+      self.open = lambda database_name, databases: ZODB.DB(storage,
+        database_name=database_name, databases=databases, **kw)
+    getConfiguration().dbtab.db_factories[name] = self
+
+  @staticmethod
+  def get(*args, **kw):
+    return getConfiguration().dbtab.getDatabaseFactory(*args, **kw)
+
+  def addMountPoint(self, *mount_points):
+    dbtab = getConfiguration().dbtab
+    self.config.mount_points += mount_points
+    for mount_point in self.getVirtualMountPaths():
+      dbtab.mount_paths[mount_point] = self.name
+
+  def close(self):
+    dbtab = getConfiguration().dbtab
+    for mount_point in self.getVirtualMountPaths():
+      del dbtab.mount_paths[mount_point]
+    del dbtab.db_factories[self.name]
+    dbtab.databases.pop(self.name).close()
+
 # decorators
 @simple_decorator
 def reindex(func):