Commit 8808bedb authored by Julien Muchembled's avatar Julien Muchembled

Stop getting preferred archive for some SQL queries when there's no archive tool

With complex security scripts to calculate user groups,
calling getPreferredArchive() from DA.__call__ still causes
"maximum recursion depth exceeded" errors when caches are empty.

In order to address the issue, we must review erp5_archive. Instead of that,
this commit first checks whether the archive tool is installed, which is anyway
faster that getting a preference. An expected failing test is also added.
parent 1aa8b2a6
......@@ -27,15 +27,17 @@
#
##############################################################################
import time
import unittest
from Testing import ZopeTestCase
from AccessControl.SecurityManagement import newSecurityManager
from Testing.ZopeTestCase import _print, PortalTestCase
from AccessControl.SecurityManagement import \
getSecurityManager, newSecurityManager
from zLOG import LOG
from DateTime import DateTime
from Products.ERP5Type.tests.utils import getExtraSqlConnectionStringList
from Products.ERP5.tests.testInventoryAPI import InventoryAPITestCase
from Products.ERP5Type.tests.utils import reindex
from Products.ERP5Type.tests.utils import reindex, createZODBPythonScript
class TestArchive(InventoryAPITestCase):
......@@ -133,7 +135,7 @@ class TestArchive(InventoryAPITestCase):
if not run: return
if not quiet:
message = 'Archive'
ZopeTestCase._print('\n%s ' % message)
_print('\n%s ' % message)
LOG('Testing... ',0,message)
portal = self.getPortal()
......@@ -342,9 +344,42 @@ class TestArchive(InventoryAPITestCase):
# check the current archive
self.assertEqual(portal_archive.getCurrentArchive(), dest)
def test_MaximumRecursionDepthExceededWithComplexSecurity(self):
skin = self.portal.portal_skins.custom
colour = self.portal.portal_categories.colour
colour.hasObject('green') or colour.newContent('green')
login = str(time.time())
script_id = ["ERP5Type_getSecurityCategoryMapping",
"ERP5Type_getSecurityCategory"]
createZODBPythonScript(skin, script_id[0], "",
"return ((%r, ('colour',)),)" % script_id[1])
createZODBPythonScript(skin, script_id[1],
"base_category_list, user_name, object, portal_type, depth=[]", """if 1:
# This should not be called recursively, or at least if should not fail.
# Because RuntimeError is catched by 'except:' clauses, we detect it
# with a static variable.
depth.append(None)
assert not portal_type, portal_type
# the following line calls Base_zSearchRelatedObjectsByCategoryList
object.getSourceDecisionRelatedValueList()
bc, = base_category_list
depth.pop()
return [] if depth else [{bc: 'green'}]
""")
person = self.portal.person_module.newContent(reference=login)
try:
self.tic()
PortalTestCase.login(self, login)
self.assertEqual(['green'], getSecurityManager().getUser().getGroups())
self.portal.portal_caches.clearAllCache()
PortalTestCase.login(self, login)
unittest.expectedFailure(self.assertEqual)(
['green'], getSecurityManager().getUser().getGroups())
finally:
skin.manage_delObjects(script_id)
self.commit()
def test_suite():
suite = unittest.TestSuite()
suite.addTest(unittest.makeSuite(TestArchive))
return suite
......@@ -162,20 +162,17 @@ def DA__call__(self, REQUEST=None, __ick__=None, src__=0, test__=0, **kw):
# Patch to implement dynamic connection id
# Connection id is retrieve from user preference
if c is None:
physical_path = self.getPhysicalPath()
# XXX cleaner solution will be needed
if 'portal_catalog' not in physical_path and\
'cmf_activity' not in self.connection_id and\
'transactionless' not in self.connection_id:
try:
archive_id = self.portal_preferences.getPreferredArchive()
except AttributeError:
pass
else:
if archive_id not in (None, ''):
if not (self.connection_id in ('cmf_activity_sql_connection',
'erp5_sql_transactionless_connection')
or 'portal_catalog' in self.getPhysicalPath()):
portal = self.getPortalObject()
if 'portal_archives' in portal.__dict__:
archive_id = portal.portal_preferences.getPreferredArchive()
if archive_id:
archive_id = archive_id.split('/')[-1]
#LOG("DA__call__, archive_id 2", 300, archive_id)
archive = self.portal_archives._getOb(archive_id, None)
archive = portal.portal_archives._getOb(archive_id, None)
if archive is not None:
c = archive.getConnectionId()
#LOG("DA call", INFO, "retrieved connection %s from preference" %(c,))
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment