[Zope3-checkins] CVS: Zope3/src/zope/app/process/tests - __init__.py:1.1.2.1 test_loggingconfig.py:1.1.2.1 test_registerrequestfactory.py:1.1.2.1 test_registerservertype.py:1.1.2.1 test_requestfactoryregistry.py:1.1.2.1 test_servertyperegistry.py:1.1.2.1 test_simpleregistry.py:1.1.2.1

Fred L. Drake, Jr. fred@zope.com
Mon, 23 Jun 2003 18:08:00 -0400


Update of /cvs-repository/Zope3/src/zope/app/process/tests
In directory cvs.zope.org:/tmp/cvs-serv24536/src/zope/app/process/tests

Added Files:
      Tag: fdrake-zconfig-in-zope-3-branch
	__init__.py test_loggingconfig.py 
	test_registerrequestfactory.py test_registerservertype.py 
	test_requestfactoryregistry.py test_servertyperegistry.py 
	test_simpleregistry.py 
Log Message:
Support ZConfig in Zope 3.

=== Added File Zope3/src/zope/app/process/tests/__init__.py ===
# Make this a package.


=== Added File Zope3/src/zope/app/process/tests/test_loggingconfig.py ===
##############################################################################
#
# Copyright (c) 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.
#
##############################################################################

"""Tests for zLOG configuration via ZConfig."""

import cStringIO as StringIO
import logging
import sys
import tempfile
import unittest

import ZConfig


from zope.app.process import datatypes
from zope.app.process import loghandlers


class TestLoggingConfig(unittest.TestCase):

    # XXX This tries to save and restore the state of logging around
    # the test.  Somewhat surgical; there may be a better way.

    def setUp(self):
        self._old_logger = logging.getLogger("event")
        self._old_level = self._old_logger.level
        self._old_handlers = self._old_logger.handlers[:]
        self._old_logger.handlers[:] = [loghandlers.NullHandler()]

    def tearDown(self):
        for h in self._old_logger.handlers:
            self._old_logger.removeHandler(h)
        for h in self._old_handlers:
            self._old_logger.addHandler(h)
        self._old_logger.setLevel(self._old_level)

    _schema = None
    _schematext = """
      <schema>
        <import package='zope.app.process'/>
        <section type='eventlog' name='*' attribute='eventlog'/>
      </schema>
    """

    def get_schema(self):
        if self._schema is None:
            sio = StringIO.StringIO(self._schematext)
            self.__class__._schema = ZConfig.loadSchemaFile(sio)
        return self._schema

    def get_config(self, text):
        conf, handler = ZConfig.loadConfigFile(self.get_schema(),
                                               StringIO.StringIO(text))
        self.assert_(not handler)
        return conf

    def test_logging_level(self):
        # Make sure the expected names are supported; it's not clear
        # how to check the values in a meaningful way.
        # Just make sure they're case-insensitive.
        convert = datatypes.logging_level
        for name in ["notset", "all", "trace", "debug", "blather",
                     "info", "warn", "warning", "error", "fatal",
                     "critical"]:
            self.assertEqual(convert(name), convert(name.upper()))
        self.assertRaises(ValueError, convert, "hopefully-not-a-valid-value")

    def test_http_method(self):
        convert = datatypes.get_or_post
        self.assertEqual(convert("get"), "GET")
        self.assertEqual(convert("GET"), "GET")
        self.assertEqual(convert("post"), "POST")
        self.assertEqual(convert("POST"), "POST")
        self.assertRaises(ValueError, convert, "")
        self.assertRaises(ValueError, convert, "foo")

    def test_syslog_facility(self):
        convert = datatypes.syslog_facility
        for name in ["auth", "authpriv", "cron", "daemon", "kern",
                     "lpr", "mail", "news", "security", "syslog",
                     "user", "uucp", "local0", "local1", "local2",
                     "local3", "local4", "local5", "local6", "local7"]:
            self.assertEqual(convert(name), name)
            self.assertEqual(convert(name.upper()), name)
        self.assertRaises(ValueError, convert, "hopefully-never-a-valid-value")

    def test_config_without_logger(self):
        conf = self.get_config("")
        self.assert_(conf.eventlog is None)

    def test_config_without_handlers(self):
        logger = self.check_simple_logger("<eventlog/>")
        # Make sure there's a NullHandler, since a warning gets
        # printed if there are no handlers:
        self.assertEqual(len(logger.handlers), 1)
        self.assert_(isinstance(logger.handlers[0],
                                loghandlers.NullHandler))

    def test_with_logfile(self):
        fn = tempfile.mktemp()
        logger = self.check_simple_logger("<eventlog>\n"
                                          "  <logfile>\n"
                                          "    path %s\n"
                                          "    level debug\n"
                                          "  </logfile>\n"
                                          "</eventlog>" % fn)
        logfile = logger.handlers[0]
        self.assertEqual(logfile.level, logging.DEBUG)
        self.assert_(isinstance(logfile, loghandlers.FileHandler))

    def test_with_stderr(self):
        self.check_standard_stream("stderr")

    def test_with_stdout(self):
        self.check_standard_stream("stdout")

    def check_standard_stream(self, name):
        old_stream = getattr(sys, name)
        conf = self.get_config("""
            <eventlog>
              <logfile>
                level info
                path %s
              </logfile>
            </eventlog>
            """ % name.upper())
        self.assert_(conf.eventlog is not None)
        # The factory has already been created; make sure it picks up
        # the stderr we set here when we create the logger and
        # handlers:
        sio = StringIO.StringIO()
        setattr(sys, name, sio)
        try:
            logger = conf.eventlog()
        finally:
            setattr(sys, name, old_stream)
        logger.warn("woohoo!")
        self.assert_(sio.getvalue().find("woohoo!") >= 0)

    def test_with_syslog(self):
        logger = self.check_simple_logger("<eventlog>\n"
                                          "  <syslog>\n"
                                          "    level error\n"
                                          "    facility local3\n"
                                          "  </syslog>\n"
                                          "</eventlog>")
        syslog = logger.handlers[0]
        self.assertEqual(syslog.level, logging.ERROR)
        self.assert_(isinstance(syslog, loghandlers.SysLogHandler))

    def test_with_http_logger_localhost(self):
        logger = self.check_simple_logger("<eventlog>\n"
                                          "  <http-logger>\n"
                                          "    level error\n"
                                          "    method post\n"
                                          "  </http-logger>\n"
                                          "</eventlog>")
        handler = logger.handlers[0]
        self.assertEqual(handler.host, "localhost")
        # XXX The "url" attribute of the handler is misnamed; it
        # really means just the selector portion of the URL.
        self.assertEqual(handler.url, "/")
        self.assertEqual(handler.level, logging.ERROR)
        self.assertEqual(handler.method, "POST")
        self.assert_(isinstance(handler, loghandlers.HTTPHandler))

    def test_with_http_logger_remote_host(self):
        logger = self.check_simple_logger("<eventlog>\n"
                                          "  <http-logger>\n"
                                          "    method get\n"
                                          "    url http://example.com/log/\n"
                                          "  </http-logger>\n"
                                          "</eventlog>")
        handler = logger.handlers[0]
        self.assertEqual(handler.host, "example.com")
        # XXX The "url" attribute of the handler is misnamed; it
        # really means just the selector portion of the URL.
        self.assertEqual(handler.url, "/log/")
        self.assertEqual(handler.level, logging.NOTSET)
        self.assertEqual(handler.method, "GET")
        self.assert_(isinstance(handler, loghandlers.HTTPHandler))

    def test_with_email_notifier(self):
        logger = self.check_simple_logger("<eventlog>\n"
                                          "  <email-notifier>\n"
                                          "    to sysadmin@example.com\n"
                                          "    to sa-pager@example.com\n"
                                          "    from zlog-user@example.com\n"
                                          "    level fatal\n"
                                          "  </email-notifier>\n"
                                          "</eventlog>")
        handler = logger.handlers[0]
        self.assertEqual(handler.toaddrs, ["sysadmin@example.com",
                                           "sa-pager@example.com"])
        self.assertEqual(handler.fromaddr, "zlog-user@example.com")
        self.assertEqual(handler.level, logging.FATAL)

    def check_simple_logger(self, text, level=logging.INFO):
        conf = self.get_config(text)
        self.assert_(conf.eventlog is not None)
        self.assertEqual(conf.eventlog.level, level)
        logger = conf.eventlog()
        self.assert_(isinstance(logger, logging.Logger))
        self.assertEqual(len(logger.handlers), 1)
        return logger


def test_suite():
    return unittest.makeSuite(TestLoggingConfig)

if __name__ == '__main__':
    unittest.main(defaultTest="test_suite")


=== Added File Zope3/src/zope/app/process/tests/test_registerrequestfactory.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.
#
##############################################################################
"""

$Id: test_registerrequestfactory.py,v 1.1.2.1 2003/06/23 22:07:59 fdrake Exp $
"""

import unittest
from zope.configuration.xmlconfig import xmlconfig
from zope.configuration.tests.basetestdirectivesxml import makeconfig
from zope.app.process.requestfactoryregistry import getRequestFactory
from zope.testing.cleanup import CleanUp
from zope.app.interfaces.startup import IPublicationRequestFactoryFactory
from zope.interface import implements

class TF:
    "test request factory"
    implements(IPublicationRequestFactoryFactory)

tf = TF()

class Test(CleanUp, unittest.TestCase):

    def testRegisterRequestFactory(self):

        xmlconfig(makeconfig(
            '''<directive
                   name="registerRequestFactory"
                   attributes="name publication request"
                   handler=
                   "zope.app.process.metaconfigure.registerRequestFactory"
                   />''',
            '''<test:registerRequestFactory
                   name="BrowserRequestFactory"
                   publication=
             "zope.app.publication.browser.BrowserPublication"
                   request = "zope.publisher.browser.BrowserRequest" />
            '''
            ))

        from zope.app.publication.browser import \
             BrowserPublication
        from zope.publisher.browser import BrowserRequest

        self.assertEqual(
            getRequestFactory('BrowserRequestFactory')._pubFactory,
            BrowserPublication)
        self.assertEqual(
            getRequestFactory('BrowserRequestFactory')._request,
            BrowserRequest)


    def testRegisterRequestFactory_w_factory(self):

        xmlconfig(makeconfig(
            '''<directive
                   name="registerRequestFactory"
                   attributes="name publication request"
                   handler=
                   "zope.app.process.metaconfigure.registerRequestFactory"
                   />''',
            '''<test:registerRequestFactory
                   name="BrowserRequestFactory"
                   factory="
                   zope.app.process.tests.test_registerrequestfactory.tf"
                   />
            '''
            ))

        import zope.app.process.tests.test_registerrequestfactory

        self.assertEqual(
            getRequestFactory('BrowserRequestFactory'),
            zope.app.process.tests.test_registerrequestfactory.tf
            )




def test_suite():
    loader = unittest.TestLoader()
    return loader.loadTestsFromTestCase(Test)


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


=== Added File Zope3/src/zope/app/process/tests/test_registerservertype.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.
#
##############################################################################
"""terServerType.py,v 1.1.2.2 2002/04/02 02:20:40 srichter Exp $
"""

import unittest
from zope.configuration.xmlconfig import xmlconfig
from zope.configuration.tests.basetestdirectivesxml import makeconfig
from zope.app.process.servertyperegistry import getServerType


class Test(unittest.TestCase):

    def testRegisterServerType(self):
        xmlconfig(makeconfig(
            '''<directive
                   name="registerServerType"
                   attributes="name publication request"
                   handler="zope.app.process.metaconfigure.registerServerType"
                   />''',
            '''<test:registerServerType
                 name = "Browser"
                 factory =
                 "zope.server.http.publisherhttpserver.PublisherHTTPServer"
                 requestFactory="BrowserRequestFactory"
                 logFactory =
                 "zope.server.http.commonhitlogger.CommonHitLogger"
                 defaultPort="8080"
                 defaultVerbose="true" />'''
            ))

        from zope.server.http.publisherhttpserver import PublisherHTTPServer
        from zope.server.http.commonhitlogger import CommonHitLogger

        self.assertEqual(getServerType('Browser')._factory,
                         PublisherHTTPServer)
        self.assertEqual(getServerType('Browser')._logFactory, CommonHitLogger)
        self.assertEqual(getServerType('Browser')._requestFactory,
                         "BrowserRequestFactory")
        self.assertEqual(getServerType('Browser')._defaultPort, 8080)
        self.assertEqual(getServerType('Browser')._defaultVerbose, 1)



def test_suite():
    loader = unittest.TestLoader()
    return loader.loadTestsFromTestCase(Test)


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


=== Added File Zope3/src/zope/app/process/tests/test_requestfactoryregistry.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.
#
##############################################################################
"""
I do not think it is necessary to do the entire SimpleRegistry tests again.
Instead we will test whether the module in itself works.

$Id: test_requestfactoryregistry.py,v 1.1.2.1 2003/06/23 22:07:59 fdrake Exp $
"""

import unittest
from zope.app.process.requestfactoryregistry import \
     registerRequestFactory, getRequestFactory
from zope.app.process.requestfactory import IRequestFactory
from zope.interface import implements


class RequestFactory:
    """RequestFactory Stub."""

    implements(IRequestFactory)


class Test(unittest.TestCase):


    def testRegistry(self):

        factory = RequestFactory()

        registerRequestFactory('factory', factory)
        self.assertEqual(getRequestFactory('factory'), factory)


def test_suite():
    loader = unittest.TestLoader()
    return loader.loadTestsFromTestCase(Test)


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


=== Added File Zope3/src/zope/app/process/tests/test_servertyperegistry.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.
#
##############################################################################
"""
I do not think it is necessary to do the entire SimpleRegistry tests again.
Instead we will test whether the module in itself works.

$Id: test_servertyperegistry.py,v 1.1.2.1 2003/06/23 22:07:59 fdrake Exp $
"""

import unittest
from zope.app.process.servertyperegistry import \
     registerServerType, getServerType
from zope.app.process.servertype import IServerType
from zope.interface import implements


class ServerType:
    """ServerType Stub."""

    implements(IServerType)


class Test(unittest.TestCase):


    def testRegistry(self):

        server = ServerType()

        registerServerType('server', server)
        self.assertEqual(getServerType('server'), server)


def test_suite():
    loader = unittest.TestLoader()
    return loader.loadTestsFromTestCase(Test)


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


=== Added File Zope3/src/zope/app/process/tests/test_simpleregistry.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.
#
##############################################################################
"""

$Id: test_simpleregistry.py,v 1.1.2.1 2003/06/23 22:07:59 fdrake Exp $
"""

import unittest
from zope.interface import Interface
from zope.app.process.simpleregistry import SimpleRegistry, \
     ZopeDuplicateRegistryEntryError, ZopeIllegalInterfaceError
from zope.interface import implements


class I1(Interface):
    pass


class I2(Interface):
    pass


class Object1:
    implements(I1)


class Object2:
    implements(I2)


class Test(unittest.TestCase):


    def testRegister(self):

        registry = SimpleRegistry(I1)
        obj1 = Object1()

        self.assertEqual(registry.objects, {})

        registry.register('obj1', obj1)
        self.assertEqual(registry.objects, {'obj1': obj1})

        registry.register('obj2', obj1)
        self.assertEqual(registry.objects, {'obj1': obj1, 'obj2': obj1})


    def testIllegalInterfaceError(self):

        registry = SimpleRegistry(I1)
        obj2 = Object2()

        self.failUnlessRaises(ZopeIllegalInterfaceError,
                              registry.register, 'obj2', obj2)


    def testDuplicateEntry(self):

        registry = SimpleRegistry(I1)
        obj1 = Object1()
        registry.register('obj1', obj1)

        self.failUnlessRaises(ZopeDuplicateRegistryEntryError,
                              registry.register, 'obj1', obj1)


    def testGet(self):

        registry = SimpleRegistry(I1)
        obj1 = Object1()
        obj2 = Object1()
        registry.objects = {'obj1': obj1, 'obj2': obj2}

        self.assertEqual(registry.get('obj1'), obj1)
        self.assertEqual(registry.get('obj2'), obj2)

        # Requesting an object that does not exist
        self.assertEqual(registry.get('obj3'), None)



def test_suite():
    loader = unittest.TestLoader()
    return loader.loadTestsFromTestCase(Test)


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