Don't throw misleading warnings about duplicate products on product path...

Don't throw misleading warnings about duplicate products on product path unless there actually are duplicate products on product path.� Also, add unit tests for product initialization.

lib/python/OFS/Application.py

-##############################################################################
+############################################################################
 #
 # Copyright (c) 2001 Zope Corporation and Contributors. All Rights Reserved.
 #
@@ -12,8 +12,8 @@
 ##############################################################################
 __doc__='''Application support
 
-$Id: Application.py,v 1.199 2003/12/20 18:56:05 chrism Exp $'''
-__version__='$Revision: 1.199 $'[11:-2]
+$Id: Application.py,v 1.200 2004/01/11 15:32:44 chrism Exp $'''
+__version__='$Revision: 1.200 $'[11:-2]
 
 import Globals,Folder,os,sys,App.Product, App.ProductRegistry, misc_
 import time, traceback, os,  Products
@@ -585,7 +585,6 @@ def install_products(app):
     Products.meta_types=Products.meta_types+tuple(meta_types)
     Globals.default__class_init__(Folder.Folder)
 
-
 def get_products():
     """ Return a list of tuples in the form:
     [(priority, dir_name, index, base_dir), ...] for each Product directory
@@ -595,11 +594,19 @@ def get_products():
     for product_dir in Products.__path__:
         product_names=os.listdir(product_dir)
         for name in product_names:
-            priority = (name != 'PluginIndexes') # import PluginIndexes 1st
-            # i is used as sort ordering in case a conflict exists
-            # between Product names.  Products will be found as
-            # per the ordering of Products.__path__
-            products.append((priority, name, i, product_dir))
+            fullpath = os.path.join(product_dir, name)
+            # Products must be directories
+            if os.path.isdir(fullpath):
+                # Products must be directories with an __init__.py[co]
+                if ( os.path.exists(os.path.join(fullpath, '__init__.py')) or
+                     os.path.exists(os.path.join(fullpath, '__init__.pyo')) or
+                     os.path.exists(os.path.join(fullpath, '__init__.pyc')) ):
+                    # import PluginIndexes 1st (why?)
+                    priority = (name != 'PluginIndexes') 
+                    # i is used as sort ordering in case a conflict exists
+                    # between Product names.  Products will be found as
+                    # per the ordering of Products.__path__
+                    products.append((priority, name, i, product_dir))
         i = i + 1
     products.sort()
     return products
@@ -620,6 +627,7 @@ def import_products():
             continue
         done[product_name]=product_dir
         import_product(product_dir, product_name, raise_exc=debug_mode)
+    return done.keys()
 
 def import_product(product_dir, product_name, raise_exc=0, log_exc=1):
     path_join=os.path.join

lib/python/OFS/tests/testProductInit.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.
+#
+##############################################################################
+
+import os, sys, unittest, tempfile, shutil, cStringIO
+
+import ZODB
+from OFS.Application import Application, AppInitializer, get_products
+import Zope.Startup
+import ZConfig
+from App.config import getConfiguration, setConfiguration
+
+TEMPNAME = tempfile.mktemp()
+TEMPPRODUCTS = os.path.join(TEMPNAME, "Products")
+TEMPPRODUCTS2 = os.path.join(TEMPNAME, "Products2")
+FAKEPRODUCTS = ['foo', 'bar', 'bee', 'baz']
+
+cfg = """
+instancehome <<INSTANCE_HOME>>
+products <<PRODUCTS>>
+products <<PRODUCTS2>>
+
+<zodb_db main>
+   mount-point /
+   <mappingstorage>
+      name mappingstorage
+   </mappingstorage>
+</zodb_db>
+"""
+
+dummy_product_init = """
+def initialize(context):
+    f=open('%s', 'w')
+    f.write('didit')
+    f.close()
+misc_ = {'a':1}
+def amethod(self):
+    pass
+methods = {'amethod':amethod}
+__ac_permissions__ = ( ('aPermission', (), () ), )
+meta_types = ( {'name':'grabass', 'action':'amethod'}, )
+"""
+
+def getSchema():
+    startup = os.path.dirname(os.path.realpath(Zope.Startup.__file__))
+    schemafile = os.path.join(startup, 'zopeschema.xml')
+    return ZConfig.loadSchema(schemafile)
+
+def getApp():
+    from ZODB.ZApplication import ZApplicationWrapper
+    DB = getConfiguration().dbtab.getDatabase('/')
+    return ZApplicationWrapper(DB, 'Application', Application, (), 'foo')()
+
+original_config = None
+
+class TestProductInit( unittest.TestCase ):
+    """ Test the application initializer object """
+
+    def setUp(self):
+        global original_config
+        if original_config is None:
+            original_config = getConfiguration()
+        self.schema = getSchema()
+        os.makedirs(TEMPNAME)
+        os.makedirs(TEMPPRODUCTS)
+        os.makedirs(TEMPPRODUCTS2)
+
+    def tearDown(self):
+        import App.config
+        del self.schema
+        App.config.setConfiguration(original_config)
+        shutil.rmtree(TEMPNAME)
+
+    def configure(self, text):
+        # We have to create a directory of our own since the existence
+        # of the directory is checked.  This handles this in a
+        # platform-independent way.
+        schema = self.schema
+        text = text.replace("<<INSTANCE_HOME>>", TEMPNAME)
+        text = text.replace("<<PRODUCTS>>", TEMPPRODUCTS)
+        text = text.replace("<<PRODUCTS2>>", TEMPPRODUCTS2)
+        sio = cStringIO.StringIO(text)
+        conf, handler = ZConfig.loadConfigFile(schema, sio)
+        from Zope.Startup.handlers import handleConfig
+        handleConfig(conf, handler)
+        self.assertEqual(conf.instancehome, TEMPNAME)
+        setConfiguration(conf)
+
+    def makeProduct(self, proddir):
+        os.makedirs(proddir)
+        f = open(os.path.join(proddir, '__init__.py'), 'w')
+        f.write('#foo')
+        f.close()
+
+    def makeFakeProducts(self):
+        for name in FAKEPRODUCTS:
+            proddir = os.path.join(TEMPPRODUCTS, name)
+            self.makeProduct(proddir)
+
+    def test_get_products(self):
+        self.makeFakeProducts()
+        self.configure(cfg)
+        from OFS.Application import get_products
+        names =  [x[1] for x in get_products()]
+        for name in FAKEPRODUCTS:
+            self.assert_(name in names)
+
+    def test_empty_dir_on_products_path_is_not_product(self):
+        self.makeFakeProducts()
+        os.makedirs(os.path.join(TEMPPRODUCTS, 'gleeb'))
+        self.configure(cfg)
+        from OFS.Application import get_products
+        names =  [x[1] for x in get_products()]
+        for name in FAKEPRODUCTS:
+            self.assert_(name in names)
+        self.assert_('gleeb' not in names)
+        
+    def test_file_on_products_path_is_not_product(self):
+        self.makeFakeProducts()
+        f = open(os.path.join(TEMPPRODUCTS, 'README.txt'), 'w')
+        f.write('#foo')
+        f.close()
+        self.configure(cfg)
+        from OFS.Application import get_products
+        names =  [x[1] for x in get_products()]
+        for name in FAKEPRODUCTS:
+            self.assert_(name in names)
+        self.assert_('README.txt' not in names)
+
+    def test_multiple_product_paths(self):
+        self.makeFakeProducts()
+        self.makeProduct(os.path.join(TEMPPRODUCTS2, 'another'))
+        self.configure(cfg)
+        from OFS.Application import get_products
+        names =  [x[1] for x in get_products()]
+        for name in FAKEPRODUCTS:
+            self.assert_(name in names)
+        self.assert_('another' in names)
+
+    def test_import_products(self):
+        self.makeFakeProducts()
+        self.configure(cfg)
+        from OFS.Application import import_products
+        names = import_products()
+        for name in FAKEPRODUCTS:
+            assert name in names
+
+    def test_import_product_throws(self):
+        self.makeProduct(os.path.join(TEMPPRODUCTS, 'abar'))
+        f = open(os.path.join(TEMPPRODUCTS, 'abar', '__init__.py'), 'w')
+        f.write('Syntax Error!')
+        f.close()
+        self.configure(cfg)
+        self.assertRaises(SyntaxError, self.import_bad_product)
+
+    def import_bad_product(self):
+        from OFS.Application import import_product
+        import_product(TEMPPRODUCTS, 'abar', raise_exc=1)
+
+    def test_install_product(self):
+        self.makeProduct(os.path.join(TEMPPRODUCTS, 'abaz'))
+        f = open(os.path.join(TEMPPRODUCTS, 'abaz', '__init__.py'), 'w')
+        doneflag = os.path.join(TEMPPRODUCTS, 'abaz', 'doneflag')
+        f.write(dummy_product_init % doneflag)
+        f.close()
+        self.configure(cfg)
+        from OFS.Application import install_product, get_folder_permissions,\
+             Application
+        import Products
+        from OFS.Folder import Folder
+        app = getApp()
+        meta_types = []
+        install_product(app, TEMPPRODUCTS, 'abaz', meta_types,
+                        get_folder_permissions(), raise_exc=1)
+        # misc_ dictionary is updated
+        self.assert_(Application.misc_.__dict__.has_key('abaz'))
+        # initialize is called
+        self.assert_(os.path.exists(doneflag))
+        # Methods installed into folder
+        self.assert_(hasattr(Folder, 'amethod'))
+        # __ac_permissions__ put into folder
+        self.assert_( ('aPermission', (),)  in
+                       Folder.__ac_permissions__)
+        # Products.meta_types updated
+        self.assert_( {'action': 'amethod', 'product': 'abaz',
+                       'name': 'grabass', 'visibility': 'Global'}
+                      in meta_types)
+
+    def test_install_products(self):
+        self.makeFakeProducts()
+        self.configure(cfg)
+        app = getApp()
+        from OFS.Application import install_products
+        install_products(app)
+        obids = app.Control_Panel.Products.objectIds()
+        for name in FAKEPRODUCTS:
+            assert name in obids
+
+def test_suite():
+    suite = unittest.TestSuite()
+    suite.addTest( unittest.makeSuite( TestProductInit ) )
+    return suite
+
+def main():
+    unittest.main(defaultTest='test_suite')
+
+if __name__ == '__main__':
+    main()