Fix #17: report traversal problems (#18)

* Fix #17: report traversal problems

* Explicitely test `published` against `None` to allow for unusual `bool` convertions of `published`

* Ability to report problems caused by method calls (such as "manage_delObjects")

* add space before inline comment to get `flake8` happy

* Added tests for the recognition of traversal and method exceptions

CHANGES.rst

@@ -4,6 +4,11 @@ Changelog
 5.2 (unreleased)
 ----------------
 
+- Ability to report problems caused by method calls (such as "manage_delObjects").
+
+- Ability to report traversal problems
+  (`#17 <https://github.com/zopefoundation/Products.SiteErrorLog/issues/17>`_)
+
 - Specify supported Python versions using ``python_requires`` in setup.py
   (`Zope#481 <https://github.com/zopefoundation/Zope/issues/481>`_)
 

src/Products/SiteErrorLog/SiteErrorLog.py

@@ -346,10 +346,17 @@ def IPubFailureSubscriber(event):
         closest to the published object that the error occured with,
         it logs no error if no published object was found.
     """
-    published = event.request.get('PUBLISHED')
-    if not published:
+    request = event.request
+    published = request.get('PUBLISHED')
+    if published is None:  # likely a traversal problem
+        parents = request.get("PARENTS")
+        if parents:
+            # partially emulate successful traversal
+            published = request["PUBLISHED"] = parents.pop(0)
+    if published is None:
         return
 
+    published = getattr(published, "__self__", published)  # method --> object
     try:
         error_log = aq_acquire(published, '__error_log__', containment=1)
     except AttributeError:

src/Products/SiteErrorLog/tests/testSiteErrorLog.py

@@ -17,16 +17,19 @@
 import unittest
 
 import transaction
+from OFS.Folder import manage_addFolder, Folder
 from Testing.makerequest import makerequest
 import Testing.testbrowser
 import Testing.ZopeTestCase
 from zope.component import adapter, provideHandler
+from zope.component.globalregistry import globalSiteManager
 from zope.event import notify
 from ZPublisher.pubevents import PubFailure
+from ZPublisher.WSGIPublisher import publish, transaction_pubevents
 import Zope2
 import Zope2.App
 
-from ..SiteErrorLog import manage_addErrorLog
+from ..SiteErrorLog import manage_addErrorLog, IPubFailureSubscriber
 
 from Products.SiteErrorLog.interfaces import IErrorRaisedEvent
 
@@ -218,3 +221,108 @@ def testSubmitRetainsIgnoredExceptionsUnchanged(self):
         ignoredExceptions = self.browser.getControl(
             label='Ignored exception types')
         self.assertEqual(ignoredExceptions.value, 'Unauthorized\nFnord')
+
+
+class WsgiErrorlogIntegrationLayer(Testing.ZopeTestCase.layer.ZopeLite):
+    """The tests using this layer commit transactions. Therefore,
+    we avoid persistent changes there and build the complete
+    support structure in this layer.
+    """
+    @classmethod
+    def setUp(cls):
+        # Apparently, other tests have already registered the
+        #   handler below, even twice
+        #   Clean up those registrations and
+        #   make a single clean registration
+        #   remember when we must remove our registration
+        regs = [r for r in globalSiteManager.registeredHandlers()
+                if r.factory is IPubFailureSubscriber
+                ]
+        if regs:
+            globalSiteManager.unregisterHandler(IPubFailureSubscriber)
+        cls._unregister = not regs
+        globalSiteManager.registerHandler(IPubFailureSubscriber)
+        # Set up our test structure
+        #   /
+        #     sel_f1/
+        #             error_log
+        #             sel_f2/
+        #                    error_log
+        app = Testing.ZopeTestCase.app()
+        # first level folder
+        manage_addFolder(app, "sel_f1")
+        sel_f1 = app.sel_f1
+        # second level folder
+        manage_addFolder(sel_f1, "sel_f2")
+        sel_f2 = sel_f1.sel_f2
+        # put an error log in each of those folders
+        # (used in `test_correct_log_*`)
+        for f in (sel_f1, sel_f2):
+            manage_addErrorLog(f)
+            el = f.error_log
+            el._ignored_exceptions = ()  # do not ignore exceptions
+        transaction.commit()
+        # make `manage_delObjects` temporarily public
+        cls._saved_roles = Folder.manage_delObjects__roles__
+        Folder.manage_delObjects__roles__ = None  # public
+
+    @classmethod
+    def tearDown(cls):
+        Folder.manage_delObjects__roles__ = cls._saved_roles
+        if cls._unregister:
+            globalSiteManager.unregisterHandler(IPubFailureSubscriber)
+        app = Testing.ZopeTestCase.app()
+        app._delOb("sel_f1")
+        transaction.commit()
+
+
+class WsgiErrorlogIntegrationTests(Testing.ZopeTestCase.ZopeTestCase):
+    layer = WsgiErrorlogIntegrationLayer
+
+    # we override `ZopeTestCase.setUp` by purpose
+    def setUp(self):
+        app = self.app = Testing.ZopeTestCase.app()
+        self.f1 = app.sel_f1
+        self.f2 = self.f1.sel_f2
+        self.el1 = self.f1.error_log
+        self.el2 = self.f2.error_log
+
+    # overridden by purpose
+    def tearDown(self):
+        self._clear_els()
+
+    def _clear_els(self):
+        for el in (self.el1, self.el2):
+            el._getLog()[:] = []
+
+    def _get_el_nos(self):
+        return tuple(len(el._getLog()) for el in (self.el1, self.el2))
+
+    def test_correct_log_traversal_2(self):
+        self._request("sel_f1/sel_f2/missing")
+        self.assertEqual(self._get_el_nos(), (0, 1))
+
+    def test_correct_log_traversal_1(self):
+        self._request("sel_f1/missing")
+        self.assertEqual(self._get_el_nos(), (1, 0))
+
+    def test_correct_log_method_2(self):
+        self._request("sel_f1/sel_f2/manage_delObjects", dict(id="missing"))
+        self.assertEqual(self._get_el_nos(), (0, 1))
+
+    def test_correct_log_method_1(self):
+        self._request("sel_f1/manage_delObjects", dict(id="missing"))
+        self.assertEqual(self._get_el_nos(), (1, 0))
+
+    def _request(self, url, params=None):
+        app = self.app
+        request = app.REQUEST
+        request.environ["PATH_INFO"] = url
+        if params:
+            request.form.update(params)
+            request.other.update(params)
+        try:
+            with transaction_pubevents(request, request.response):
+                publish(request, (app, "Zope", False))
+        except Exception:
+            pass