3956162881

Build Type

push

github

Committed by Michael Howitz

Commit Message

Update to deprecation warning free releases.

Run Details

4401 of 7036 branches covered (62.55%)

Branch coverage included in aggregate %.

27161 of 31488 relevant lines covered (86.26%)

0.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.79

/src/OFS/Application.py

##############################################################################
#
# Copyright (c) 2002 Zope Foundation and Contributors.
#
# 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.
#
##############################################################################
"""Application support
"""

import os
import sys
from logging import getLogger

import Products
import transaction
from AccessControl import ClassSecurityInfo
from AccessControl.class_init import InitializeClass
from AccessControl.Permission import ApplicationDefaultPermissions
from AccessControl.Permissions import view_management_screens
from Acquisition import aq_base
from App import FactoryDispatcher
from App.ApplicationManager import ApplicationManager
from App.ProductContext import ProductContext
from App.version_txt import getZopeVersion
from DateTime import DateTime
from OFS.FindSupport import FindSupport
from OFS.metaconfigure import get_packages_to_initialize
from OFS.metaconfigure import package_initialized
from OFS.userfolder import UserFolder
from webdav.NullResource import NullResource
from zExceptions import Forbidden
from zExceptions import Redirect as RedirectException
from zope.interface import implementer

from . import Folder
from . import misc_
from .interfaces import IApplication
from .misc_ import Misc_


LOG = getLogger('Application')

APP_MANAGER = None


@implementer(IApplication)
class Application(ApplicationDefaultPermissions, Folder.Folder, FindSupport):
    """Top-level system object"""

    security = ClassSecurityInfo()

    title = 'Zope'
    __defined_roles__ = ('Manager', 'Anonymous', 'Owner')
    __error_log__ = None
    isTopLevelPrincipiaApplicationObject = 1

    p_ = misc_.p_
    misc_ = misc_.misc_
    _reserved_names = ('Control_Panel', )

    # This class-default __allow_groups__ ensures that the
    # emergency user can still access the system if the top-level
    # UserFolder is deleted. This is necessary to allow people
    # to replace the top-level UserFolder object.
    __allow_groups__ = UserFolder()

    def __init__(self):
        # Initialize users
        uf = UserFolder()
        self.__allow_groups__ = uf
        self._setObject('acl_users', uf)

    def getId(self):
        try:
            return self.REQUEST['SCRIPT_NAME'][1:]
        except (KeyError, TypeError):
            return self.title

    def title_and_id(self):
        return self.title

    def title_or_id(self):
        return self.title

    def __class_init__(self):
        InitializeClass(self)

    @property
    def Control_Panel(self):
        return APP_MANAGER.__of__(self)

    def Redirect(self, destination, URL1):
        # Utility function to allow user-controlled redirects.
        # No docstring please, we do not want an open redirect
        # available as url.
        if destination.find('//') >= 0:
            raise RedirectException(destination)
        raise RedirectException(f"{URL1}/{destination}")

    ZopeRedirect = Redirect

    def __bobo_traverse__(self, REQUEST, name=None):
        if name is None:
            # Make this more explicit, otherwise getattr(self, name)
            # would raise a TypeError getattr(): attribute name must be string
            return None

        if name == 'Control_Panel':
            return APP_MANAGER.__of__(self)
        try:
            return getattr(self, name)
        except AttributeError:
            pass

        try:
            return self[name]
        except KeyError:
            pass

        method = REQUEST.get('REQUEST_METHOD', 'GET')

        if method not in ('GET', 'POST'):
            return NullResource(self, name, REQUEST).__of__(self)

        # Waaa. unrestrictedTraverse calls us with a fake REQUEST.
        # There is probably a better fix for this.
        try:
            REQUEST.RESPONSE.notFoundError(f"{name}\n{method}")
        except AttributeError:
            raise KeyError(name)

    def ZopeTime(self, *args):
        """Utility function to return current date/time"""
        return DateTime(*args)

    @security.protected(view_management_screens)
    def ZopeVersion(self, major=False):
        """Utility method to return the Zope version

        Restricted to ZMI to prevent information disclosure
        """
        zversion = getZopeVersion()
        if major:
            return zversion.major
        else:
            version = f'{zversion.major}.{zversion.minor}.{zversion.micro}'
            if zversion.status:
                version += f'.{zversion.status}{zversion.release}'

            return version

    def DELETE(self, REQUEST, RESPONSE):
        """Delete a resource object."""
        self.dav__init(REQUEST, RESPONSE)
        raise Forbidden('This resource cannot be deleted.')

    def MOVE(self, REQUEST, RESPONSE):
        """Move a resource to a new location."""
        self.dav__init(REQUEST, RESPONSE)
        raise Forbidden('This resource cannot be moved.')

    def absolute_url(self, relative=0):
        """The absolute URL of the root object is BASE1 or "/".
        """
        if relative:
            return ''
        try:
            # Take advantage of computed URL cache
            return self.REQUEST['BASE1']
        except (AttributeError, KeyError):
            return '/'

    def absolute_url_path(self):
        """The absolute URL path of the root object is BASEPATH1 or "/".
        """
        try:
            return self.REQUEST['BASEPATH1'] or '/'
        except (AttributeError, KeyError):
            return '/'

    def virtual_url_path(self):
        """The virtual URL path of the root object is empty.
        """
        return ''

    def getPhysicalRoot(self):
        return self

    def getPhysicalPath(self):
        # Get the physical path of the object.
        #
        # Returns a path (an immutable sequence of strings) that can be used to
        # access this object again later, for example in a copy/paste
        # operation.  getPhysicalRoot() and getPhysicalPath() are designed to
        # operate together.
        #
        # We're at the base of the path.
        return ('', )


InitializeClass(Application)


def initialize(app):
    initializer = AppInitializer(app)
    initializer.initialize()


class AppInitializer:
    """ Initialize an Application object (called at startup) """

    def __init__(self, app):
        self.app = (app,)

    def getApp(self):
        # this is probably necessary, but avoid acquisition anyway
        return self.app[0]

    def commit(self, note):
        transaction.get().note(note)
        transaction.commit()

    def initialize(self):
        # make sure to preserve relative ordering of calls below.
        self.install_app_manager()
        self.install_required_roles()
        self.install_inituser()
        self.install_products()
        self.install_standards()
        self.install_virtual_hosting()
        self.install_root_view()

    def install_app_manager(self):
        global APP_MANAGER
        APP_MANAGER = ApplicationManager()

        # Remove persistent Control Panel.
        app = self.getApp()
        app._p_activate()

        if 'Control_Panel' in list(app.__dict__.keys()):
            del app.__dict__['Control_Panel']
            app._objects = tuple(i for i in app._objects
                                 if i['id'] != 'Control_Panel')
            self.commit('Removed persistent Control_Panel')

    def install_required_roles(self):
        app = self.getApp()

        # Ensure that Owner role exists.
        if hasattr(app, '__ac_roles__') and not ('Owner' in app.__ac_roles__):
            app.__ac_roles__ = app.__ac_roles__ + ('Owner',)
            self.commit('Added Owner role')

        # ensure the Authenticated role exists.
        if hasattr(app, '__ac_roles__'):
            if 'Authenticated' not in app.__ac_roles__:
                app.__ac_roles__ = app.__ac_roles__ + ('Authenticated',)
                self.commit('Added Authenticated role')

    def install_inituser(self):
        app = self.getApp()
        # Install the initial user.
        if hasattr(app, 'acl_users'):
            users = app.acl_users
            if hasattr(users, '_createInitialUser'):
                app.acl_users._createInitialUser()
                self.commit('Created initial user')
            users = aq_base(users)
            migrated = getattr(users, '_ofs_migrated', False)
            if not migrated:
                klass = users.__class__
                from OFS.userfolder import UserFolder
                if klass is UserFolder:
                    # zope.deferredimport does a thourough job, so the class
                    # looks like it's from the new location already. And we
                    # don't want to migrate any custom user folders here.
                    users.__class__ = UserFolder
                    users._ofs_migrated = True
                    users._p_changed = True
                    app._p_changed = True
                    transaction.get().note('Migrated user folder')
                    transaction.commit()

    def install_virtual_hosting(self):
        app = self.getApp()
        if 'virtual_hosting' not in app:
            from Products.SiteAccess.VirtualHostMonster import \
                VirtualHostMonster
            any_vhm = [obj for obj in app.values()
                       if isinstance(obj, VirtualHostMonster)]
            if not any_vhm:
                vhm = VirtualHostMonster()
                vhm.id = 'virtual_hosting'
                vhm.addToContainer(app)
                self.commit('Added virtual_hosting')

    def install_root_view(self):
        app = self.getApp()
        if 'index_html' not in app:
            from Products.PageTemplates.ZopePageTemplate import \
                ZopePageTemplate
            root_pt = ZopePageTemplate('index_html')
            root_pt.pt_setTitle('Auto-generated default page')
            app._setObject('index_html', root_pt)
            self.commit('Added default view for root object')

    def install_products(self):
        return install_products(self.getApp())

    def install_standards(self):
        app = self.getApp()
        if getattr(app, '_standard_objects_have_been_added', None) is not None:
            delattr(app, '_standard_objects_have_been_added')
        if getattr(app, '_initializer_registry', None) is not None:
            delattr(app, '_initializer_registry')
        transaction.get().note('Removed unused application attributes.')
        transaction.commit()


def install_products(app=None):
    folder_permissions = get_folder_permissions()
    meta_types = []
    done = {}
    for priority, product_name, index, product_dir in get_products():
        # For each product, we will import it and try to call the
        # intialize() method in the product __init__ module. If
        # the method doesnt exist, we put the old-style information
        # together and do a default initialization.
        if product_name in done:
            continue
        done[product_name] = 1
        install_product(app, product_dir, product_name, meta_types,
                        folder_permissions)

    # Delayed install of packages-as-products
    for module, init_func in tuple(get_packages_to_initialize()):
        install_package(app, module, init_func)

    Products.meta_types = Products.meta_types + tuple(meta_types)
    InitializeClass(Folder.Folder)


def _is_package(product_dir, product_name):
    package_dir = os.path.join(product_dir, product_name)
    if not os.path.isdir(package_dir):
        return False

    init_py = os.path.join(package_dir, '__init__.py')
    if not os.path.exists(init_py) and \
       not os.path.exists(init_py + 'c') and \
       not os.path.exists(init_py + 'o'):
        return False
    return True


def get_products():
    """ Return a list of tuples in the form:
    [(priority, dir_name, index, base_dir), ...] for each Product directory
    found, sort before returning """
    products = []
    i = 0
    for product_dir in Products.__path__:
        product_names = os.listdir(product_dir)
        for product_name in product_names:
            if _is_package(product_dir, product_name):
                # 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((0, product_name, i, product_dir))
        i = i + 1
    products.sort()
    return products


def import_products():
    done = {}
    for priority, product_name, index, product_dir in get_products():
        if product_name in done:
            LOG.warning(
                'Duplicate Product name: '
                'After loading Product %r from %r, '
                'I skipped the one in %r.' % (
                    product_name, done[product_name], product_dir))
            continue
        done[product_name] = product_dir
        import_product(product_dir, product_name)
    return list(done.keys())


def import_product(product_dir, product_name, raise_exc=None):
    if not _is_package(product_dir, product_name):
        return

    global_dict = globals()
    product = __import__("Products.%s" % product_name,
                         global_dict, global_dict, ('__doc__', ))
    if hasattr(product, '__module_aliases__'):
        for k, v in product.__module_aliases__:
            if k not in sys.modules:
                if isinstance(v, str) and v in sys.modules:
                    v = sys.modules[v]
                sys.modules[k] = v


def get_folder_permissions():
    folder_permissions = {}
    for p in Folder.Folder.__ac_permissions__:
        permission, names = p[:2]
        folder_permissions[permission] = names
    return folder_permissions


def install_product(app, product_dir, product_name, meta_types,
                    folder_permissions, raise_exc=None):
    if not _is_package(product_dir, product_name):
        return

    __traceback_info__ = product_name
    global_dict = globals()
    product = __import__("Products.%s" % product_name,
                         global_dict, global_dict, ('__doc__', ))

    # Install items into the misc_ namespace, used by products
    # and the framework itself to store common static resources
    # like icon images.
    misc_ = pgetattr(product, 'misc_', {})
    if misc_:
        if isinstance(misc_, dict):
            misc_ = Misc_(product_name, misc_)
        setattr(Application.misc_, product_name, misc_)

    productObject = FactoryDispatcher.Product(product_name)
    context = ProductContext(productObject, app, product)

    # Look for an 'initialize' method in the product.
    initmethod = pgetattr(product, 'initialize', None)
    if initmethod is not None:
        initmethod(context)


def install_package(app, module, init_func, raise_exc=None):
    """Installs a Python package like a product."""
    name = module.__name__
    product = FactoryDispatcher.Product(name)
    product.package_name = name

    if init_func is not None:
        newContext = ProductContext(product, app, module)
        init_func(newContext)

    package_initialized(module, init_func)


def pgetattr(product, name, default=install_products, __init__=0):
    if not __init__ and hasattr(product, name):
        return getattr(product, name)
    if hasattr(product, '__init__'):
        product = product.__init__
        if hasattr(product, name):
            return getattr(product, name)

    if default is not install_products:
        return default

    raise AttributeError(name)

1	##############################################################################
2	#
3	# Copyright (c) 2002 Zope Foundation and Contributors.
4	#
5	# This software is subject to the provisions of the Zope Public License,
6	# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
7	# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
8	# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
9	# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
10	# FOR A PARTICULAR PURPOSE.
11	#
12	##############################################################################
13	"""Application support	1✔
14	"""
15
16	import os	1✔
17	import sys	1✔
18	from logging import getLogger	1✔
19
20	import Products	1✔
21	import transaction	1✔
22	from AccessControl import ClassSecurityInfo	1✔
23	from AccessControl.class_init import InitializeClass	1✔
24	from AccessControl.Permission import ApplicationDefaultPermissions	1✔
25	from AccessControl.Permissions import view_management_screens	1✔
26	from Acquisition import aq_base	1✔
27	from App import FactoryDispatcher	1✔
28	from App.ApplicationManager import ApplicationManager	1✔
29	from App.ProductContext import ProductContext	1✔
30	from App.version_txt import getZopeVersion	1✔
31	from DateTime import DateTime	1✔
32	from OFS.FindSupport import FindSupport	1✔
33	from OFS.metaconfigure import get_packages_to_initialize	1✔
34	from OFS.metaconfigure import package_initialized	1✔
35	from OFS.userfolder import UserFolder	1✔
36	from webdav.NullResource import NullResource	1✔
37	from zExceptions import Forbidden	1✔
38	from zExceptions import Redirect as RedirectException	1✔
39	from zope.interface import implementer	1✔
40
41	from . import Folder	1✔
42	from . import misc_	1✔
43	from .interfaces import IApplication	1✔
44	from .misc_ import Misc_	1✔
45
46
47	LOG = getLogger('Application')	1✔
48
49	APP_MANAGER = None	1✔
50
51
52	@implementer(IApplication)	1✔
53	class Application(ApplicationDefaultPermissions, Folder.Folder, FindSupport):	1✔
54	"""Top-level system object"""
55
56	security = ClassSecurityInfo()	1✔
57
58	title = 'Zope'	1✔
59	__defined_roles__ = ('Manager', 'Anonymous', 'Owner')	1✔
60	__error_log__ = None	1✔
61	isTopLevelPrincipiaApplicationObject = 1	1✔
62
63	p_ = misc_.p_	1✔
64	misc_ = misc_.misc_	1✔
65	_reserved_names = ('Control_Panel', )	1✔
66
67	# This class-default __allow_groups__ ensures that the
68	# emergency user can still access the system if the top-level
69	# UserFolder is deleted. This is necessary to allow people
70	# to replace the top-level UserFolder object.
71	__allow_groups__ = UserFolder()	1✔
72
73	def __init__(self):	1✔
74	# Initialize users
75	uf = UserFolder()	1✔
76	self.__allow_groups__ = uf	1✔
77	self._setObject('acl_users', uf)	1✔
78
79	def getId(self):	1✔
80	try:	1✔
81	return self.REQUEST['SCRIPT_NAME'][1:]	1✔
82	except (KeyError, TypeError):	1✔
83	return self.title	1✔
84
85	def title_and_id(self):	1✔
86	return self.title	1✔
87
88	def title_or_id(self):	1✔
89	return self.title	1✔
90
91	def __class_init__(self):	1✔
92	InitializeClass(self)	1✔
93
94	@property	1✔
95	def Control_Panel(self):	1✔
96	return APP_MANAGER.__of__(self)	1✔
97
98	def Redirect(self, destination, URL1):	1✔
99	# Utility function to allow user-controlled redirects.
100	# No docstring please, we do not want an open redirect
101	# available as url.
102	if destination.find('//') >= 0:	1!
103	raise RedirectException(destination)	1✔
104	raise RedirectException(f"{URL1}/{destination}")	×
105
106	ZopeRedirect = Redirect	1✔
107
108	def __bobo_traverse__(self, REQUEST, name=None):	1✔
109	if name is None:	1✔
110	# Make this more explicit, otherwise getattr(self, name)
111	# would raise a TypeError getattr(): attribute name must be string
112	return None	1✔
113
114	if name == 'Control_Panel':	1!
115	return APP_MANAGER.__of__(self)	×
116	try:	1✔
117	return getattr(self, name)	1✔
118	except AttributeError:	1✔
119	pass	1✔
120
121	try:	1✔
122	return self[name]	1✔
123	except KeyError:	1✔
124	pass	1✔
125
126	method = REQUEST.get('REQUEST_METHOD', 'GET')	1✔
127
128	if method not in ('GET', 'POST'):	1!
129	return NullResource(self, name, REQUEST).__of__(self)	×
130
131	# Waaa. unrestrictedTraverse calls us with a fake REQUEST.
132	# There is probably a better fix for this.
133	try:	1✔
134	REQUEST.RESPONSE.notFoundError(f"{name}\n{method}")	1✔
135	except AttributeError:	1✔
136	raise KeyError(name)	1✔
137
138	def ZopeTime(self, *args):	1✔
139	"""Utility function to return current date/time"""
140	return DateTime(*args)	1✔
141
142	@security.protected(view_management_screens)	1✔
143	def ZopeVersion(self, major=False):	1✔
144	"""Utility method to return the Zope version
145
146	Restricted to ZMI to prevent information disclosure
147	"""
148	zversion = getZopeVersion()	1✔
149	if major:	1!
150	return zversion.major	1✔
151	else:
152	version = f'{zversion.major}.{zversion.minor}.{zversion.micro}'	×
153	if zversion.status:	×
154	version += f'.{zversion.status}{zversion.release}'	×
155
156	return version	×
157
158	def DELETE(self, REQUEST, RESPONSE):	1✔
159	"""Delete a resource object."""
160	self.dav__init(REQUEST, RESPONSE)	×
161	raise Forbidden('This resource cannot be deleted.')	×
162
163	def MOVE(self, REQUEST, RESPONSE):	1✔
164	"""Move a resource to a new location."""
165	self.dav__init(REQUEST, RESPONSE)	×
166	raise Forbidden('This resource cannot be moved.')	×
167
168	def absolute_url(self, relative=0):	1✔
169	"""The absolute URL of the root object is BASE1 or "/".
170	"""
171	if relative:	1✔
172	return ''	1✔
173	try:	1✔
174	# Take advantage of computed URL cache
175	return self.REQUEST['BASE1']	1✔
176	except (AttributeError, KeyError):	×
177	return '/'	×
178
179	def absolute_url_path(self):	1✔
180	"""The absolute URL path of the root object is BASEPATH1 or "/".
181	"""
182	try:	1✔
183	return self.REQUEST['BASEPATH1'] or '/'	1✔
184	except (AttributeError, KeyError):	×
185	return '/'	×
186
187	def virtual_url_path(self):	1✔
188	"""The virtual URL path of the root object is empty.
189	"""
190	return ''	1✔
191
192	def getPhysicalRoot(self):	1✔
193	return self	1✔
194
195	def getPhysicalPath(self):	1✔
196	# Get the physical path of the object.
197	#
198	# Returns a path (an immutable sequence of strings) that can be used to
199	# access this object again later, for example in a copy/paste
200	# operation. getPhysicalRoot() and getPhysicalPath() are designed to
201	# operate together.
202	#
203	# We're at the base of the path.
204	return ('', )	1✔
205
206
207	InitializeClass(Application)	1✔
208
209
210	def initialize(app):	1✔
211	initializer = AppInitializer(app)	1✔
212	initializer.initialize()	1✔
213
214
215	class AppInitializer:	1✔
216	""" Initialize an Application object (called at startup) """
217
218	def __init__(self, app):	1✔
219	self.app = (app,)	1✔
220
221	def getApp(self):	1✔
222	# this is probably necessary, but avoid acquisition anyway
223	return self.app[0]	1✔
224
225	def commit(self, note):	1✔
226	transaction.get().note(note)	1✔
227	transaction.commit()	1✔
228
229	def initialize(self):	1✔
230	# make sure to preserve relative ordering of calls below.
231	self.install_app_manager()	1✔
232	self.install_required_roles()	1✔
233	self.install_inituser()	1✔
234	self.install_products()	1✔
235	self.install_standards()	1✔
236	self.install_virtual_hosting()	1✔
237	self.install_root_view()	1✔
238
239	def install_app_manager(self):	1✔
240	global APP_MANAGER
241	APP_MANAGER = ApplicationManager()	1✔
242
243	# Remove persistent Control Panel.
244	app = self.getApp()	1✔
245	app._p_activate()	1✔
246
247	if 'Control_Panel' in list(app.__dict__.keys()):	1!
248	del app.__dict__['Control_Panel']	×
249	app._objects = tuple(i for i in app._objects	×
250	if i['id'] != 'Control_Panel')
251	self.commit('Removed persistent Control_Panel')	×
252
253	def install_required_roles(self):	1✔
254	app = self.getApp()	1✔
255
256	# Ensure that Owner role exists.
257	if hasattr(app, '__ac_roles__') and not ('Owner' in app.__ac_roles__):	1!
258	app.__ac_roles__ = app.__ac_roles__ + ('Owner',)	×
259	self.commit('Added Owner role')	×
260
261	# ensure the Authenticated role exists.
262	if hasattr(app, '__ac_roles__'):	1!
263	if 'Authenticated' not in app.__ac_roles__:	1!
264	app.__ac_roles__ = app.__ac_roles__ + ('Authenticated',)	×
265	self.commit('Added Authenticated role')	×
266
267	def install_inituser(self):	1✔
268	app = self.getApp()	1✔
269	# Install the initial user.
270	if hasattr(app, 'acl_users'):	1!
271	users = app.acl_users	1✔
272	if hasattr(users, '_createInitialUser'):	1!
273	app.acl_users._createInitialUser()	1✔
274	self.commit('Created initial user')	1✔
275	users = aq_base(users)	1✔
276	migrated = getattr(users, '_ofs_migrated', False)	1✔
277	if not migrated:	1!
278	klass = users.__class__	×
279	from OFS.userfolder import UserFolder	×
280	if klass is UserFolder:	×
281	# zope.deferredimport does a thourough job, so the class
282	# looks like it's from the new location already. And we
283	# don't want to migrate any custom user folders here.
284	users.__class__ = UserFolder	×
285	users._ofs_migrated = True	×
286	users._p_changed = True	×
287	app._p_changed = True	×
288	transaction.get().note('Migrated user folder')	×
289	transaction.commit()	×
290
291	def install_virtual_hosting(self):	1✔
292	app = self.getApp()	1✔
293	if 'virtual_hosting' not in app:	1!
294	from Products.SiteAccess.VirtualHostMonster import \	1✔
295	VirtualHostMonster
296	any_vhm = [obj for obj in app.values()	1✔
297	if isinstance(obj, VirtualHostMonster)]
298	if not any_vhm:	1!
299	vhm = VirtualHostMonster()	1✔
300	vhm.id = 'virtual_hosting'	1✔
301	vhm.addToContainer(app)	1✔
302	self.commit('Added virtual_hosting')	1✔
303
304	def install_root_view(self):	1✔
305	app = self.getApp()	1✔
306	if 'index_html' not in app:	1!
307	from Products.PageTemplates.ZopePageTemplate import \	1✔
308	ZopePageTemplate
309	root_pt = ZopePageTemplate('index_html')	1✔
310	root_pt.pt_setTitle('Auto-generated default page')	1✔
311	app._setObject('index_html', root_pt)	1✔
312	self.commit('Added default view for root object')	1✔
313
314	def install_products(self):	1✔
315	return install_products(self.getApp())	1✔
316
317	def install_standards(self):	1✔
318	app = self.getApp()	1✔
319	if getattr(app, '_standard_objects_have_been_added', None) is not None:	1!
320	delattr(app, '_standard_objects_have_been_added')	×
321	if getattr(app, '_initializer_registry', None) is not None:	1!
322	delattr(app, '_initializer_registry')	×
323	transaction.get().note('Removed unused application attributes.')	1✔
324	transaction.commit()	1✔
325
326
327	def install_products(app=None):	1✔
328	folder_permissions = get_folder_permissions()	1✔
329	meta_types = []	1✔
330	done = {}	1✔
331	for priority, product_name, index, product_dir in get_products():	1✔
332	# For each product, we will import it and try to call the
333	# intialize() method in the product __init__ module. If
334	# the method doesnt exist, we put the old-style information
335	# together and do a default initialization.
336	if product_name in done:	1!
337	continue	×
338	done[product_name] = 1	1✔
339	install_product(app, product_dir, product_name, meta_types,	1✔
340	folder_permissions)
341
342	# Delayed install of packages-as-products
343	for module, init_func in tuple(get_packages_to_initialize()):	1✔
344	install_package(app, module, init_func)	1✔
345
346	Products.meta_types = Products.meta_types + tuple(meta_types)	1✔
347	InitializeClass(Folder.Folder)	1✔
348
349
350	def _is_package(product_dir, product_name):	1✔
351	package_dir = os.path.join(product_dir, product_name)	1✔
352	if not os.path.isdir(package_dir):	1✔
353	return False	1✔
354
355	init_py = os.path.join(package_dir, '__init__.py')	1✔
356	if not os.path.exists(init_py) and \	1✔
357	not os.path.exists(init_py + 'c') and \
358	not os.path.exists(init_py + 'o'):
359	return False	1✔
360	return True	1✔
361
362
363	def get_products():	1✔
364	""" Return a list of tuples in the form:
365	[(priority, dir_name, index, base_dir), ...] for each Product directory
366	found, sort before returning """
367	products = []	1✔
368	i = 0	1✔
369	for product_dir in Products.__path__:	1✔
370	product_names = os.listdir(product_dir)	1✔
371	for product_name in product_names:	1✔
372	if _is_package(product_dir, product_name):	1✔
373	# i is used as sort ordering in case a conflict exists
374	# between Product names. Products will be found as
375	# per the ordering of Products.__path__
376	products.append((0, product_name, i, product_dir))	1✔
377	i = i + 1	1✔
378	products.sort()	1✔
379	return products	1✔
380
381
382	def import_products():	1✔
383	done = {}	1✔
384	for priority, product_name, index, product_dir in get_products():	1✔
385	if product_name in done:	1!
386	LOG.warning(	×
387	'Duplicate Product name: '
388	'After loading Product %r from %r, '
389	'I skipped the one in %r.' % (
390	product_name, done[product_name], product_dir))
391	continue	×
392	done[product_name] = product_dir	1✔
393	import_product(product_dir, product_name)	1✔
394	return list(done.keys())	1✔
395
396
397	def import_product(product_dir, product_name, raise_exc=None):	1✔
398	if not _is_package(product_dir, product_name):	1!
399	return	×
400
401	global_dict = globals()	1✔
402	product = __import__("Products.%s" % product_name,	1✔
403	global_dict, global_dict, ('__doc__', ))
404	if hasattr(product, '__module_aliases__'):	1!
405	for k, v in product.__module_aliases__:	×
406	if k not in sys.modules:	×
407	if isinstance(v, str) and v in sys.modules:	×
408	v = sys.modules[v]	×
409	sys.modules[k] = v	×
410
411
412	def get_folder_permissions():	1✔
413	folder_permissions = {}	1✔
414	for p in Folder.Folder.__ac_permissions__:	1!
415	permission, names = p[:2]	×
416	folder_permissions[permission] = names	×
417	return folder_permissions	1✔
418
419
420	def install_product(app, product_dir, product_name, meta_types,	1✔
421	folder_permissions, raise_exc=None):
422	if not _is_package(product_dir, product_name):	1!
423	return	×
424
425	__traceback_info__ = product_name	1✔
426	global_dict = globals()	1✔
427	product = __import__("Products.%s" % product_name,	1✔
428	global_dict, global_dict, ('__doc__', ))
429
430	# Install items into the misc_ namespace, used by products
431	# and the framework itself to store common static resources
432	# like icon images.
433	misc_ = pgetattr(product, 'misc_', {})	1✔
434	if misc_:	1!
435	if isinstance(misc_, dict):	×
436	misc_ = Misc_(product_name, misc_)	×
437	setattr(Application.misc_, product_name, misc_)	×
438
439	productObject = FactoryDispatcher.Product(product_name)	1✔
440	context = ProductContext(productObject, app, product)	1✔
441
442	# Look for an 'initialize' method in the product.
443	initmethod = pgetattr(product, 'initialize', None)	1✔
444	if initmethod is not None:	1✔
445	initmethod(context)	1✔
446
447
448	def install_package(app, module, init_func, raise_exc=None):	1✔
449	"""Installs a Python package like a product."""
450	name = module.__name__	1✔
451	product = FactoryDispatcher.Product(name)	1✔
452	product.package_name = name	1✔
453
454	if init_func is not None:	1!
455	newContext = ProductContext(product, app, module)	1✔
456	init_func(newContext)	1✔
457
458	package_initialized(module, init_func)	1✔
459
460
461	def pgetattr(product, name, default=install_products, __init__=0):	1✔
462	if not __init__ and hasattr(product, name):	1✔
463	return getattr(product, name)	1✔
464	if hasattr(product, '__init__'):	1!
465	product = product.__init__	1✔
466	if hasattr(product, name):	1!
467	return getattr(product, name)	×
468
469	if default is not install_products:	1!
470	return default	1✔
471
472	raise AttributeError(name)	×

zopefoundation / Zope / 3956162881

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous