Skip to content

Commit

Permalink
Add 'HMACKeyMetadata.{exists,reload,update,delete}' API methods. (#8045)
Browse files Browse the repository at this point in the history
  • Loading branch information
@tseaver
tseaver committed Jul 2, 2019
1 parent 8fbb465 commit 0e25908
Show file tree
Hide file tree
Showing 2 changed files with 291 additions and 1 deletion.
61 changes: 61 additions & 0 deletions storage/google/cloud/storage/hmac_key.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.

from google.cloud.exceptions import NotFound
from google.cloud._helpers import _rfc3339_to_datetime


Expand Down Expand Up @@ -124,3 +125,63 @@ def updated(self):
value = self._properties.get("updated")
if value is not None:
return _rfc3339_to_datetime(value)

@property
def path(self):
"""Resource path for the metadata's key."""

if self.access_id is None:
raise ValueError("No 'access_id' set.")

project = self.project
if project is None:
project = self._client.project

return "/projects/{}/hmacKeys/{}".format(project, self.access_id)

def exists(self):
"""Determine whether or not the key for this metadata exists.
:rtype: bool
:returns: True if the key exists in Cloud Storage.
"""
try:
self._client._connection.api_request(method="GET", path=self.path)
except NotFound:
return False
else:
return True

def reload(self):
"""Reload properties from Cloud Storage.
:raises :class:`~google.api_core.exceptions.NotFound`:
if the key does not exist on the back-end.
"""
self._properties = self._client._connection.api_request(
method="GET", path=self.path
)

def update(self):
"""Save writable properties to Cloud Storage.
:raises :class:`~google.api_core.exceptions.NotFound`:
if the key does not exist on the back-end.
"""
payload = {"state": self.state}
self._properties = self._client._connection.api_request(
method="POST", path=self.path, data=payload
)

def delete(self):
"""Delete the key from Cloud Storage.
:raises :class:`~google.api_core.exceptions.NotFound`:
if the key does not exist on the back-end.
"""
if self.state != self.INACTIVE_STATE:
raise ValueError("Cannot delete key if not in 'INACTIVE' state.")

self._client._connection.api_request(
method="DELETE", path=self.path
)
231 changes: 230 additions & 1 deletion storage/tests/unit/test_hmac_key.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@

import unittest

import mock


class TestHMACKeyMetadata(unittest.TestCase):
@staticmethod
Expand All @@ -24,7 +26,7 @@ def _get_target_class():

def _make_one(self, client=None, *args, **kw):
if client is None:
client = object()
client = _Client()
return self._get_target_class()(client, *args, **kw)

def test_ctor_defaults(self):
Expand Down Expand Up @@ -114,3 +116,230 @@ def test_updated_getter(self):
now_stamp = "{}Z".format(now.isoformat())
metadata._properties["updated"] = now_stamp
self.assertEqual(metadata.updated, now.replace(tzinfo=UTC))

def test_path_wo_access_id(self):
metadata = self._make_one()

with self.assertRaises(ValueError):
metadata.path

def test_path_w_access_id_wo_project(self):
access_id = "ACCESS-ID"
client = _Client()
metadata = self._make_one()
metadata._properties["accessId"] = access_id

expected_path = "/projects/{}/hmacKeys/{}".format(
client.DEFAULT_PROJECT, access_id
)
self.assertEqual(metadata.path, expected_path)

def test_path_w_access_id_w_explicit_project(self):
access_id = "ACCESS-ID"
project = "OTHER-PROJECT"
metadata = self._make_one()
metadata._properties["accessId"] = access_id
metadata._properties["projectId"] = project

expected_path = "/projects/{}/hmacKeys/{}".format(project, access_id)
self.assertEqual(metadata.path, expected_path)

def test_exists_miss_no_project_set(self):
from google.cloud.exceptions import NotFound

access_id = "ACCESS-ID"
connection = mock.Mock(spec=["api_request"])
connection.api_request.side_effect = NotFound("testing")
client = _Client(connection)
metadata = self._make_one(client)
metadata._properties["accessId"] = access_id

self.assertFalse(metadata.exists())

expected_path = "/projects/{}/hmacKeys/{}".format(
client.DEFAULT_PROJECT, access_id
)
expected_kwargs = {"method": "GET", "path": expected_path}
connection.api_request.assert_called_once_with(**expected_kwargs)

def test_exists_hit_w_project_set(self):
project = "PROJECT-ID"
access_id = "ACCESS-ID"
email = "service-account@example.com"
resource = {
"kind": "storage#hmacKeyMetadata",
"accessId": access_id,
"serviceAccountEmail": email,
}
connection = mock.Mock(spec=["api_request"])
connection.api_request.return_value = resource
client = _Client(connection)
metadata = self._make_one(client)
metadata._properties["accessId"] = access_id
metadata._properties["projectId"] = project

self.assertTrue(metadata.exists())

expected_path = "/projects/{}/hmacKeys/{}".format(project, access_id)
expected_kwargs = {"method": "GET", "path": expected_path}
connection.api_request.assert_called_once_with(**expected_kwargs)

def test_reload_miss_no_project_set(self):
from google.cloud.exceptions import NotFound

access_id = "ACCESS-ID"
connection = mock.Mock(spec=["api_request"])
connection.api_request.side_effect = NotFound("testing")
client = _Client(connection)
metadata = self._make_one(client)
metadata._properties["accessId"] = access_id

with self.assertRaises(NotFound):
metadata.reload()

expected_path = "/projects/{}/hmacKeys/{}".format(
client.DEFAULT_PROJECT, access_id
)
expected_kwargs = {"method": "GET", "path": expected_path}
connection.api_request.assert_called_once_with(**expected_kwargs)

def test_reload_hit_w_project_set(self):
project = "PROJECT-ID"
access_id = "ACCESS-ID"
email = "service-account@example.com"
resource = {
"kind": "storage#hmacKeyMetadata",
"accessId": access_id,
"serviceAccountEmail": email,
}
connection = mock.Mock(spec=["api_request"])
connection.api_request.return_value = resource
client = _Client(connection)
metadata = self._make_one(client)
metadata._properties["accessId"] = access_id
metadata._properties["projectId"] = project

metadata.reload()

self.assertEqual(metadata._properties, resource)

expected_path = "/projects/{}/hmacKeys/{}".format(project, access_id)
expected_kwargs = {"method": "GET", "path": expected_path}
connection.api_request.assert_called_once_with(**expected_kwargs)

def test_update_miss_no_project_set(self):
from google.cloud.exceptions import NotFound

access_id = "ACCESS-ID"
connection = mock.Mock(spec=["api_request"])
connection.api_request.side_effect = NotFound("testing")
client = _Client(connection)
metadata = self._make_one(client)
metadata._properties["accessId"] = access_id
metadata.state = "INACTIVE"

with self.assertRaises(NotFound):
metadata.update()

expected_path = "/projects/{}/hmacKeys/{}".format(
client.DEFAULT_PROJECT, access_id
)
expected_kwargs = {
"method": "POST",
"path": expected_path,
"data": {
"state": "INACTIVE",
},
}
connection.api_request.assert_called_once_with(**expected_kwargs)

def test_update_hit_w_project_set(self):
project = "PROJECT-ID"
access_id = "ACCESS-ID"
email = "service-account@example.com"
resource = {
"kind": "storage#hmacKeyMetadata",
"accessId": access_id,
"serviceAccountEmail": email,
"state": "ACTIVE"
}
connection = mock.Mock(spec=["api_request"])
connection.api_request.return_value = resource
client = _Client(connection)
metadata = self._make_one(client)
metadata._properties["accessId"] = access_id
metadata._properties["projectId"] = project
metadata.state = "ACTIVE"

metadata.update()

self.assertEqual(metadata._properties, resource)

expected_path = "/projects/{}/hmacKeys/{}".format(project, access_id)
expected_kwargs = {
"method": "POST",
"path": expected_path,
"data": {
"state": "ACTIVE",
},
}
connection.api_request.assert_called_once_with(**expected_kwargs)

def test_delete_not_inactive(self):
metadata = self._make_one()
for state in ("ACTIVE", "DELETED"):
metadata._properties["state"] = state

with self.assertRaises(ValueError):
metadata.delete()

def test_delete_miss_no_project_set(self):
from google.cloud.exceptions import NotFound

access_id = "ACCESS-ID"
connection = mock.Mock(spec=["api_request"])
connection.api_request.side_effect = NotFound("testing")
client = _Client(connection)
metadata = self._make_one(client)
metadata._properties["accessId"] = access_id
metadata.state = "INACTIVE"

with self.assertRaises(NotFound):
metadata.delete()

expected_path = "/projects/{}/hmacKeys/{}".format(
client.DEFAULT_PROJECT, access_id
)
expected_kwargs = {
"method": "DELETE",
"path": expected_path,
}
connection.api_request.assert_called_once_with(**expected_kwargs)

def test_delete_hit_w_project_set(self):
project = "PROJECT-ID"
access_id = "ACCESS-ID"
connection = mock.Mock(spec=["api_request"])
connection.api_request.return_value = {}
client = _Client(connection)
metadata = self._make_one(client)
metadata._properties["accessId"] = access_id
metadata._properties["projectId"] = project
metadata.state = "INACTIVE"

metadata.delete()

expected_path = "/projects/{}/hmacKeys/{}".format(project, access_id)
expected_kwargs = {
"method": "DELETE",
"path": expected_path,
}
connection.api_request.assert_called_once_with(**expected_kwargs)


class _Client(object):
DEFAULT_PROJECT = "project-123"

def __init__(self, connection=None, project=DEFAULT_PROJECT):
self._connection = connection
self.project = project

0 comments on commit 0e25908

Please sign in to comment.