You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@chemistry.apache.org by jp...@apache.org on 2017/03/05 20:03:33 UTC
svn commit: r1785568 - in /chemistry/cmislib/trunk/src:
cmislib/browser/binding.py cmislib/domain.py tests/test_browser.py
Author: jpotts
Date: Sun Mar 5 20:03:33 2017
New Revision: 1785568
URL: http://svn.apache.org/viewvc?rev=1785568&view=rev
Log:
Implement applyACL for the browser binding
Added:
chemistry/cmislib/trunk/src/tests/test_browser.py
Modified:
chemistry/cmislib/trunk/src/cmislib/browser/binding.py
chemistry/cmislib/trunk/src/cmislib/domain.py
Modified: chemistry/cmislib/trunk/src/cmislib/browser/binding.py
URL: http://svn.apache.org/viewvc/chemistry/cmislib/trunk/src/cmislib/browser/binding.py?rev=1785568&r1=1785567&r2=1785568&view=diff
==============================================================================
--- chemistry/cmislib/trunk/src/cmislib/browser/binding.py (original)
+++ chemistry/cmislib/trunk/src/cmislib/browser/binding.py Sun Mar 5 20:03:33 2017
@@ -540,9 +540,17 @@ class BrowserCmisObject(object):
# get the root folder URL
aclUrl = self._repository.getRootFolderUrl() + "?objectId=" + self.id + "&cmisaction=applyACL"
- aclJSON = ACLSerializer().toJSON(acl)
+ fields = {}
+ for i, entry in enumerate(acl.getAddedAces()):
+ fields['addACEPrincipal[%d]' % i] = entry.principalId
+ for j, perm in enumerate(entry.permissions):
+ fields['addACEPermission[%d][%d]'% (i, j)] = perm
+ for i, entry in enumerate(acl.getRemovedAces()):
+ fields['removeACEPrincipal[%d]' % i] = entry.principalId
+ for j, perm in enumerate(entry.permissions):
+ fields['removeACEPermission[%d][%d]' % (i, j)] = perm
- contentType, body = encode_multipart_formdata(None, StringIO.StringIO(aclJSON), 'application/json')
+ contentType, body = encode_multipart_formdata(fields, None, None)
# invoke the URL
result = self._cmisClient.binding.post(aclUrl.encode('utf-8'),
@@ -2723,19 +2731,35 @@ class BrowserACL(ACL):
the list--the constructor will convert it to a list for you.
"""
+ self._entries = {}
if aceList:
- self._entries = aceList
- else:
- self._entries = {}
+ for ace in aceList:
+ if not isinstance(ace, BrowserACE):
+ raise CmisException('Items into the aceList must be an instance of the BrowserACEclass.')
+ self._entries[ace.principalId] = ace
+
if data:
self._data = data
self._entries = self._getEntriesFromData()
else:
self._data = None
+ self._originalEntries = self._copy_entries()
+
self.logger = logging.getLogger('cmislib.browser.binding.BrowserACL')
self.logger.debug('Creating an instance of BrowserACL')
+ def _copy_entries(self):
+ """
+ Internal method used to keep a copy of the original entries of ACL
+ :return:
+ """
+
+ result = {}
+ for principalId, ace in self._entries.iteritems():
+ result[principalId] = ace.copy()
+ return result
+
def _getEntriesFromData(self):
"""
@@ -2797,11 +2821,8 @@ class BrowserACL(ACL):
>>> acl.addEntry(ACE('jpotts', 'cmis:write', 'true'))
>>> acl.entries
{'jpotts': <cmislib.model.ACE object at 0x1012c7310>, 'jsmith': <cmislib.model.ACE object at 0x100528490>}
- >>> acl.getXmlDoc()
- <xml.dom.minidom.Document instance at 0x1012cbb90>
>>> acl.clearEntries()
>>> acl.entries
- >>> acl.getXmlDoc()
"""
self._entries.clear()
@@ -2838,6 +2859,69 @@ class BrowserACL(ACL):
entries = property(getEntries)
+ def getOriginalEntries(self):
+ return self._originalEntries
+
+ originalEntries = property(getOriginalEntries)
+
+ def getRemovedAces(self):
+
+ """
+ Returns a list of removed ACE. The list is based on a difference
+ between the original data and the current state
+
+ """
+ entries = self.entries
+ originalEntries = self.originalEntries
+ removedAces = []
+ for principalId, original in originalEntries.iteritems():
+ current = entries.get(principalId)
+ if not current:
+ removedAces.append(original.copy())
+ continue
+ if current.direct != original.direct:
+ removedAces.append(original.copy())
+ continue
+ originalPerms = set(original.permissions)
+ currentPerms = set(current.permissions)
+ removedPerms = originalPerms - currentPerms
+ if removedPerms:
+ removedAces.append(BrowserACE(
+ principalId=principalId,
+ permissions=list(removedPerms),
+ direct=original.direct
+ ))
+ return removedAces
+
+ def getAddedAces(self):
+
+ """
+ Returns the list of new ACE. The list is based on a difference
+ between the original data and the current state
+
+ """
+ entries = self.entries
+ originalEntries = self.originalEntries
+ addedAces = []
+ for principalId, current in entries.iteritems():
+ original = originalEntries.get(principalId)
+ if not original:
+ addedAces.append(current.copy())
+ continue
+ if current.direct != original.direct:
+ addedAces.append(current.copy())
+ continue
+ originalPerms = set(original.permissions)
+ currentPerms = set(current.permissions)
+ addedPerms = currentPerms - originalPerms
+ if addedPerms:
+ addedAces.append(BrowserACE(
+ principalId=principalId,
+ permissions=list(addedPerms),
+ direct=current.direct
+ ))
+ return addedAces
+
class BrowserACE(ACE):
Modified: chemistry/cmislib/trunk/src/cmislib/domain.py
URL: http://svn.apache.org/viewvc/chemistry/cmislib/trunk/src/cmislib/domain.py?rev=1785568&r1=1785567&r2=1785568&view=diff
==============================================================================
--- chemistry/cmislib/trunk/src/cmislib/domain.py (original)
+++ chemistry/cmislib/trunk/src/cmislib/domain.py Sun Mar 5 20:03:33 2017
@@ -2024,6 +2024,24 @@ class ACE(object):
"""Getter for permissions"""
return self._permissions
+ def __cmp__(self, other):
+ if (
+ isinstance(other, self.__class__) and
+ self.principalId == other.principalId and
+ self.direct == other.direct and
+ not(set(self.permissions) ^ set(other.permissions))
+ ):
+ return 0
+ return -1
+
+ def copy(self):
+ """
+ return a deep copy of the ace instance
+ """
+ return self.__class__(
+ principalId=self.principalId, permissions=self.permissions[:],
+ direct=self.direct)
+
class ChangeEntry(object):
Added: chemistry/cmislib/trunk/src/tests/test_browser.py
URL: http://svn.apache.org/viewvc/chemistry/cmislib/trunk/src/tests/test_browser.py?rev=1785568&view=auto
==============================================================================
--- chemistry/cmislib/trunk/src/tests/test_browser.py (added)
+++ chemistry/cmislib/trunk/src/tests/test_browser.py Sun Mar 5 20:03:33 2017
@@ -0,0 +1,102 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+"""
+Unit tests for logic unique to the Browser binding
+"""
+
+import unittest
+from unittest import TestSuite, TestLoader
+from cmislib.browser.binding import BrowserACE
+from cmislib.browser.binding import BrowserACL
+
+
+class BrowserACLTest(unittest.TestCase):
+
+ def setUp(self):
+ self.aceUser1 = BrowserACE(
+ principalId='user1',permissions='cmis:read', direct=True)
+ self.aceUser2 = BrowserACE(
+ principalId='user2', permissions=['cmis:read', 'cmis:write'],
+ direct=False)
+ self.acl = BrowserACL(aceList=[self.aceUser1, self.aceUser2])
+
+ def test_original_entries(self):
+ originalEntries = self.acl.originalEntries
+ for entry in [self.aceUser1, self.aceUser2]:
+ copy = originalEntries.get(entry.principalId)
+ self.assertTrue(copy)
+ # check we have 2 different instances of the same object
+ self.assertNotEquals(id(entry), id(copy))
+ self.assertEqual(entry, copy)
+
+ def test_get_removed_aces(self):
+ # test the complete removal of an alc entry
+ self.acl.removeEntry(self.aceUser1.principalId)
+ removedAces = self.acl.getRemovedAces()
+ self.assertEqual(len(removedAces), 1)
+ self.assertEqual(removedAces[0], self.aceUser1)
+ # test partial removal of an entry (delete all + add an existing one
+ # with same direct)
+ self.acl.removeEntry(self.aceUser2.principalId)
+ self.acl.addEntry(self.aceUser2.principalId, 'cmis:write',
+ direct=False)
+ removedAces = self.acl.getRemovedAces()
+ self.assertEqual(len(removedAces), 2)
+ toCheck = None
+ for removedAce in removedAces:
+ if removedAce.principalId == self.aceUser2.principalId:
+ toCheck = removedAce
+ self.assertTrue(toCheck)
+ self.assertEqual(toCheck.principalId, self.aceUser2.principalId)
+ self.assertListEqual(toCheck.permissions, ['cmis:read'])
+
+ def test_get_added_aces(self):
+ # add new entry for a new princpal
+ self.acl.addEntry('user3', 'cmis:all')
+ addedAces = self.acl.getAddedAces()
+ self.assertEqual(len(addedAces), 1)
+ self.assertEqual(addedAces[0], BrowserACE('user3', 'cmis:all', True))
+ # add a new entry for the same principal
+ self.acl.addEntry(
+ 'user3', ['cmis:all', 'cmis:write'])
+ addedAces = self.acl.getAddedAces()
+ self.assertEqual(len(addedAces), 1)
+ self.assertEqual(
+ addedAces[0],
+ BrowserACE('user3', ['cmis:all', 'cmis:write'], True))
+ # add a new entry for an exising principal
+ self.acl.addEntry(
+ self.aceUser1.principalId, ['cmis:read','cmis:write'])
+ addedAces = self.acl.getAddedAces()
+ self.assertEqual(len(addedAces), 2)
+ toCheck = None
+ for addedAce in addedAces:
+ if addedAce.principalId == self.aceUser1.principalId:
+ toCheck = addedAce
+ self.assertTrue(toCheck)
+ self.assertEqual(
+ toCheck,
+ BrowserACE(self.aceUser1.principalId, ['cmis:write'], True))
+
+if __name__ == "__main__":
+ tts = TestSuite()
+ tts.addTests(TestLoader().loadTestsFromTestCase(BrowserACLTest))
+ unittest.TextTestRunner().run(tts)