[Zope3-checkins] CVS: Zope3/src/zope/app/errorservice - __init__.py:1.1 configure.zcml:1.1 error.txt:1.1 interfaces.py:1.1 tests.py:1.1

Stephan Richter srichter at cosmos.phy.tufts.edu
Wed Mar 10 08:43:36 EST 2004


Update of /cvs-repository/Zope3/src/zope/app/errorservice
In directory cvs.zope.org:/tmp/cvs-serv4477/src/zope/app/errorservice

Added Files:
	__init__.py configure.zcml error.txt interfaces.py tests.py 
Log Message:


Moved error reporting service to zope.app.errorservice. Provided
modulealiases, so that your old ZODB will still work.


=== Added File Zope3/src/zope/app/errorservice/__init__.py ===
##############################################################################
#
# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Error reporting service

This is a port of the Zope 2 error reporting object

$Id: __init__.py,v 1.1 2004/03/10 13:43:35 srichter Exp $
"""
import time
import logging
from persistent import Persistent
from random import random
from thread import allocate_lock
from types import StringTypes

from zope.exceptions.exceptionformatter import format_exception
from zope.interface import implements
from zope.app.interfaces.services.service import ISimpleService
from zope.app.container.contained import Contained
from interfaces import IErrorReportingService, ILocalErrorReportingService

#Restrict the rate at which errors are sent to the Event Log
_rate_restrict_pool = {}

# The number of seconds that must elapse on average between sending two
# exceptions of the same name into the the Event Log. one per minute.
_rate_restrict_period = 60

# The number of exceptions to allow in a burst before the above limit
# kicks in. We allow five exceptions, before limiting them to one per
# minute.
_rate_restrict_burst = 5

# _temp_logs holds the logs.
_temp_logs = {}  # { oid -> [ traceback string ] }

cleanup_lock = allocate_lock()

class ErrorReportingService(Persistent, Contained):
    """Error Reporting Service
    """
    implements(IErrorReportingService,
               ILocalErrorReportingService,
               ISimpleService,
               )

    keep_entries = 20
    copy_to_zlog = 0
    _ignored_exceptions = ('Unauthorized',)


    def _getLog(self):
        """Returns the log for this object.
        Careful, the log is shared between threads.
        """
        log = _temp_logs.get(self._p_oid, None)
        if log is None:
            log = []
            _temp_logs[self._p_oid] = log
        return log

    # Exceptions that happen all the time, so we dont need
    # to log them. Eventually this should be configured
    # through-the-web.
    def raising(self, info, request=None):
        """Log an exception.
        Called by ZopePublication.handleException method.
        """
        now = time.time()
        try:
            tb_text = None
            tb_html = None

            strtype = str(getattr(info[0], '__name__', info[0]))
            if strtype in self._ignored_exceptions:
                return

            if not isinstance(info[2], StringTypes):
                tb_text = ''.join(
                        format_exception(*info, **{'as_html': 0}))
                tb_html = ''.join(
                    format_exception(*info, **{'as_html': 1}))
            else:
                tb_text = info[2]

            url = None
            username = None
            req_html = None
            if request:
                # XXX: Temporary fix, which Steve should undo. URL is
                #      just too HTTPRequest-specific.
                if hasattr(request, 'URL'):
                    url = request.URL
                try:
                    # XXX: UnauthenticatedPrincipal does not have getLogin()
                    if hasattr(request.user, 'getLogin'):
                        login = request.user.getLogin()
                    else:
                        login = 'unauthenticated'
                    username = ', '.join(map(str, (login,
                                          request.user.id,
                                          request.user.title,
                                          request.user.description
                                         )))
                # When there's an unauthorized access, request.user is
                # not set, so we get an AttributeError
                # XXX is this right? Surely request.user should be set!
                # XXX Answer: Catching AttributeError is correct for the
                #             simple reason that UnauthenticatedUser (which
                #             I always use during coding), has no 'getLogin()'
                #             method. However, for some reason this except
                #             does **NOT** catch these errors.
                except AttributeError:
                    pass

                req_html = ''.join(['%s : %s<br>' % item
                                    for item in request.items()])
            try:
                strv = str(info[1])
            # A call to str(obj) could raise anything at all.
            # We'll ignore these errors, and print something
            # useful instead, but also log the error.
            except:
                logging.getLogger('SiteError').exception(
                    'Error in ErrorReportingService while getting a str '
                    'representation of an object')
                strv = '<unprintable %s object>' % (
                        str(type(info[1]).__name__)
                        )

            log = self._getLog()
            entry_id = str(now) + str(random()) # Low chance of collision

            log.append({
                'type': strtype,
                'value': strv,
                'time': time.ctime(now),
                'id': entry_id,
                'tb_text': tb_text,
                'tb_html': tb_html,
                'username': username,
                'url': url,
                'req_html': req_html,
                })
            cleanup_lock.acquire()
            try:
                if len(log) >= self.keep_entries:
                    del log[:-self.keep_entries]
            finally:
                cleanup_lock.release()

            if self.copy_to_zlog:
                self._do_copy_to_zlog(now, strtype, str(url), info)
        finally:
            info = None

    def _do_copy_to_zlog(self, now, strtype, url, info):
        # XXX info is unused; logging.exception() will call sys.exc_info()
        # work around this with an evil hack
        when = _rate_restrict_pool.get(strtype,0)
        if now > when:
            next_when = max(when,
                            now - _rate_restrict_burst*_rate_restrict_period)
            next_when += _rate_restrict_period
            _rate_restrict_pool[strtype] = next_when
            try:
                raise info[0], info[1], info[2]
            except:
                logging.getLogger('SiteError').exception(str(url))

    def getProperties(self):
        return {
            'keep_entries': self.keep_entries,
            'copy_to_zlog': self.copy_to_zlog,
            'ignored_exceptions': self._ignored_exceptions,
            }

    def setProperties(self, keep_entries, copy_to_zlog=0,
                      ignored_exceptions=()):
        """Sets the properties of this site error log.
        """
        copy_to_zlog = bool(copy_to_zlog)
        self.keep_entries = int(keep_entries)
        self.copy_to_zlog = copy_to_zlog
        self._ignored_exceptions = tuple(
                filter(None, map(str, ignored_exceptions))
                )

    def getLogEntries(self):
        """Returns the entries in the log, most recent first.

        Makes a copy to prevent changes.
        """
        res = [entry.copy() for entry in self._getLog()]
        res.reverse()
        return res

    def getLogEntryById(self, id):
        """Returns the specified log entry.
        Makes a copy to prevent changes.  Returns None if not found.
        """
        for entry in self._getLog():
            if entry['id'] == id:
                return entry.copy()
        return None

class RootErrorReportingService(ErrorReportingService):
    rootId = 'root'
    
    def _getLog(self):
        """Returns the log for this object.
        Careful, the log is shared between threads.
        """
        log = _temp_logs.get(self.rootId, None)
        if log is None:
            log = []
            _temp_logs[self.rootId] = log
        return log

globalErrorReportingService = RootErrorReportingService()

def _cleanup_temp_log():
    _temp_logs.clear()

_clear = _cleanup_temp_log
# Register our cleanup with Testing.CleanUp to make writing unit tests simpler.
from zope.testing.cleanup import addCleanUp
addCleanUp(_clear)
del addCleanUp


=== Added File Zope3/src/zope/app/errorservice/configure.zcml ===
<configure xmlns="http://namespaces.zope.org/zope">

  <!-- For backward compatibility -->

  <modulealias
      module="zope.app.errorservice"
      alias="zope.app.services.error"
      />

  <modulealias
      module="zope.app.errorservice"
      alias="zope.app.services.errorr"
      />

  <modulealias
      module=".interfaces"
      alias="zope.app.interfaces.services.error"
      />

  <!-- Error Reporting Service -->

  <serviceType
      id="ErrorLogging"
      interface=".interfaces.IErrorReportingService" 
      />

  <content class=".ErrorReportingService">
    <factory
        id="ErrorLogging"
        />
    <require
        permission="zope.Public"
        interface=".interfaces.IErrorReportingService"
        />
    <require
        permission="zope.ManageServices"
        interface=".interfaces.ILocalErrorReportingService"
        />
    </content>

  <content class=".RootErrorReportingService">
    <require
        permission="zope.Public"
        interface=".interfaces.IErrorReportingService"
        />
    <require
        permission="zope.ManageServices"
        interface=".interfaces.ILocalErrorReportingService"
        />
    </content>

  <service 
      serviceType="ErrorLogging"
      permission="zope.Public"
      component=".globalErrorReportingService" 
      />

  <include package=".browser" />

</configure>

=== Added File Zope3/src/zope/app/errorservice/error.txt ===
Under construction.


=== Added File Zope3/src/zope/app/errorservice/interfaces.py ===
##############################################################################
#
# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Error Reporting Service interfaces

$Id: interfaces.py,v 1.1 2004/03/10 13:43:35 srichter Exp $
"""
from zope.interface import Interface

class IErrorReportingService(Interface):
    """Error Reporting Service
    """

    def raising(info, request=None):
        """Logs an exception.
        """

class ILocalErrorReportingService(Interface):
    """Local Error Reporting Service

    Included management functions
    """

    def getProperties():
        """Gets the properties as dictionary.

        keep_entries, copy_to_logfile, ignored_exceptions
        """

    def setProperties(keep_entries, copy_to_zlog=0, ignored_exceptions=(),
                      RESPONSE=None):
        """Sets the properties

        keep_entries, copy_to_logfile, ignored_exceptions
        """

    def getLogEntries():
        """Returns the entries in the log, most recent first.
        """

    def getLogEntryById(id):
        """Return LogEntry by ID
        """


=== Added File Zope3/src/zope/app/errorservice/tests.py ===
##############################################################################
#
# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Error Reporting Service Tests

$Id: tests.py,v 1.1 2004/03/10 13:43:35 srichter Exp $
"""
import sys
from unittest import TestCase, TestLoader, TextTestRunner
from zope.app.errorservice import ErrorReportingService
from zope.testing.cleanup import CleanUp
from zope.exceptions.exceptionformatter import format_exception


class C1:
    def getAnErrorInfo(self):
        exc_info = None
        try:
            someerror()
        except:
            exc_info = sys.exc_info()
        return exc_info


class ErrorReportingServiceTests(CleanUp, TestCase):
    def test_checkForEmpryLog(self):
        # Test Check Empty Log
        errService = ErrorReportingService()
        getProp = errService.getLogEntries()
        self.failIf(getProp)

    def test_checkProperties(self):
        # Test Properties test
        errService = ErrorReportingService()
        setProp = {
            'keep_entries':10,
            'copy_to_zlog':1,
            'ignored_exceptions':()
            }
        errService.setProperties(**setProp)
        getProp = errService.getProperties()
        self.assertEqual(setProp, getProp)

    def test_ErrorLog(self):
        # Test for Logging Error.  Create one error and check whether its
        # logged or not.
        errService = ErrorReportingService()
        exc_info = C1().getAnErrorInfo()
        errService.raising(exc_info)
        getErrLog = errService.getLogEntries()
        self.assertEquals(1, len(getErrLog))

        tb_text = ''.join(format_exception(*exc_info, **{'as_html': 0}))

        err_id =  getErrLog[0]['id']
        self.assertEquals(tb_text,
                          errService.getLogEntryById(err_id)['tb_text'])




def test_suite():
    loader=TestLoader()
    return loader.loadTestsFromTestCase(ErrorReportingServiceTests)

if __name__=='__main__':
    TextTestRunner().run(test_suite())




More information about the Zope3-Checkins mailing list