Skip to content

Commit

Permalink
- stricter flake8 configuration
Browse files Browse the repository at this point in the history
  • Loading branch information
dataflake committed Apr 13, 2019
1 parent 92cc5df commit ccdaf0a
Show file tree
Hide file tree
Showing 7 changed files with 127 additions and 75 deletions.
2 changes: 1 addition & 1 deletion .gitignore
Expand Up @@ -8,8 +8,8 @@ bin/
coverage.xml
develop-eggs/
dist/
htmlcov/
include/
lib/
local/
parts/
src/Products.SiteErrorLog.egg-info/
40 changes: 36 additions & 4 deletions setup.cfg
Expand Up @@ -3,9 +3,41 @@ ignore =
*.cfg
bootstrap.py

[flake8]
ignore = N801,N802,N803,N805,N806,N812,E301,W503
exclude = bootstrap.py

[bdist_wheel]
universal = 1

[isort]
force_single_line = True
combine_as_imports = True
sections = FUTURE,STDLIB,THIRDPARTY,ZOPE,FIRSTPARTY,LOCALFOLDER
known_third_party = six
default_section = ZOPE
line_length = 79
lines_after_imports = 2
not_skip =
__init__.py

[flake8]
ignore =
# W503 line break before binary operator: is no longer requested by PEP-8
W503,
no-accept-encodings = True
doctests = True
htmldir = parts/flake8
exclude =
bootstrap.py

[coverage:run]
branch = True
source = src
omit =

[coverage:report]
fail_under = 85.00
ignore_errors = True
precision = 2
show_missing = False
sort = Name

[coverage:html]
directory = parts/coverage
68 changes: 31 additions & 37 deletions src/Products/SiteErrorLog/SiteErrorLog.py
Expand Up @@ -14,33 +14,37 @@
"""Site error log module.
"""

import logging
import os
import sys
import time
import logging
from random import random
try:
# Python 3
from _thread import allocate_lock
except ImportError:
# Python 2
from thread import allocate_lock
from zope.event import notify
from zope.component import adapter
from .interfaces import ErrorRaisedEvent

from AccessControl.class_init import InitializeClass
from AccessControl.SecurityInfo import ClassSecurityInfo
from AccessControl.SecurityManagement import getSecurityManager
from AccessControl.unauthorized import Unauthorized
from Acquisition import aq_base
from Acquisition import aq_acquire
from Acquisition import aq_base
from App.Dialogs import MessageDialog
from OFS.SimpleItem import SimpleItem
from Products.PageTemplates.PageTemplateFile import PageTemplateFile
from zExceptions.ExceptionFormatter import format_exception
from zope.component import adapter
from zope.event import notify
from ZPublisher.interfaces import IPubFailure

from .interfaces import ErrorRaisedEvent


try:
# Python 3
from _thread import allocate_lock
except ImportError:
# Python 2
from thread import allocate_lock


LOG = logging.getLogger('Zope.SiteErrorLog')

# Permission names
Expand Down Expand Up @@ -99,23 +103,21 @@ class SiteErrorLog(SimpleItem):
{'label': 'Log', 'action': 'manage_main'},
) + SimpleItem.manage_options

security.declareProtected(use_error_logging, 'manage_main')
security.declareProtected(use_error_logging, 'manage_main') # NOQA: D001
manage_main = PageTemplateFile('main.pt', _www)

security.declareProtected(use_error_logging, 'showEntry')
security.declareProtected(use_error_logging, 'showEntry') # NOQA: D001
showEntry = PageTemplateFile('showEntry.pt', _www)

security.declarePrivate('manage_beforeDelete')

@security.private
def manage_beforeDelete(self, item, container):
if item is self:
try:
del container.__error_log__
except AttributeError:
pass

security.declarePrivate('manage_afterAdd')

@security.private
def manage_afterAdd(self, item, container):
if item is self:
container.__error_log__ = aq_base(self)
Expand All @@ -138,8 +140,7 @@ def _getLog(self):
temp_logs[self._p_oid] = log
return log

security.declareProtected(use_error_logging, 'forgetEntry')

@security.protected(use_error_logging)
def forgetEntry(self, id, REQUEST=None):
"""Removes an entry from the error log."""
log = self._getLog()
Expand All @@ -161,8 +162,7 @@ def forgetEntry(self, id, REQUEST=None):
# through-the-web.
_ignored_exceptions = ('Unauthorized', 'NotFound', 'Redirect')

security.declarePrivate('raising')

@security.private
def raising(self, info):
"""Log an exception.
Expand Down Expand Up @@ -250,25 +250,22 @@ def _do_copy_to_zlog(self, now, strtype, entry_id, url, tb_text):
_rate_restrict_pool[strtype] = next_when
LOG.error('%s %s\n%s' % (entry_id, url, tb_text.rstrip()))

security.declareProtected(use_error_logging, 'getProperties')

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

security.declareProtected(log_to_event_log, 'checkEventLogPermission')

@security.protected(log_to_event_log)
def checkEventLogPermission(self):
if not getSecurityManager().checkPermission(log_to_event_log, self):
raise Unauthorized('You do not have the "%s" permission.' %
log_to_event_log)
return 1

security.declareProtected(use_error_logging, 'setProperties')

@security.protected(use_error_logging)
def setProperties(self, keep_entries, copy_to_zlog=0,
ignored_exceptions=(), RESPONSE=None):
"""Sets the properties of this site error log.
Expand All @@ -288,8 +285,7 @@ def setProperties(self, keep_entries, copy_to_zlog=0,
'%s/manage_main?manage_tabs_message=Changed+properties.' %
self.absolute_url())

security.declareProtected(use_error_logging, 'getLogEntries')

@security.protected(use_error_logging)
def getLogEntries(self):
"""Returns the entries in the log, most recent first.
Expand All @@ -300,8 +296,7 @@ def getLogEntries(self):
res.reverse()
return res

security.declareProtected(use_error_logging, 'getLogEntryById')

@security.protected(use_error_logging)
def getLogEntryById(self, id):
"""Returns the specified log entry.
Expand All @@ -312,8 +307,7 @@ def getLogEntryById(self, id):
return entry.copy()
return None

security.declareProtected(use_error_logging, 'getLogEntryAsText')

@security.protected(use_error_logging)
def getLogEntryAsText(self, id, RESPONSE=None):
"""Returns the specified log entry.
Expand Down Expand Up @@ -350,14 +344,14 @@ def IPubFailureSubscriber(event):
request = event.request
published = request.get('PUBLISHED')
if published is None: # likely a traversal problem
parents = request.get("PARENTS")
parents = request.get('PARENTS')
if parents:
# partially emulate successful traversal
published = request["PUBLISHED"] = parents.pop(0)
published = request['PUBLISHED'] = parents.pop(0)
if published is None:
return

published = getattr(published, "__self__", published) # method --> object
published = getattr(published, '__self__', published) # method --> object
try:
error_log = aq_acquire(published, '__error_log__', containment=1)
except AttributeError:
Expand Down
3 changes: 1 addition & 2 deletions src/Products/SiteErrorLog/interfaces.py
@@ -1,6 +1,5 @@
# -*- coding: utf-8 -*-
from zope.interface import implementer
from zope.interface import Interface
from zope.interface import implementer


class IErrorRaisedEvent(Interface):
Expand Down
7 changes: 5 additions & 2 deletions src/Products/SiteErrorLog/tests/testInitialization.py
Expand Up @@ -18,10 +18,13 @@
import unittest

import Products
from App.config import getConfiguration, setConfiguration
from OFS.Application import Application, AppInitializer
from App.config import getConfiguration
from App.config import setConfiguration
from OFS.Application import AppInitializer
from OFS.Application import Application
from Zope2.Startup.options import ZopeWSGIOptions


test_cfg = """
instancehome {instance_home}
Expand Down
47 changes: 25 additions & 22 deletions src/Products/SiteErrorLog/tests/testSiteErrorLog.py
Expand Up @@ -16,22 +16,25 @@
import sys
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
import transaction
import Zope2
import Zope2.App
from OFS.Folder import Folder
from OFS.Folder import manage_addFolder
from Testing.makerequest import makerequest
from zope.component import adapter
from zope.component import 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, IPubFailureSubscriber
from ZPublisher.WSGIPublisher import publish
from ZPublisher.WSGIPublisher import transaction_pubevents

from Products.SiteErrorLog.interfaces import IErrorRaisedEvent
from ..interfaces import IErrorRaisedEvent
from ..SiteErrorLog import IPubFailureSubscriber
from ..SiteErrorLog import manage_addErrorLog


class SiteErrorLogTests(unittest.TestCase):
Expand Down Expand Up @@ -120,15 +123,15 @@ def testForgetException(self):

# Create a predictable error
try:
raise AttributeError("DummyAttribute")
raise AttributeError('DummyAttribute')
except AttributeError:
self.app.REQUEST['PUBLISHED'] = elog
notify(PubFailure(self.app.REQUEST, sys.exc_info(), False))

previous_log_length = len(elog.getLogEntries())

entries = elog.getLogEntries()
self.assertEqual(entries[0]['value'], "DummyAttribute")
self.assertEqual(entries[0]['value'], 'DummyAttribute')

# Kick it
elog.forgetEntry(entries[0]['id'])
Expand Down Expand Up @@ -170,7 +173,7 @@ def testEntryID(self):

# Create a predictable error
try:
raise AttributeError("DummyAttribute")
raise AttributeError('DummyAttribute')
except AttributeError:
self.app.REQUEST['PUBLISHED'] = elog
notify(PubFailure(self.app.REQUEST, sys.exc_info(), False))
Expand Down Expand Up @@ -250,10 +253,10 @@ def setUp(cls):
# error_log
app = Testing.ZopeTestCase.app()
# first level folder
manage_addFolder(app, "sel_f1")
manage_addFolder(app, 'sel_f1')
sel_f1 = app.sel_f1
# second level folder
manage_addFolder(sel_f1, "sel_f2")
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_*`)
Expand All @@ -272,7 +275,7 @@ def tearDown(cls):
if cls._unregister:
globalSiteManager.unregisterHandler(IPubFailureSubscriber)
app = Testing.ZopeTestCase.app()
app._delOb("sel_f1")
app._delOb('sel_f1')
transaction.commit()


Expand All @@ -299,30 +302,30 @@ 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._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._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._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._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
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))
publish(request, (app, 'Zope', False))
except Exception:
pass

0 comments on commit ccdaf0a

Please sign in to comment.