Skip to content

Z
Zope
Commit 9c6ed42d authored by Hanno Schlichting's avatar Hanno Schlichting
Browse files
Options

flake8

parent 4d387766
Hide whitespace changes
Inline Side-by-side
Showing with 57 additions and 62 deletions
src/Zope2/ClassFactory.py
View file @ 9c6ed42d
......@@ -14,4 +14,4 @@
"""
# Stub in case anyone depends on this
from App.ClassFactory import ClassFactory
from App.ClassFactory import ClassFactory # NOQA
src/Zope2/Startup/datatypes.py
View file @ 9c6ed42d
......@@ -13,37 +13,35 @@
##############################################################################
"""Datatypes for the Zope schema for use with ZConfig."""
import cStringIO
import os
from UserDict import UserDict
import traceback
from ZConfig.components.logger import logger
from ZODB.config import ZODBDatabase
# generic datatypes
def security_policy_implementation(value):
value = value.upper()
ok = ('PYTHON', 'C')
if value not in ok:
raise ValueError, (
"security-policy-implementation must be one of %s" % repr(ok)
)
raise ValueError(
"security-policy-implementation must be one of %r" % ok)
return value
def datetime_format(value):
value = value.lower()
ok = ('us', 'international')
if value not in ok:
raise ValueError, (
"datetime-format must be one of %r" % repr(ok)
)
raise ValueError("datetime-format must be one of %r" % ok)
return value
def cgi_environment(section):
return section.environ
# Datatype for the access and trace logs
# (the loghandler datatypes come from the zLOG package)
class LoggerFactory(logger.LoggerFactory):
"""
......@@ -59,25 +57,25 @@ class LoggerFactory(logger.LoggerFactory):
section.propagate = False
logger.LoggerFactory.__init__(self, section)
# DNS resolver
def dns_resolver(hostname):
# DNS resolver
from ZServer.medusa import resolver
return resolver.caching_resolver(hostname)
# mount-point definition
def mount_point(value):
# mount-point definition
if not value:
raise ValueError, 'mount-point must not be empty'
raise ValueError('mount-point must not be empty')
if not value.startswith('/'):
raise ValueError, ("mount-point '%s' is invalid: mount points must "
"begin with a slash" % value)
raise ValueError("mount-point '%s' is invalid: mount points must "
"begin with a slash" % value)
return value
# A datatype that converts a Python dotted-path-name to an object
def importable_name(name):
# A datatype that converts a Python dotted-path-name to an object
try:
components = name.split('.')
start = components[0]
......@@ -93,31 +91,31 @@ def importable_name(name):
package = __import__(n, g, g, component)
return package
except ImportError:
import traceback, cStringIO
IO = cStringIO.StringIO()
traceback.print_exc(file=IO)
raise ValueError(
'The object named by "%s" could not be imported\n%s' % (name, IO.getvalue()))
'The object named by "%s" could not be imported\n%s' % (
name, IO.getvalue()))
# A datatype that ensures that a dotted path name can be resolved but
# returns the name instead of the object
def python_dotted_path(name):
ob = importable_name(name) # will fail in course
# A datatype that ensures that a dotted path name can be resolved but
# returns the name instead of the object
ob = importable_name(name) # NOQA - will fail in course
return name
class zdaemonEnvironDict(UserDict):
# zdaemon 2 expects to use a 'mapping' attribute of the environ object.
@property
def mapping(self):
return self.data
# Datatype for the root configuration object
# (default values for some computed paths, configures the dbtab)
def root_config(section):
# Datatype for the root configuration object
# (default values for some computed paths, configures the dbtab)
from ZConfig import ConfigurationError
from ZConfig.matcher import SectionValue
if section.environment is None:
......@@ -136,8 +134,8 @@ def root_config(section):
if not section.databases:
section.databases = []
mount_factories = {} # { name -> factory}
mount_points = {} # { virtual path -> name }
mount_factories = {} # { name -> factory}
mount_points = {} # { virtual path -> name }
dup_err = ('Invalid configuration: ZODB databases named "%s" and "%s" are '
'both configured to use the same mount point, named "%s"')
......@@ -146,7 +144,7 @@ def root_config(section):
name = database.config.getSectionName()
mount_factories[name] = database
for point in points:
if mount_points.has_key(point):
if point in mount_points:
raise ConfigurationError(dup_err % (mount_points[point],
name, point))
mount_points[point] = name
......@@ -167,6 +165,7 @@ def root_config(section):
return section
class ZopeDatabase(ZODBDatabase):
""" A ZODB database datatype that can handle an extended set of
attributes for use by DBTab """
......@@ -223,6 +222,7 @@ class ZopeDatabase(ZODBDatabase):
return (real_root, real_path, container_class)
raise LookupError('Nothing known about mount path %s' % mount_path)
def default_zpublisher_encoding(value):
# This is a bit clunky but necessary :-(
# These modules are imported during the configuration process
......@@ -235,6 +235,7 @@ def default_zpublisher_encoding(value):
HTTPResponse.default_encoding = value
return value
class DBTab:
"""A Zope database configuration, similar in purpose to /etc/fstab.
"""
......@@ -249,17 +250,14 @@ class DBTab:
"""
return self.mount_paths.items()
def listDatabaseNames(self):
"""Returns a sequence of names.
"""
return self.db_factories.keys()
def hasDatabase(self, name):
"""Returns true if name is the name of a configured database."""
return self.db_factories.has_key(name)
return name in self.db_factories
def _mountPathError(self, mount_path):
from ZConfig import ConfigurationError
......@@ -285,7 +283,7 @@ class DBTab:
def getDatabaseFactory(self, mount_path=None, name=None):
if name is None:
name = self.getName(mount_path)
if not self.db_factories.has_key(name):
if name not in self.db_factories:
raise KeyError('%s is not a configured database' % repr(name))
return self.db_factories[name]
......@@ -295,12 +293,9 @@ class DBTab:
self._mountPathError(mount_path)
return name
# class factories (potentially) used by the class-factory parameter in
# zopeschema.xml
def minimalClassFactory(jar, module, name,
_silly=('__doc__',), _globals={},
):
_silly=('__doc__',), _globals={}):
"""Minimal class factory.
If any class is not found, this class factory will propagate
......@@ -309,9 +304,9 @@ def minimalClassFactory(jar, module, name,
m = __import__(module, _globals, _globals, _silly)
return getattr(m, name)
def simpleClassFactory(jar, module, name,
_silly=('__doc__',), _globals={},
):
_silly=('__doc__',), _globals={}):
"""Class factory.
"""
import OFS.Uninstalled
......
src/Zope2/Startup/tests/test_schema.py
View file @ 9c6ed42d
......@@ -24,18 +24,18 @@ import Products
from Zope2.Startup import datatypes
from App.config import getConfiguration
TEMPNAME = tempfile.mktemp()
TEMPPRODUCTS = os.path.join(TEMPNAME, "Products")
TEMPVAR = os.path.join(TEMPNAME, "var")
def getSchema():
startup = os.path.dirname(os.path.realpath(Zope2.Startup.__file__))
schemafile = os.path.join(startup, 'zopeschema.xml')
return ZConfig.loadSchema(schemafile)
class StartupTestCase(unittest.TestCase):
schema = None
......@@ -68,8 +68,6 @@ class StartupTestCase(unittest.TestCase):
return conf, handler
def test_load_config_template(self):
schema = self.schema
cfg = getConfiguration()
import Zope2.utilities
base = os.path.dirname(Zope2.utilities.__file__)
fn = os.path.join(base, "skel", "etc", "base.conf.in")
......@@ -102,7 +100,8 @@ class StartupTestCase(unittest.TestCase):
""")
items = conf.environment.items()
items.sort()
self.assertEqual(items, [("FEARFACTORY", "rocks"), ("NSYNC","doesnt")])
self.assertEqual(
items, [("FEARFACTORY", "rocks"), ("NSYNC", "doesnt")])
def test_ms_public_header(self):
from Zope2.Startup import config
......@@ -207,7 +206,3 @@ class StartupTestCase(unittest.TestCase):
default-zpublisher-encoding iso-8859-15
""")
self.assertEqual(conf.default_zpublisher_encoding, 'iso-8859-15')
def test_suite():
return unittest.makeSuite(StartupTestCase)
src/Zope2/Startup/tests/test_warnfilter.py
View file @ 9c6ed42d
......@@ -122,6 +122,3 @@ class TestWarnFilter(unittest.TestCase):
category A.Module.That.Doesnt.Exist
</warnfilter>
""")
def test_suite():
return unittest.makeSuite(TestWarnFilter)
src/Zope2/Startup/warnfilter.py
View file @ 9c6ed42d
......@@ -14,41 +14,44 @@
"""Datatypes for warning filter component """
import re
import warnings
def warn_category(category):
import re, types
if not category:
return Warning
if re.match("^[a-zA-Z0-9_]+$", category):
try:
cat = eval(category)
except NameError:
raise ValueError("unknown warning category: %s" % `category`)
raise ValueError("unknown warning category: %r" % category)
else:
i = category.rfind(".")
module = category[:i]
klass = category[i+1:]
klass = category[i + 1:]
try:
m = __import__(module, None, None, [klass])
except ImportError:
raise ValueError("invalid module name: %s" % `module`)
raise ValueError("invalid module name: %r" % module)
try:
cat = getattr(m, klass)
except AttributeError:
raise ValueError("unknown warning category: %s" % `category`)
raise ValueError("unknown warning category: %r" % category)
if (not isinstance(cat, type(Warning)) or
not issubclass(cat, Warning)):
raise ValueError("invalid warning category: %s" % `category`)
not issubclass(cat, Warning)):
raise ValueError("invalid warning category: %r" % category)
return cat
def warn_action(val):
OK = ("error", "ignore", "always", "default", "module", "once")
if val not in OK:
raise ValueError, "warning action %s not one of %s" % (val, OK)
raise ValueError("warning action %s not one of %s" % (val, OK))
return val
def warning_filter_handler(section):
import warnings
# add the warning filter
warnings.filterwarnings(section.action, section.message, section.category,
section.module, section.lineno)
......
src/Zope2/__init__.py
View file @ 9c6ed42d
......@@ -33,8 +33,13 @@
# old behavior is likely to cause problems as ZODB backends, like ZEO,
# gain new features.
import os
from Zope2.Startup.run import configure
_began_startup = 0
def startup():
"""Initialize the Zope Package and provide a published module"""
global _began_startup
......@@ -46,18 +51,19 @@ def startup():
from Zope2.App.startup import startup as _startup
_startup()
def app(*args, **kw):
"""Utility for scripts to open a connection to the database"""
startup()
return bobo_application(*args, **kw)
def debug(*args, **kw):
"""Utility to try a Zope request using the interactive interpreter"""
startup()
import ZPublisher
return ZPublisher.test('Zope2', *args, **kw)
from Zope2.Startup.run import configure
def _configure():
# Load configuration file from (optional) environment variable
......@@ -67,6 +73,7 @@ def _configure():
if configfile is not None:
configure(configfile)
# Zope2.App.startup.startup() sets the following variables in this module.
DB = None
bobo_application = None
......@@ -76,8 +83,6 @@ zpublisher_exception_hook = None
__bobo_before__ = None
import os
if os.environ.get('ZOPE_COMPATIBLE_STARTUP'):
# Open the database immediately (see comment above).
startup()
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
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7