feat(storage): support optionsRequestedPolicyVersion

api_core/tests/unit/test_iam.py

@@ -14,6 +14,8 @@
 
 import pytest
 
+from google.api_core.iam import _DICT_ACCESS_MSG, InvalidOperationException
+
 
 class TestPolicy:
     @staticmethod
@@ -37,7 +39,7 @@ def test_ctor_defaults(self):
         assert dict(policy) == {}
 
     def test_ctor_explicit(self):
-        VERSION = 17
+        VERSION = 1
         ETAG = "ETAG"
         empty = frozenset()
         policy = self._make_one(ETAG, VERSION)
@@ -53,6 +55,21 @@ def test___getitem___miss(self):
         policy = self._make_one()
         assert policy["nonesuch"] == set()
 
+    def test___getitem___version3(self):
+        policy = self._make_one("DEADBEEF", 3)
+        with pytest.raises(InvalidOperationException, match=_DICT_ACCESS_MSG):
+            policy["role"]
+
+    def test___getitem___with_conditions(self):
+        USER = "user:phred@example.com"
+        CONDITION = {"expression": "2 > 1"}
+        policy = self._make_one("DEADBEEF", 1)
+        policy.bindings = [
+            {"role": "role/reader", "members": [USER], "condition": CONDITION}
+        ]
+        with pytest.raises(InvalidOperationException, match=_DICT_ACCESS_MSG):
+            policy["role/reader"]
+
     def test___setitem__(self):
         USER = "user:phred@example.com"
         PRINCIPALS = set([USER])
@@ -62,18 +79,70 @@ def test___setitem__(self):
         assert len(policy) == 1
         assert dict(policy) == {"rolename": PRINCIPALS}
 
+    def test__set_item__overwrite(self):
+        USER = "user:phred@example.com"
+        ALL_USERS = "allUsers"
+        MEMBERS = set([ALL_USERS])
+        policy = self._make_one()
+        policy["rolename"] = [USER]
+        policy["rolename"] = [ALL_USERS]
+        assert policy["rolename"] == MEMBERS
+        assert len(policy) == 1
+        assert dict(policy) == {"rolename": MEMBERS}
+
+    def test___setitem___version3(self):
+        policy = self._make_one("DEADBEEF", 3)
+        with pytest.raises(InvalidOperationException, match=_DICT_ACCESS_MSG):
+            policy["role/reader"] = ["user:phred@example.com"]
+
+    def test___setitem___with_conditions(self):
+        USER = "user:phred@example.com"
+        CONDITION = {"expression": "2 > 1"}
+        policy = self._make_one("DEADBEEF", 1)
+        policy.bindings = [
+            {"role": "role/reader", "members": set([USER]), "condition": CONDITION}
+        ]
+        with pytest.raises(InvalidOperationException, match=_DICT_ACCESS_MSG):
+            policy["role/reader"] = ["user:phred@example.com"]
+
     def test___delitem___hit(self):
         policy = self._make_one()
-        policy._bindings["rolename"] = ["phred@example.com"]
-        del policy["rolename"]
-        assert len(policy) == 0
-        assert dict(policy) == {}
+        policy.bindings = [
+            {"role": "to/keep", "members": set(["phred@example.com"])},
+            {"role": "to/remove", "members": set(["phred@example.com"])}
+        ]
+        del policy["to/remove"]
+        assert len(policy) == 1
+        assert dict(policy) == {"to/keep": set(["phred@example.com"])}
 
     def test___delitem___miss(self):
         policy = self._make_one()
         with pytest.raises(KeyError):
             del policy["nonesuch"]
 
+    def test___delitem___version3(self):
+        policy = self._make_one("DEADBEEF", 3)
+        with pytest.raises(InvalidOperationException, match=_DICT_ACCESS_MSG):
+            del policy["role/reader"]
+
+    def test___delitem___with_conditions(self):
+        USER = "user:phred@example.com"
+        CONDITION = {"expression": "2 > 1"}
+        policy = self._make_one("DEADBEEF", 1)
+        policy.bindings = [
+            {"role": "role/reader", "members": set([USER]), "condition": CONDITION}
+        ]
+        with pytest.raises(InvalidOperationException, match=_DICT_ACCESS_MSG):
+            del policy["role/reader"]
+
+    def test_bindings_property(self):
+        USER = "user:phred@example.com"
+        CONDITION = {"expression": "2 > 1"}
+        policy = self._make_one()
+        BINDINGS = [{"role": "role/reader", "members": set([USER]), "condition": CONDITION}]
+        policy.bindings = BINDINGS
+        assert policy.bindings == BINDINGS
+
     def test_owners_getter(self):
         from google.api_core.iam import OWNER_ROLE
 
@@ -94,7 +163,7 @@ def test_owners_setter(self):
         with warnings.catch_warnings(record=True) as warned:
             policy.owners = [MEMBER]
 
-        warning, = warned
+        (warning,) = warned
         assert warning.category is DeprecationWarning
         assert policy[OWNER_ROLE] == expected
 
@@ -118,7 +187,7 @@ def test_editors_setter(self):
         with warnings.catch_warnings(record=True) as warned:
             policy.editors = [MEMBER]
 
-        warning, = warned
+        (warning,) = warned
         assert warning.category is DeprecationWarning
         assert policy[EDITOR_ROLE] == expected
 
@@ -142,41 +211,77 @@ def test_viewers_setter(self):
         with warnings.catch_warnings(record=True) as warned:
             policy.viewers = [MEMBER]
 
-        warning, = warned
+        (warning,) = warned
         assert warning.category is DeprecationWarning
         assert policy[VIEWER_ROLE] == expected
 
     def test_user(self):
+        import warnings
+
         EMAIL = "phred@example.com"
         MEMBER = "user:%s" % (EMAIL,)
         policy = self._make_one()
-        assert policy.user(EMAIL) == MEMBER
+        with warnings.catch_warnings(record=True) as warned:
+            assert policy.user(EMAIL) == MEMBER
+
+        (warning,) = warned
+        assert warning.category is DeprecationWarning
 
     def test_service_account(self):
+        import warnings
+
         EMAIL = "phred@example.com"
         MEMBER = "serviceAccount:%s" % (EMAIL,)
         policy = self._make_one()
-        assert policy.service_account(EMAIL) == MEMBER
+        with warnings.catch_warnings(record=True) as warned:
+            assert policy.service_account(EMAIL) == MEMBER
+
+        (warning,) = warned
+        assert warning.category is DeprecationWarning
 
     def test_group(self):
+        import warnings
+
         EMAIL = "phred@example.com"
         MEMBER = "group:%s" % (EMAIL,)
         policy = self._make_one()
-        assert policy.group(EMAIL) == MEMBER
+        with warnings.catch_warnings(record=True) as warned:
+            assert policy.group(EMAIL) == MEMBER
+
+        (warning,) = warned
+        assert warning.category is DeprecationWarning
 
     def test_domain(self):
+        import warnings
+
         DOMAIN = "example.com"
         MEMBER = "domain:%s" % (DOMAIN,)
         policy = self._make_one()
-        assert policy.domain(DOMAIN) == MEMBER
+        with warnings.catch_warnings(record=True) as warned:
+            assert policy.domain(DOMAIN) == MEMBER
+
+        (warning,) = warned
+        assert warning.category is DeprecationWarning
 
     def test_all_users(self):
+        import warnings
+
         policy = self._make_one()
-        assert policy.all_users() == "allUsers"
+        with warnings.catch_warnings(record=True) as warned:
+            assert policy.all_users() == "allUsers"
+
+        (warning,) = warned
+        assert warning.category is DeprecationWarning
 
     def test_authenticated_users(self):
+        import warnings
+
         policy = self._make_one()
-        assert policy.authenticated_users() == "allAuthenticatedUsers"
+        with warnings.catch_warnings(record=True) as warned:
+            assert policy.authenticated_users() == "allAuthenticatedUsers"
+
+        (warning,) = warned
+        assert warning.category is DeprecationWarning
 
     def test_from_api_repr_only_etag(self):
         empty = frozenset()
@@ -201,7 +306,7 @@ def test_from_api_repr_complete(self):
         VIEWER2 = "user:phred@example.com"
         RESOURCE = {
             "etag": "DEADBEEF",
-            "version": 17,
+            "version": 1,
             "bindings": [
                 {"role": OWNER_ROLE, "members": [OWNER1, OWNER2]},
                 {"role": EDITOR_ROLE, "members": [EDITOR1, EDITOR2]},
@@ -211,7 +316,7 @@ def test_from_api_repr_complete(self):
         klass = self._get_target_class()
         policy = klass.from_api_repr(RESOURCE)
         assert policy.etag == "DEADBEEF"
-        assert policy.version == 17
+        assert policy.version == 1
         assert policy.owners, frozenset([OWNER1 == OWNER2])
         assert policy.editors, frozenset([EDITOR1 == EDITOR2])
         assert policy.viewers, frozenset([VIEWER1 == VIEWER2])
@@ -220,19 +325,24 @@ def test_from_api_repr_complete(self):
             EDITOR_ROLE: set([EDITOR1, EDITOR2]),
             VIEWER_ROLE: set([VIEWER1, VIEWER2]),
         }
+        assert policy.bindings == [
+            {"role": OWNER_ROLE, "members": set([OWNER1, OWNER2])},
+            {"role": EDITOR_ROLE, "members": set([EDITOR1, EDITOR2])},
+            {"role": VIEWER_ROLE, "members": set([VIEWER1, VIEWER2])},
+        ]
 
     def test_from_api_repr_unknown_role(self):
         USER = "user:phred@example.com"
         GROUP = "group:cloud-logs@google.com"
         RESOURCE = {
             "etag": "DEADBEEF",
-            "version": 17,
+            "version": 1,
             "bindings": [{"role": "unknown", "members": [USER, GROUP]}],
         }
         klass = self._get_target_class()
         policy = klass.from_api_repr(RESOURCE)
         assert policy.etag == "DEADBEEF"
-        assert policy.version == 17
+        assert policy.version == 1
         assert dict(policy), {"unknown": set([GROUP == USER])}
 
     def test_to_api_repr_defaults(self):
@@ -276,13 +386,13 @@ def test_to_api_repr_full(self):
             {"role": EDITOR_ROLE, "members": [EDITOR1, EDITOR2]},
             {"role": VIEWER_ROLE, "members": [VIEWER1, VIEWER2]},
         ]
-        policy = self._make_one("DEADBEEF", 17)
+        policy = self._make_one("DEADBEEF", 1)
         with warnings.catch_warnings(record=True):
             policy.owners = [OWNER1, OWNER2]
             policy.editors = [EDITOR1, EDITOR2]
             policy.viewers = [VIEWER1, VIEWER2]
         resource = policy.to_api_repr()
         assert resource["etag"] == "DEADBEEF"
-        assert resource["version"] == 17
+        assert resource["version"] == 1
         key = operator.itemgetter("role")
         assert sorted(resource["bindings"], key=key) == sorted(BINDINGS, key=key)