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

67.8

/src/OFS/Lockable.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.
#
##############################################################################

from AccessControl.class_init import InitializeClass
from AccessControl.Permissions import webdav_lock_items
from AccessControl.Permissions import webdav_manage_locks
from AccessControl.Permissions import webdav_unlock_items
from AccessControl.SecurityInfo import ClassSecurityInfo
from Acquisition import aq_base
from OFS.EtagSupport import EtagSupport
from OFS.interfaces import ILockItem
from OFS.interfaces import IWriteLock
from Persistence import PersistentMapping
from zope.interface import implementer


@implementer(IWriteLock)
class LockableItem(EtagSupport):
    """Implements the WriteLock interface.
    """

    # Protect methods using declarative security
    security = ClassSecurityInfo()

    # Setting default roles for permissions - we want owners of conent
    # to be able to lock.
    security.setPermissionDefault(webdav_lock_items, ('Manager', 'Owner',))
    security.setPermissionDefault(webdav_unlock_items, ('Manager', 'Owner',))

    @security.private
    def wl_lockmapping(self, killinvalids=0, create=0):
        """ if 'killinvalids' is 1, locks who are no longer valid
        will be deleted """

        try:
            locks = getattr(self, '_dav_writelocks', None)
        except Exception:
            locks = None

        if locks is None:
            if create:
                locks = self._dav_writelocks = PersistentMapping()
            else:
                # Don't generate a side effect transaction.
                locks = {}
            return locks
        elif killinvalids:
            # Delete invalid locks
            for token, lock in list(locks.items()):
                if not lock.isValid():
                    del locks[token]
            if (not locks) and hasattr(aq_base(self),
                                       '__no_valid_write_locks__'):
                self.__no_valid_write_locks__()
            return locks
        else:
            return locks

    @security.public
    def wl_lockItems(self, killinvalids=0):
        return list(self.wl_lockmapping(killinvalids).items())

    @security.public
    def wl_lockValues(self, killinvalids=0):
        return list(self.wl_lockmapping(killinvalids).values())

    @security.public
    def wl_lockTokens(self, killinvalids=0):
        return list(self.wl_lockmapping(killinvalids).keys())

    # TODO: Security Declaration
    def wl_hasLock(self, token, killinvalids=0):
        if not token:
            return 0
        return token in list(self.wl_lockmapping(killinvalids).keys())

    @security.public
    def wl_isLocked(self):
        # returns true if 'self' is locked at all
        # We set 'killinvalids' to 1 to delete all locks who are no longer
        # valid (timeout has been exceeded)
        locks = self.wl_lockmapping(killinvalids=1)

        if list(locks.keys()):
            return 1
        else:
            return 0

    @security.protected(webdav_lock_items)
    def wl_setLock(self, locktoken, lock):
        locks = self.wl_lockmapping(create=1)
        if ILockItem.providedBy(lock):
            if locktoken == lock.getLockToken():
                locks[locktoken] = lock
            else:
                raise ValueError('Lock tokens do not match')
        else:
            raise ValueError('Lock does not implement the LockItem Interface')

    @security.public
    def wl_getLock(self, locktoken):
        locks = self.wl_lockmapping(killinvalids=1)
        return locks.get(locktoken, None)

    @security.protected(webdav_unlock_items)
    def wl_delLock(self, locktoken):
        locks = self.wl_lockmapping()
        if locktoken in locks:
            del locks[locktoken]

    @security.protected(webdav_manage_locks)
    def wl_clearLocks(self):
        # Called by lock management machinery to quickly and effectively
        # destroy all locks.
        try:
            locks = self.wl_lockmapping()
            locks.clear()
        except Exception:
            # The locks may be totally messed up, so we'll just delete
            # and replace.
            if hasattr(self, '_dav_writelocks'):
                del self._dav_writelocks
            if IWriteLock.providedBy(self):
                self._dav_writelocks = PersistentMapping()

        # Call into a special hook used by LockNullResources to delete
        # themselves.  Could be used by other objects who want to deal
        # with the state of empty locks.
        if hasattr(aq_base(self), '__no_valid_write_locks__'):
            self.__no_valid_write_locks__()


InitializeClass(LockableItem)


def wl_isLocked(ob):
    """ Returns true if the object is locked, returns 0 if the object
    is not locked or does not implement the WriteLockInterface """
    return wl_isLockable(ob) and ob.wl_isLocked()


def wl_isLockable(ob):
    return IWriteLock.providedBy(ob)

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
14	from AccessControl.class_init import InitializeClass	1✔
15	from AccessControl.Permissions import webdav_lock_items	1✔
16	from AccessControl.Permissions import webdav_manage_locks	1✔
17	from AccessControl.Permissions import webdav_unlock_items	1✔
18	from AccessControl.SecurityInfo import ClassSecurityInfo	1✔
19	from Acquisition import aq_base	1✔
20	from OFS.EtagSupport import EtagSupport	1✔
21	from OFS.interfaces import ILockItem	1✔
22	from OFS.interfaces import IWriteLock	1✔
23	from Persistence import PersistentMapping	1✔
24	from zope.interface import implementer	1✔
25
26
27	@implementer(IWriteLock)	1✔
28	class LockableItem(EtagSupport):	1✔
29	"""Implements the WriteLock interface.
30	"""
31
32	# Protect methods using declarative security
33	security = ClassSecurityInfo()	1✔
34
35	# Setting default roles for permissions - we want owners of conent
36	# to be able to lock.
37	security.setPermissionDefault(webdav_lock_items, ('Manager', 'Owner',))	1✔
38	security.setPermissionDefault(webdav_unlock_items, ('Manager', 'Owner',))	1✔
39
40	@security.private	1✔
41	def wl_lockmapping(self, killinvalids=0, create=0):	1✔
42	""" if 'killinvalids' is 1, locks who are no longer valid
43	will be deleted """
44
45	try:	1✔
46	locks = getattr(self, '_dav_writelocks', None)	1✔
47	except Exception:	×
48	locks = None	×
49
50	if locks is None:	1✔
51	if create:	1✔
52	locks = self._dav_writelocks = PersistentMapping()	1✔
53	else:
54	# Don't generate a side effect transaction.
55	locks = {}	1✔
56	return locks	1✔
57	elif killinvalids:	1!
58	# Delete invalid locks
59	for token, lock in list(locks.items()):	1✔
60	if not lock.isValid():	1!
61	del locks[token]	×
62	if (not locks) and hasattr(aq_base(self),	1!
63	'__no_valid_write_locks__'):
64	self.__no_valid_write_locks__()	×
65	return locks	1✔
66	else:
67	return locks	×
68
69	@security.public	1✔
70	def wl_lockItems(self, killinvalids=0):	1✔
71	return list(self.wl_lockmapping(killinvalids).items())	×
72
73	@security.public	1✔
74	def wl_lockValues(self, killinvalids=0):	1✔
75	return list(self.wl_lockmapping(killinvalids).values())	×
76
77	@security.public	1✔
78	def wl_lockTokens(self, killinvalids=0):	1✔
79	return list(self.wl_lockmapping(killinvalids).keys())	×
80
81	# TODO: Security Declaration
82	def wl_hasLock(self, token, killinvalids=0):	1✔
83	if not token:	×
84	return 0	×
85	return token in list(self.wl_lockmapping(killinvalids).keys())	×
86
87	@security.public	1✔
88	def wl_isLocked(self):	1✔
89	# returns true if 'self' is locked at all
90	# We set 'killinvalids' to 1 to delete all locks who are no longer
91	# valid (timeout has been exceeded)
92	locks = self.wl_lockmapping(killinvalids=1)	1✔
93
94	if list(locks.keys()):	1✔
95	return 1	1✔
96	else:
97	return 0	1✔
98
99	@security.protected(webdav_lock_items)	1✔
100	def wl_setLock(self, locktoken, lock):	1✔
101	locks = self.wl_lockmapping(create=1)	1✔
102	if ILockItem.providedBy(lock):	1!
103	if locktoken == lock.getLockToken():	1!
104	locks[locktoken] = lock	1✔
105	else:
106	raise ValueError('Lock tokens do not match')	×
107	else:
108	raise ValueError('Lock does not implement the LockItem Interface')	×
109
110	@security.public	1✔
111	def wl_getLock(self, locktoken):	1✔
112	locks = self.wl_lockmapping(killinvalids=1)	×
113	return locks.get(locktoken, None)	×
114
115	@security.protected(webdav_unlock_items)	1✔
116	def wl_delLock(self, locktoken):	1✔
117	locks = self.wl_lockmapping()	×
118	if locktoken in locks:	×
119	del locks[locktoken]	×
120
121	@security.protected(webdav_manage_locks)	1✔
122	def wl_clearLocks(self):	1✔
123	# Called by lock management machinery to quickly and effectively
124	# destroy all locks.
125	try:	1✔
126	locks = self.wl_lockmapping()	1✔
127	locks.clear()	1✔
128	except Exception:	×
129	# The locks may be totally messed up, so we'll just delete
130	# and replace.
131	if hasattr(self, '_dav_writelocks'):	×
132	del self._dav_writelocks	×
133	if IWriteLock.providedBy(self):	×
134	self._dav_writelocks = PersistentMapping()	×
135
136	# Call into a special hook used by LockNullResources to delete
137	# themselves. Could be used by other objects who want to deal
138	# with the state of empty locks.
139	if hasattr(aq_base(self), '__no_valid_write_locks__'):	1!
140	self.__no_valid_write_locks__()	×
141
142
143	InitializeClass(LockableItem)	1✔
144
145
146	def wl_isLocked(ob):	1✔
147	""" Returns true if the object is locked, returns 0 if the object
148	is not locked or does not implement the WriteLockInterface """
149	return wl_isLockable(ob) and ob.wl_isLocked()	1✔
150
151
152	def wl_isLockable(ob):	1✔
153	return IWriteLock.providedBy(ob)	1✔

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