diff --git a/rwm.py b/rwm.py index 1ed29626275461d5e55c045fc59eee27b9eba48e..8433e93c220bce38cf40fc6ea450c0908b6cc4a3 100755 --- a/rwm.py +++ b/rwm.py @@ -65,6 +65,16 @@ def wrap_output(process): return process.returncode +def size_fmt(num): + """print value formated with human readable units""" + + for unit in ['B', 'KiB', 'MiB', 'GiB', 'TiB', 'PiB', 'EiB', 'ZiB']: + if abs(num) < 1024.0: + return f'{num:0.1f} {unit}' + num /= 1024.0 + return f'{num:0.1f} YiB' + + @dataclasses.dataclass class BackupResult: """backup results data container""" @@ -239,11 +249,10 @@ class StorageManager: output = [] for bucket in buckets: - result = { - "name": bucket.name, - "policy": "OK" if self.storage_check_policy(bucket.name) else "FAILED", - "owner": self.bucket_owner(bucket.name).split("$")[-1] - } + result = {} + result["name"] = bucket.name + result["policy"] = "OK" if self.storage_check_policy(bucket.name) else "FAILED" + result["owner"] = self.bucket_owner(bucket.name).split("$")[-1] if result["policy"] == "OK": user_statement = self._policy_statements_user(self.bucket_policy(bucket.name))[0] @@ -253,12 +262,22 @@ class StorageManager: if show_full: result["objects"] = 0 - result["old_versions"] = 0 result["delete_markers"] = 0 + result["old_versions"] = 0 + result["size"] = 0 + result["old_size"] = 0 + for page in paginator.paginate(Bucket=bucket.name): for obj in page.get("Versions", []): - result["objects" if obj["IsLatest"] else "old_versions"] += 1 + if obj["IsLatest"]: + result["objects"] += 1 + result["size"] += obj["Size"] + else: + result["old_versions"] += 1 + result["old_size"] += obj["Size"] result["delete_markers"] += len(page.get("DeleteMarkers", [])) + result["size"] = size_fmt(result["size"]) + result["old_size"] = size_fmt(result["old_size"]) output.append(result) diff --git a/tests/test_default.py b/tests/test_default.py index 2386f8ded3788e1bff852e45b16ea8514f1ec07c..2059e6f24c7a9018be8b6328bd74323dfba7ee25 100644 --- a/tests/test_default.py +++ b/tests/test_default.py @@ -5,7 +5,7 @@ from subprocess import CompletedProcess from unittest.mock import Mock, patch import rwm -from rwm import is_sublist, main as rwm_main, wrap_output +from rwm import is_sublist, main as rwm_main, size_fmt, wrap_output def test_sublist(): @@ -22,6 +22,13 @@ def test_wrap_output(): assert wrap_output(CompletedProcess(args='dummy', returncode=11, stdout="dummy", stderr="dummy")) == 11 +def test_size_fmt(): + """test sizefmt""" + + assert size_fmt(1024) == "1.0 KiB" + assert size_fmt(10**25) == "8.3 YiB" + + def test_main(tmpworkdir: str): # pylint: disable=unused-argument """test main""" diff --git a/tests/test_storage.py b/tests/test_storage.py index e83fea2cfdf262a0dfa6f2921069dc62cef23e42..c6254b3ba28e4ff4c7b58605f54398807432ed18 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -161,6 +161,8 @@ def test_storage_list( radosuser_admin.bucket_create("no-acl-dummy") bucket = radosuser_admin.storage_create(bucket_name, target_username) bucket.upload_fileobj(BytesIO(b"dummydata1"), "dummykey") + bucket.upload_fileobj(BytesIO(b"dummydata1"), "dummykey1") + bucket.Object("dummykey1").delete() assert len(radosuser_admin.storage_list(show_full=True, name_filter="a")) == 2