From 17878948dee26f94a883dae37c0e8ed9bd3eff83 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Radoslav=20Bod=C3=B3?= <bodik@cesnet.cz>
Date: Thu, 11 Apr 2024 16:19:22 +0200
Subject: [PATCH] rwm: drop error handling boilerplate for cli commands, also
drops storage_check_policy command
---
rwm.py | 31 +++++--------------------------
tests/test_default.py | 2 --
tests/test_rwm.py | 18 ++++--------------
tests/test_storage.py | 2 +-
4 files changed, 10 insertions(+), 43 deletions(-)
diff --git a/rwm.py b/rwm.py
index c629667..1a8692e 100755
--- a/rwm.py
+++ b/rwm.py
@@ -163,8 +163,8 @@ class StorageManager:
def storage_create(self, bucket_name, target_username):
"""create policed bucket"""
- if (not bucket_name) or (not target_username):
- raise ValueError("must specify value for bucket and user")
+ if not target_username:
+ raise ValueError("must specify value for bucket user")
bucket = self.bucket_create(bucket_name)
tenant, admin_username = bucket.Acl().owner["ID"].split("$")
@@ -333,6 +333,7 @@ class StorageManager:
def storage_save_state(self, bucket_name):
"""save storage state into itself"""
+ # explicit error handling here, it's used during backup process
try:
bucket_state = self._bucket_state(bucket_name)
now = datetime.now().astimezone().isoformat()
@@ -481,30 +482,13 @@ class RWM:
def storage_create(self, bucket_name, target_username) -> int:
"""storage create command"""
- try:
- self.storage_manager.storage_create(bucket_name, target_username)
- except (ClientError, BotoCoreError, ValueError) as exc:
- logger.error("rwm storage_create error, %s", (exc))
- return 1
+ self.storage_manager.storage_create(bucket_name, target_username)
return 0
def storage_delete(self, bucket_name) -> int:
"""storage delete command"""
- try:
- self.storage_manager.storage_delete(bucket_name)
- except (ClientError, BotoCoreError) as exc:
- logger.error("rwm storage_delete error, %s", (exc))
- return 1
- return 0
-
- def storage_check_policy(self, bucket_name) -> int:
- """storage check policy command"""
-
- ret, msg = (0, "OK") if self.storage_manager.storage_check_policy(bucket_name) else (1, "FAILED")
- logger.debug("bucket policy: %s", json.dumps(self.storage_manager.bucket_policy(bucket_name), indent=4))
- print(msg)
- return ret
+ return self.storage_manager.storage_delete(bucket_name)
def storage_list(self, show_full=False, name_filter="") -> int:
"""storage_list command"""
@@ -569,9 +553,6 @@ def parse_arguments(argv):
storage_delete_cmd_parser = subparsers.add_parser("storage_delete", help="delete storage")
storage_delete_cmd_parser.add_argument("bucket_name", help="bucket name")
- storage_check_policy_cmd_parser = subparsers.add_parser("storage_check_policy", help="check bucket policies; use --debug to show policy")
- storage_check_policy_cmd_parser.add_argument("bucket_name", help="bucket name")
-
storage_list_cmd_parser = subparsers.add_parser("storage_list", help="list storages")
storage_list_cmd_parser.add_argument("--full", action="store_true", help="show object counts")
storage_list_cmd_parser.add_argument("--filter", default="", help="name filter regex")
@@ -624,8 +605,6 @@ def main(argv=None): # pylint: disable=too-many-branches
ret = rwmi.storage_create(args.bucket_name, args.target_username)
if args.command == "storage_delete":
ret = rwmi.storage_delete(args.bucket_name)
- if args.command == "storage_check_policy":
- ret = rwmi.storage_check_policy(args.bucket_name)
if args.command == "storage_list":
ret = rwmi.storage_list(args.full, args.filter)
if args.command == "storage_drop_versions":
diff --git a/tests/test_default.py b/tests/test_default.py
index c9fece7..549de11 100644
--- a/tests/test_default.py
+++ b/tests/test_default.py
@@ -55,8 +55,6 @@ def test_main(tmpworkdir: str): # pylint: disable=unused-argument
assert rwm_main(["storage_create", "bucket", "user"]) == 0
with patch.object(rwm.RWM, "storage_delete", mock_ok):
assert rwm_main(["storage_delete", "bucket"]) == 0
- with patch.object(rwm.RWM, "storage_check_policy", mock_ok):
- assert rwm_main(["storage_check_policy", "bucket"]) == 0
with patch.object(rwm.RWM, "storage_list", mock_ok):
assert rwm_main(["storage_list"]) == 0
with patch.object(rwm.RWM, "storage_drop_versions", mock_ok):
diff --git a/tests/test_rwm.py b/tests/test_rwm.py
index 34002d9..1f5628c 100644
--- a/tests/test_rwm.py
+++ b/tests/test_rwm.py
@@ -5,6 +5,8 @@ from pathlib import Path
from subprocess import CompletedProcess
from unittest.mock import Mock, patch
+import pytest
+
import rwm
@@ -194,8 +196,8 @@ def test_storage_create(tmpworkdir: str, microceph: str, radosuser_admin: rwm.St
bucket_name = "testbuck"
assert trwm.storage_create(bucket_name, "testnx") == 0
- assert trwm.storage_create("!invalid", "testnx") == 1
- assert trwm.storage_create(bucket_name, "") == 1
+ with pytest.raises(ValueError):
+ trwm.storage_create(bucket_name, "")
def test_storage_delete(tmpworkdir: str, microceph: str, radosuser_admin: rwm.StorageManager): # pylint: disable=unused-argument
@@ -230,18 +232,6 @@ def test_storage_delete(tmpworkdir: str, microceph: str, radosuser_admin: rwm.St
assert trwm.storage_delete(bucket_name) == 0
assert not trwm.storage_manager.bucket_exist(bucket_name)
- assert trwm.storage_delete(bucket_name) == 1
-
-
-def test_storage_check_policy(tmpworkdir: str): # pylint: disable=unused-argument
- """test storage check policy"""
-
- trwm = rwm.RWM({})
-
- mock = Mock(return_value=False)
- with patch.object(rwm.StorageManager, "storage_check_policy", mock):
- assert trwm.storage_check_policy("dummy") == 1
-
def test_storage_list(tmpworkdir: str): # pylint: disable=unused-argument
"""test storage_list"""
diff --git a/tests/test_storage.py b/tests/test_storage.py
index c6254b3..99d18b9 100644
--- a/tests/test_storage.py
+++ b/tests/test_storage.py
@@ -60,7 +60,7 @@ def test_storage_create(
radosuser_test2.list_objects(bucket.name)
-def test_storage_delete(
+def test_storage_object_versioning(
tmpworkdir: str,
microceph: str,
radosuser_admin: rwm.StorageManager,
--
GitLab