Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 17 additions & 18 deletions src/stratis_cli/_actions/_list_pool.py
Original file line number Diff line number Diff line change
Expand Up @@ -140,17 +140,20 @@ def __init__(
"""
from ._data import MODev # noqa: PLC0415

(increased, decreased) = (set(), set())
(increased, decreased, unknown) = (set(), set(), set())
for _, info in devs_to_search:
modev = MODev(info)
size = Range(modev.TotalPhysicalSize())
observed_size = get_property(modev.NewPhysicalSize(), Range, size)
if observed_size > size: # pragma: no cover
increased.add(modev.Pool())
if observed_size < size: # pragma: no cover
decreased.add(modev.Pool())
try:
size = Range(modev.TotalPhysicalSize())
observed_size = get_property(modev.NewPhysicalSize(), Range, size)
if observed_size > size: # pragma: no cover
increased.add(modev.Pool())
elif observed_size < size: # pragma: no cover
decreased.add(modev.Pool())
except DbusClientMissingPropertyError:
unknown.add(modev.Pool())
Comment thread
coderabbitai[bot] marked this conversation as resolved.

(self.increased, self.decreased) = (increased, decreased)
(self.increased, self.decreased, self.unknown) = (increased, decreased, unknown)

def alert_codes(self, pool_object_path: str) -> list[PoolDeviceSizeChangeAlert]:
"""
Expand All @@ -159,18 +162,14 @@ def alert_codes(self, pool_object_path: str) -> list[PoolDeviceSizeChangeAlert]:
:param pool_object_path: the pool object path
:returns: the codes
"""
if (
pool_object_path in self.increased and pool_object_path in self.decreased
): # pragma: no cover
return [
PoolDeviceSizeChangeAlert.DEVICE_SIZE_INCREASED,
PoolDeviceSizeChangeAlert.DEVICE_SIZE_DECREASED,
]
alerts = []
if pool_object_path in self.increased: # pragma: no cover
return [PoolDeviceSizeChangeAlert.DEVICE_SIZE_INCREASED]
alerts.append(PoolDeviceSizeChangeAlert.DEVICE_SIZE_INCREASED)
if pool_object_path in self.decreased: # pragma: no cover
return [PoolDeviceSizeChangeAlert.DEVICE_SIZE_DECREASED]
return []
alerts.append(PoolDeviceSizeChangeAlert.DEVICE_SIZE_DECREASED)
if pool_object_path in self.unknown:
alerts.append(PoolDeviceSizeChangeAlert.DEVICE_SIZE_CHANGE_UNKNOWN)
return alerts


def list_pools(
Expand Down
15 changes: 14 additions & 1 deletion src/stratis_cli/_alerts.py
Original file line number Diff line number Diff line change
Expand Up @@ -118,12 +118,16 @@ class PoolDeviceSizeChangeAlert(IntEnum):

DEVICE_SIZE_INCREASED = 1
DEVICE_SIZE_DECREASED = 2
DEVICE_SIZE_CHANGE_UNKNOWN = 3

def __str__(self) -> str:
if self is PoolDeviceSizeChangeAlert.DEVICE_SIZE_INCREASED:
return f"{Level.INFO}DS{str(self.value).zfill(3)}"

if self is PoolDeviceSizeChangeAlert.DEVICE_SIZE_DECREASED:
if self in (
PoolDeviceSizeChangeAlert.DEVICE_SIZE_DECREASED,
PoolDeviceSizeChangeAlert.DEVICE_SIZE_CHANGE_UNKNOWN,
):
return f"{Level.WARNING}DS{str(self.value).zfill(3)}"

assert_never(self) # pragma: no cover
Expand All @@ -144,6 +148,12 @@ def explain(self) -> str:
"decreased in size."
)

if self is PoolDeviceSizeChangeAlert.DEVICE_SIZE_CHANGE_UNKNOWN:
return (
"At least one device belonging to this pool does not have "
"complete size information."
)

assert_never(self) # pragma: no cover

def summarize(self) -> str:
Expand All @@ -156,6 +166,9 @@ def summarize(self) -> str:
if self is PoolDeviceSizeChangeAlert.DEVICE_SIZE_DECREASED:
return "A device in this pool has decreased in size."

if self is PoolDeviceSizeChangeAlert.DEVICE_SIZE_CHANGE_UNKNOWN:
return "A device in this pool has an unknown size."

assert_never(self) # pragma: no cover


Expand Down
28 changes: 28 additions & 0 deletions tests/integration/pool/test_list.py
Original file line number Diff line number Diff line change
Expand Up @@ -364,6 +364,34 @@ def test_dropping_properties(self):
with self.subTest(property_name=property_name, options=options):
TEST_RUNNER(self._MENU + options)

def test_dropping_dev_properties(self):
"""
Verify no exception thrown if any of device properties are dropped.
"""
import stratis_cli # noqa: PLC0415
from stratis_cli import _actions # noqa: PLC0415
from stratis_cli._actions._introspect import SPECS # noqa: PLC0415

dev_spec = SPECS[_actions._constants.BLOCKDEV_INTERFACE]
spec = ElementTree.fromstring(dev_spec)

for property_name in [
prop.attrib["name"] for prop in spec.findall("./property")
]:
with patch.object(
stratis_cli._actions._data.MODev, # pyright: ignore
property_name,
autospec=True,
side_effect=DbusClientMissingPropertyError(
"oops",
stratis_cli._actions._constants.BLOCKDEV_INTERFACE, # pyright: ignore
property_name,
),
):
for options in [[], [f"--name={self._POOLNAME}"]]:
with self.subTest(property_name=property_name, options=options):
TEST_RUNNER(self._MENU + options)


class List7TestCase(SimTestCase):
"""
Expand Down
Loading