zookeeper/zookeeper-contrib/zookeeper-contrib-zkpython/src/test/acl_test.py at 91666c5a0b9ea1b463f41ec9f93bb2fa2ba9e68f · apache/zookeeper · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
#!/usr/bin/python
#
#  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.

import zookeeper, zktestbase, unittest, threading

ZOO_OPEN_ACL_UNSAFE = {"perms":zookeeper.PERM_ALL, "scheme":"world", "id" :"anyone"}
ZOO_ACL_READ = {"perms":zookeeper.PERM_READ, "scheme": "world",
                "id":"anyone"}
class ACLTest(zktestbase.TestBase):
    """Test whether basic ACL setting and getting work correctly"""
    # to do: startup and teardown via scripts?
    def setUp(self):
      zktestbase.TestBase.setUp(self)
      try:
        zookeeper.delete(self.handle, "/zk-python-acltest")
        zookeeper.delete(self.handle, "/zk-python-aacltest")
      except Exception:
        pass

    def test_sync_acl(self):
      self.assertEqual(self.connected, True)
      ret = zookeeper.create(self.handle, "/zk-python-acltest", "nodecontents", [ZOO_OPEN_ACL_UNSAFE], zookeeper.EPHEMERAL)
      acls = zookeeper.get_acl(self.handle, "/zk-python-acltest")
      self.assertEqual(acls[1], [ZOO_OPEN_ACL_UNSAFE])
      self.assertRaises(zookeeper.InvalidACLException,zookeeper.set_acl,self.handle, "/zk-python-acltest", -1, ZOO_ACL_READ)
      zookeeper.set_acl(self.handle, "/zk-python-acltest", -1, [ZOO_ACL_READ])
      acls = zookeeper.get_acl(self.handle, "/zk-python-acltest")
      self.assertEqual(acls[1], [ZOO_ACL_READ])


    def test_async_acl(self):
      self.cv = threading.Condition()
      self.cv = threading.Condition()
      def aget_callback(handle, rc, acl, stat):
        self.cv.acquire()
        self.callback_flag = True
        self.rc = rc
        self.acl = acl
        self.stat = stat
        self.cv.notify()
        self.cv.release()

      def aset_callback(handle, rc):
        self.cv.acquire()
        self.callback_flag = True
        self.rc = rc
        self.cv.notify()
        self.cv.release()

      self.assertEqual(self.connected, True, "Not connected!")
      ret = zookeeper.create(self.handle, "/zk-python-aacltest", "nodecontents", [ZOO_OPEN_ACL_UNSAFE], zookeeper.EPHEMERAL)

      self.cv.acquire()
      zookeeper.aget_acl(self.handle, "/zk-python-aacltest", aget_callback)
      self.cv.wait(15)
      self.cv.release()

      self.assertEqual(self.callback_flag, True, "aget_acl timed out")
      self.assertEqual(self.rc, zookeeper.OK, "aget failed")
      self.assertEqual(self.acl, [ZOO_OPEN_ACL_UNSAFE], "Wrong ACL returned from aget")

      self.cv.acquire()
      self.callback_flag = False
      zookeeper.aset_acl(self.handle, "/zk-python-aacltest", -1, [ZOO_ACL_READ], aset_callback)
      self.cv.wait(15)
      self.cv.release()

      self.assertEqual(self.callback_flag, True, "aset_acl timed out")
      self.assertEqual(self.rc, zookeeper.OK, "aset failed")
      acls = zookeeper.get_acl(self.handle, "/zk-python-aacltest")
      self.assertEqual(acls[1], [ZOO_ACL_READ], "Wrong ACL returned from get when aset")

    def test_invalid_acl(self):
      self.assertRaises(zookeeper.InvalidACLException,
                        zookeeper.create,
                        self.handle,
                        "/zk-python-aclverifytest",
                        "",
                        None,
                        zookeeper.EPHEMERAL)

    def test_invalid_acl2(self):
      """Verify all required keys are present in the ACL."""
      invalid_acl = [{"schema": "digest", "id": "zebra"}]
      self.assertRaises(zookeeper.InvalidACLException,
                        zookeeper.create,
                        self.handle,
                        "/zk-python-aclverifytest",
                        "",
                        invalid_acl,
                        zookeeper.EPHEMERAL)

if __name__ == '__main__':
    unittest.main()