Commit 10c46508 authored by Stefan H. Holek's avatar Stefan H. Holek

Put ZopeLite on a layer.

parent 26880a5b
......@@ -26,6 +26,7 @@ $Id$
"""
import os, sys, time
import layer
# Allow code to tell it is run by the test framework
os.environ['ZOPETESTCASE'] = '1'
......@@ -105,7 +106,12 @@ _write('.')
_patched = False
@layer.onsetup
def _apply_patches():
# Do not patch a running Zope
if Zope2._began_startup:
return
# Avoid expensive product import
def null_import_products(): pass
OFS.Application.import_products = null_import_products
......@@ -126,9 +132,17 @@ def _apply_patches():
global _patched
_patched = True
# Do not patch a running Zope
if not Zope2._began_startup:
_apply_patches()
_apply_patches()
_theApp = None
@layer.onsetup
def _startup():
global _theApp
_theApp = Zope2.app()
# Start ZopeLite
_startup()
# Allow test authors to install Zope products into the test environment. Note
# that installProduct() must be called at module level -- never from tests.
......@@ -137,7 +151,6 @@ from OFS.Application import install_product, install_package
from OFS.Folder import Folder
import Products
_theApp = Zope2.app()
_installedProducts = {}
_installedPackages = {}
......@@ -145,7 +158,13 @@ def hasProduct(name):
'''Checks if a product can be found along Products.__path__'''
return name in [n[1] for n in get_products()]
@layer.onsetup
def installProduct(name, quiet=0):
'''Installs a Zope product at layer setup time.'''
quiet = 1 # Ignore argument
_installProduct(name, quiet)
def _installProduct(name, quiet=0):
'''Installs a Zope product.'''
start = time.time()
meta_types = []
......@@ -170,8 +189,14 @@ def hasPackage(name):
'''Checks if a package has been registered with five:registerPackage.'''
return name in [m.__name__ for m in getattr(Products, '_registered_packages', [])]
@layer.onsetup
def installPackage(name, quiet=0):
'''Installs a registered Python package like a Zope product.'''
'''Installs a registered Python package at layer setup time.'''
quiet = 1 # Ignore argument
_installPackage(name, quiet)
def _installPackage(name, quiet=0):
'''Installs a registered Python package.'''
start = time.time()
if _patched and not _installedPackages.has_key(name):
for module, init_func in getattr(Products, '_packages_to_initialize', []):
......@@ -187,27 +212,8 @@ def installPackage(name, quiet=0):
else:
if not quiet: _print('Installing %s ... NOT FOUND\n' % name)
def _load_control_panel():
# Loading the Control_Panel of an existing ZODB may take
# a while; print another dot if it does.
start = time.time()
max = (start - _start) / 4
_exec('_theApp.Control_Panel')
_theApp.Control_Panel
if (time.time() - start) > max:
_write('.')
def _install_products():
installProduct('PluginIndexes', 1) # Must install first
installProduct('OFSP', 1)
#installProduct('ExternalMethod', 1)
#installProduct('ZSQLMethods', 1)
#installProduct('ZGadflyDA', 1)
#installProduct('MIMETools', 1)
#installProduct('MailHost', 1)
_load_control_panel()
_install_products()
installProduct('PluginIndexes', 1) # Must install first
installProduct('OFSP', 1)
# So people can use ZopeLite.app()
app = Zope2.app
......
......@@ -17,6 +17,7 @@ $Id$
import ZopeLite as Zope2
import utils
import layer
from ZopeLite import hasProduct
from ZopeLite import installProduct
......
......@@ -21,12 +21,12 @@ import transaction
import utils
import interfaces
import connections
import layer
from zope.interface import implements
from AccessControl.SecurityManagement import noSecurityManager
def app():
'''Opens a ZODB connection and returns the app object.'''
app = Zope2.app()
......@@ -34,18 +34,20 @@ def app():
connections.register(app)
return app
def close(app):
'''Closes the app's ZODB connection.'''
connections.close(app)
class TestCase(unittest.TestCase, object):
'''Base test case for Zope testing
'''
implements(interfaces.IZopeTestCase)
layer = layer.ZopeLite
def afterSetUp(self):
'''Called after setUp() has completed. This is
far and away the most useful hook.
......
##############################################################################
#
# Copyright (c) 2007 Zope Corporation and Contributors. All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (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.
#
##############################################################################
"""ZopeLite layer
$Id$
"""
_deferred_setup = []
class ZopeLite:
'''The most base layer'''
@classmethod
def setUp(cls):
'''Brings up the ZopeLite environment.'''
for func, args, kw in _deferred_setup:
func(*args, **kw)
@classmethod
def tearDown(cls):
'''ZopeLite doesn't support tear down.
We don't raise NotImplementedError to avoid
triggering the testrunner's "resume layer"
mechanism.
See zope.testing.testrunner-layers-ntd.txt
'''
ZopeLiteLayer = ZopeLite
def onsetup(func):
'''Defers a function call to layer setup.
Used as a decorator.
'''
def deferred_func(*args, **kw):
_deferred_setup.append((func, args, kw))
return deferred_func
def appcall(func):
'''Defers a function call to layer setup.
Used as a decorator.
In addition, this decorator implements the appcall
protocol:
* The decorated function expects 'app' as first argument.
* If 'app' is provided by the caller, the function is
called immediately.
* If 'app' is omitted or None, the 'app' argument is
provided by the decorator, and the function call is
deferred to ZopeLite layer setup.
Also see utils.appcall.
'''
def appcalled_func(*args, **kw):
if args and args[0] is not None:
return func(*args, **kw)
if kw.get('app') is not None:
return func(*args, **kw)
def caller(*args, **kw):
import utils
utils.appcall(func, *args, **kw)
_deferred_setup.append((caller, args, kw))
return appcalled_func
......@@ -30,23 +30,31 @@ if __name__ == '__main__':
from Testing import ZopeTestCase
from Testing.ZopeTestCase import layer
from Testing.ZopeTestCase import utils
from Testing.ZopeTestCase import transaction
from Globals import SOFTWARE_HOME
examples_path = os.path.join(SOFTWARE_HOME, '..', '..', 'skel', 'import', 'Examples.zexp')
examples_path = os.path.abspath(examples_path)
# Open ZODB connection
app = ZopeTestCase.app()
class ShoppingCartLayer(layer.ZopeLite):
# Set up sessioning objects
ZopeTestCase.utils.setupCoreSessions(app)
@classmethod
def setUp(cls):
# Set up sessioning objects
utils.appcall(utils.setupCoreSessions)
# Set up example applications
if not hasattr(app, 'Examples'):
ZopeTestCase.utils.importObjectFromFile(app, examples_path)
# Set up example applications
utils.appcall(utils.importObjectFromFile, examples_path, quiet=1)
# Close ZODB connection
ZopeTestCase.close(app)
@classmethod
def tearDown(cls):
def cleanup(app):
app._delObject('Examples')
transaction.commit()
utils.appcall(cleanup)
class DummyOrder:
......@@ -63,6 +71,8 @@ class TestShoppingCart(ZopeTestCase.ZopeTestCase):
_setup_fixture = 0 # No default fixture
layer = ShoppingCartLayer
def afterSetUp(self):
self.cart = self.app.Examples.ShoppingCart
# Put SESSION object into REQUEST
......
......@@ -46,8 +46,7 @@ import urllib
ZopeTestCase.utils.setupSiteErrorLog()
# Start the web server
host, port = ZopeTestCase.utils.startZServer(4)
folder_url = 'http://%s:%d/%s' %(host, port, ZopeTestCase.folder_name)
ZopeTestCase.utils.startZServer()
class ManagementOpener(urllib.FancyURLopener):
......@@ -55,6 +54,7 @@ class ManagementOpener(urllib.FancyURLopener):
def prompt_user_passwd(self, host, realm):
return ('manager', 'secret')
class UnauthorizedOpener(urllib.FancyURLopener):
'''Raises Unauthorized when prompted'''
def prompt_user_passwd(self, host, realm):
......@@ -67,6 +67,8 @@ class TestWebserver(ZopeTestCase.ZopeTestCase):
uf = self.folder.acl_users
uf.userFolderAddUser('manager', 'secret', ['Manager'], [])
self.folder_url = self.folder.absolute_url()
# A simple document
self.folder.addDTMLDocument('index_html', file='index_html called')
......@@ -99,7 +101,7 @@ class TestWebserver(ZopeTestCase.ZopeTestCase):
def testURLAccessPublicObject(self):
# Test web access to a public resource
urllib._urlopener = ManagementOpener()
page = urllib.urlopen(folder_url+'/index_html').read()
page = urllib.urlopen(self.folder_url+'/index_html').read()
self.assertEqual(page, 'index_html called')
def testAccessProtectedObject(self):
......@@ -110,7 +112,7 @@ class TestWebserver(ZopeTestCase.ZopeTestCase):
def testURLAccessProtectedObject(self):
# Test web access to a protected resource
urllib._urlopener = ManagementOpener()
page = urllib.urlopen(folder_url+'/secret_html').read()
page = urllib.urlopen(self.folder_url+'/secret_html').read()
self.assertEqual(page, 'secret_html called')
def testSecurityOfPublicObject(self):
......@@ -125,7 +127,7 @@ class TestWebserver(ZopeTestCase.ZopeTestCase):
# Test web security of a public resource
urllib._urlopener = UnauthorizedOpener()
try:
urllib.urlopen(folder_url+'/index_html')
urllib.urlopen(self.folder_url+'/index_html')
except Unauthorized:
# Convert error to failure
self.fail('Unauthorized')
......@@ -143,7 +145,7 @@ class TestWebserver(ZopeTestCase.ZopeTestCase):
# Test web security of a protected resource
urllib._urlopener = UnauthorizedOpener()
try:
urllib.urlopen(folder_url+'/secret_html')
urllib.urlopen(self.folder_url+'/secret_html')
except Unauthorized:
pass # Test passed
else:
......@@ -161,13 +163,9 @@ class TestWebserver(ZopeTestCase.ZopeTestCase):
def testURLModifyObject(self):
# Test a transaction that actually commits something
urllib._urlopener = ManagementOpener()
page = urllib.urlopen(folder_url+'/index_html/change_title?title=Foo').read()
page = urllib.urlopen(self.folder_url+'/index_html/change_title?title=Foo').read()
self.assertEqual(page, 'Foo')
def testAbsoluteURL(self):
# Test absolute_url
self.assertEqual(self.folder.absolute_url(), folder_url)
class TestSandboxedWebserver(ZopeTestCase.Sandboxed, TestWebserver):
'''Demonstrates that tests involving ZServer threads can also be
......@@ -182,7 +180,7 @@ class TestSandboxedWebserver(ZopeTestCase.Sandboxed, TestWebserver):
# same connection as the main thread, allowing us to
# see changes made to 'index_html' right away.
urllib._urlopener = ManagementOpener()
urllib.urlopen(folder_url+'/index_html/change_title?title=Foo')
urllib.urlopen(self.folder_url+'/index_html/change_title?title=Foo')
self.assertEqual(self.folder.index_html.title, 'Foo')
def testCanCommit(self):
......
......@@ -25,7 +25,9 @@ if __name__ == '__main__':
from Testing import ZopeTestCase
import transaction
from Testing.ZopeTestCase import layer
from Testing.ZopeTestCase import utils
from Testing.ZopeTestCase import transaction
from AccessControl.Permissions import add_documents_images_and_files
from AccessControl.Permissions import delete_objects
......@@ -34,6 +36,35 @@ import tempfile
folder_name = ZopeTestCase.folder_name
cutpaste_permissions = [add_documents_images_and_files, delete_objects]
# Dummy object
from OFS.SimpleItem import SimpleItem
class DummyObject(SimpleItem):
id = 'dummy'
foo = None
_v_foo = None
_p_foo = None
class ZODBCompatLayer(layer.ZopeLite):
@classmethod
def setUp(cls):
def setup(app):
app._setObject('dummy1', DummyObject())
app._setObject('dummy2', DummyObject())
transaction.commit()
utils.appcall(setup)
@classmethod
def tearDown(cls):
def cleanup(app):
app._delObject('dummy1')
app._delObject('dummy2')
transaction.commit()
utils.appcall(cleanup)
class TestCopyPaste(ZopeTestCase.ZopeTestCase):
......@@ -159,22 +190,6 @@ class TestImportExport(ZopeTestCase.ZopeTestCase):
App.config.setConfiguration(config)
# Dummy object
from OFS.SimpleItem import SimpleItem
class DummyObject(SimpleItem):
id = 'dummy'
foo = None
_v_foo = None
_p_foo = None
app = ZopeTestCase.app()
app._setObject('dummy1', DummyObject())
app._setObject('dummy2', DummyObject())
transaction.commit()
ZopeTestCase.close(app)
class TestAttributesOfCleanObjects(ZopeTestCase.ZopeTestCase):
'''This testcase shows that _v_ and _p_ attributes are NOT bothered
by transaction boundaries, if the respective object is otherwise
......@@ -194,7 +209,9 @@ class TestAttributesOfCleanObjects(ZopeTestCase.ZopeTestCase):
This testcase exploits the fact that test methods are sorted by name.
'''
layer = ZODBCompatLayer
def afterSetUp(self):
self.dummy = self.app.dummy1 # See above
......@@ -256,6 +273,8 @@ class TestAttributesOfDirtyObjects(ZopeTestCase.ZopeTestCase):
This testcase exploits the fact that test methods are sorted by name.
'''
layer = ZODBCompatLayer
def afterSetUp(self):
self.dummy = self.app.dummy2 # See above
self.dummy.touchme = 1 # Tag, you're dirty
......
......@@ -23,16 +23,15 @@ import sys
import time
import random
import transaction
import layer
def setupCoreSessions(app=None):
@layer.appcall
def setupCoreSessions(app):
'''Sets up the session_data_manager e.a.'''
from Acquisition import aq_base
commit = 0
if app is None:
return appcall(setupCoreSessions)
if not hasattr(app, 'temp_folder'):
from Products.TemporaryFolder.TemporaryFolder import MountedTemporaryFolder
tf = MountedTemporaryFolder('temp_folder', 'Temporary Folder')
......@@ -68,11 +67,9 @@ def setupCoreSessions(app=None):
transaction.commit()
def setupZGlobals(app=None):
@layer.appcall
def setupZGlobals(app):
'''Sets up the ZGlobals BTree required by ZClasses.'''
if app is None:
return appcall(setupZGlobals)
root = app._p_jar.root()
if not root.has_key('ZGlobals'):
from BTrees.OOBTree import OOBTree
......@@ -80,11 +77,9 @@ def setupZGlobals(app=None):
transaction.commit()
def setupSiteErrorLog(app=None):
@layer.appcall
def setupSiteErrorLog(app):
'''Sets up the error_log object required by ZPublisher.'''
if app is None:
return appcall(setupSiteErrorLog)
if not hasattr(app, 'error_log'):
try:
from Products.SiteErrorLog.SiteErrorLog import SiteErrorLog
......@@ -135,13 +130,13 @@ def makerequest(app, stdout=sys.stdout):
return _makerequest(app, stdout=stdout, environ=environ)
def appcall(function, *args, **kw):
def appcall(func, *args, **kw):
'''Calls a function passing 'app' as first argument.'''
from base import app, close
app = app()
args = (app,) + args
try:
return function(*args, **kw)
return func(*args, **kw)
finally:
transaction.abort()
close(app)
......
......@@ -20,9 +20,10 @@ from Testing import ZopeTestCase
from Testing.ZopeTestCase import ZopeDocFileSuite
from Testing.ZopeTestCase import ZopeDocTestSuite
from Testing.ZopeTestCase import transaction
from Testing.ZopeTestCase import layer
class TestLayer:
class TestLayer(layer.ZopeLite):
"""
If the layer is extracted properly, we should see the following
variable
......
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