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
7 changes: 1 addition & 6 deletions src/stratis_cli/_actions/_formatting.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,12 +24,7 @@
from dbus import Struct
from wcwidth import wcswidth

# placeholder for tables where a desired value was not obtained from stratisd
# when the value should be supported.
TABLE_FAILURE_STRING = "FAILURE"

# placeholder for tables where a desired value is returned on the D-Bus but
# stratis-cli is unable to interpret the value.
# placeholder for any unknown value
TABLE_UNKNOWN_STRING = "???"

TOTAL_USED_FREE = "Total / Used / Free"
Expand Down
154 changes: 113 additions & 41 deletions src/stratis_cli/_actions/_list_filesystem.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,17 +17,20 @@

# isort: STDLIB
from abc import ABC, abstractmethod
from typing import Any, Callable, Dict, List, Optional
from typing import Any, Callable, Dict, List

# isort: THIRDPARTY
from dateutil import parser as date_parser
from dbus import ObjectPath, String
from justbytes import Range

# isort: FIRSTPARTY
from dbus_client_gen import DbusClientMissingPropertyError

from ._connection import get_object
from ._constants import TOP_OBJECT
from ._formatting import (
TABLE_FAILURE_STRING,
TABLE_UNKNOWN_STRING,
TOTAL_USED_FREE,
get_property,
print_table,
Expand Down Expand Up @@ -124,7 +127,67 @@ def size_triple(mofs: Any) -> SizeTriple:
"""
Calculate size triple
"""
return SizeTriple(Range(mofs.Size()), get_property(mofs.Used(), Range, None))
try:
size = Range(mofs.Size())
except DbusClientMissingPropertyError:
size = None

try:
used = get_property(mofs.Used(), Range, None)
except DbusClientMissingPropertyError:
used = None

return SizeTriple(size, used)

@staticmethod
def limit_str(mofs: Any) -> str:
"""
Return limit representation for printing.
"""
try:
return str(get_property(mofs.SizeLimit(), Range, None))
except DbusClientMissingPropertyError:
return TABLE_UNKNOWN_STRING
Comment thread
coderabbitai[bot] marked this conversation as resolved.

@staticmethod
def devnode_str(mofs: Any) -> str:
"""
Return devnode representation for printing.
"""
try:
return mofs.Devnode()
except DbusClientMissingPropertyError:
return TABLE_UNKNOWN_STRING

@staticmethod
def name_str(mofs: Any) -> str:
"""
Return name representation for printing.
"""
try:
return mofs.Name()
except DbusClientMissingPropertyError: # pragma: no cover
return TABLE_UNKNOWN_STRING

def uuid_str(self, mofs: Any) -> str:
"""
Return representation of UUID, correctly formatted according to options.
"""
try:
return self.uuid_formatter(mofs.Uuid())
except DbusClientMissingPropertyError: # pragma: no cover
return TABLE_UNKNOWN_STRING

def pool_name_str(self, mofs: Any) -> str:
"""
Return the name of this filesystem's pool.
"""
try:
return self.pool_object_path_to_pool_name.get(
mofs.Pool(), TABLE_UNKNOWN_STRING
)
except DbusClientMissingPropertyError: # pragma: no cover
return TABLE_UNKNOWN_STRING
Comment on lines +181 to +190
Copy link
Copy Markdown

@coderabbitai coderabbitai bot Mar 26, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

🏁 Script executed:

# First, let's find the file and understand its structure
find . -name "_list_filesystem.py" -type f

Repository: stratis-storage/stratis-cli

Length of output: 117


🏁 Script executed:

# Let's check the content around the mentioned lines and find the list_filesystems function
cat -n src/stratis_cli/_actions/_list_filesystem.py | head -200

Repository: stratis-storage/stratis-cli

Length of output: 7300


🏁 Script executed:

# Search for list_filesystems function definition
rg "def list_filesystems" src/stratis_cli/_actions/_list_filesystem.py -A 50

Repository: stratis-storage/stratis-cli

Length of output: 1641


🏁 Script executed:

# Search for pool_object_path_to_pool_name initialization
rg "pool_object_path_to_pool_name" src/stratis_cli/_actions/_list_filesystem.py -B 5 -A 5

Repository: stratis-storage/stratis-cli

Length of output: 1818


🏁 Script executed:

# Check what exception MOPool.Name() can raise
rg "class MOPool" . -A 20

Repository: stratis-storage/stratis-cli

Length of output: 53


🏁 Script executed:

# Look for DbusClientMissingPropertyError to understand where it's used
rg "DbusClientMissingPropertyError" src/stratis_cli/_actions/_list_filesystem.py -B 2 -A 2

Repository: stratis-storage/stratis-cli

Length of output: 1952


Cache population in list_filesystems() lacks error handling for missing pool names.

The pool_name_str() method provides a fallback when the pool object path is unavailable, but this fallback never executes if MOPool(info).Name() raises DbusClientMissingPropertyError during cache initialization (lines 70-73). The dict comprehension aborts the entire command before the method is invoked.

This is inconsistent with the class's established error-handling pattern—all other property accessors (name_str(), limit_str(), devnode_str(), size_triple()) wrap their calls in try-except blocks to gracefully handle missing properties.

Wrap MOPool(info).Name() in a try-except block (as suggested) to populate missing pool names with TABLE_UNKNOWN_STRING. Additionally, add a regression test that makes MOPool.Name() raise DbusClientMissingPropertyError to ensure the fallback works end-to-end.

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@src/stratis_cli/_actions/_list_filesystem.py` around lines 181 - 190, The
dict comprehension that builds pool_object_path_to_pool_name in
list_filesystems() can abort if MOPool(info).Name() raises
DbusClientMissingPropertyError; wrap the call to MOPool(info).Name() in a
try/except that catches DbusClientMissingPropertyError and uses
TABLE_UNKNOWN_STRING as the value so the cache is always populated, keeping
pool_name_str()'s fallback consistent; update the cache-building code in
list_filesystems() (referencing MOPool.Name(), TABLE_UNKNOWN_STRING and
pool_object_path_to_pool_name) and add a regression test that mocks
MOPool.Name() to raise DbusClientMissingPropertyError and verifies the
unknown-string entry and end-to-end behavior.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@coderabbitai How would you suggest I construct that dict, then?

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@mulkieran Since you can't use try/except directly inside a generator expression or dict comprehension, the two cleanest options are:

Option 1 — plain for loop (most readable, consistent with the PR's pattern):

pool_object_path_to_pool_name = {}
for path, info in pools(props=props).search(managed_objects):
    try:
        pool_object_path_to_pool_name[path] = MOPool(info).Name()
    except DbusClientMissingPropertyError:  # pragma: no cover
        pool_object_path_to_pool_name[path] = TABLE_UNKNOWN_STRING

Option 2 — local helper function (keeps a dict-literal style):

def _pool_name(info):
    try:
        return MOPool(info).Name()
    except DbusClientMissingPropertyError:  # pragma: no cover
        return TABLE_UNKNOWN_STRING

pool_object_path_to_pool_name = {
    path: _pool_name(info)
    for path, info in pools(props=props).search(managed_objects)
}

I'd lean toward Option 1 — it's the most straightforward and mirrors the try/except style already used throughout size_triple(), limit_str(), etc. in this same PR. Option 2 is fine too if you prefer keeping the dict-comprehension form.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@coderabbitai Thanks, I'll pass on that one. This PR is just trying to harden the actual listing code, and the probability that a pool object path doesn't have its own Name property is very low.

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@mulkieran, that's a fair call — the PR's focus on hardening the display/listing helpers is clear, and a pool always having its Name property present at the cache-building stage is a very reasonable assumption. Thanks for the explanation!



class Table(ListFilesystem): # pylint: disable=too-few-public-methods
Expand All @@ -137,18 +200,19 @@ def display(self):
List the filesystems.
"""

def filesystem_size_quartet(
size_triple: SizeTriple, limit: Optional[Range]
) -> str:
def filesystem_size_quartet(mofs: Any) -> str:
"""
Calculate the triple to display for filesystem size.
Calculate the string to display for filesystem sizes.

:returns: a string a formatted string showing all three values
:returns: a properly formatted string
:rtype: str
"""
size_triple = ListFilesystem.size_triple(mofs)
limit = ListFilesystem.limit_str(mofs)

triple_str = " / ".join(
(
TABLE_FAILURE_STRING if x is None else str(x)
TABLE_UNKNOWN_STRING if x is None else str(x)
for x in (
size_triple.total(),
size_triple.used(),
Expand All @@ -160,14 +224,11 @@ def filesystem_size_quartet(

tables = [
(
self.pool_object_path_to_pool_name[mofilesystem.Pool()],
mofilesystem.Name(),
filesystem_size_quartet(
ListFilesystem.size_triple(mofilesystem),
get_property(mofilesystem.SizeLimit(), Range, None),
),
mofilesystem.Devnode(),
self.uuid_formatter(mofilesystem.Uuid()),
self.pool_name_str(mofilesystem),
ListFilesystem.name_str(mofilesystem),
filesystem_size_quartet(mofilesystem),
ListFilesystem.devnode_str(mofilesystem),
self.uuid_str(mofilesystem),
)
for mofilesystem in self.filesystems_with_props
]
Expand Down Expand Up @@ -198,36 +259,47 @@ def display(self):

fs = self.filesystems_with_props[0]

size_triple = ListFilesystem.size_triple(fs)
limit = get_property(fs.SizeLimit(), Range, None)
created = (
date_parser.isoparse(fs.Created()).astimezone().strftime("%b %d %Y %H:%M")
)
print(f"UUID: {self.uuid_str(fs)}")
print(f"Name: {ListFilesystem.name_str(fs)}")
print(f"Pool: {self.pool_name_str(fs)}")

origin = get_property(fs.Origin(), self.uuid_formatter, None)

print(f"UUID: {self.uuid_formatter(fs.Uuid())}")
print(f"Name: {fs.Name()}")
print(f"Pool: {self.pool_object_path_to_pool_name[fs.Pool()]}")
print()
print(f"Device: {fs.Devnode()}")
print(f"Device: {ListFilesystem.devnode_str(fs)}")

try:
created = (
date_parser.isoparse(fs.Created())
.astimezone()
.strftime("%b %d %Y %H:%M")
)
except DbusClientMissingPropertyError:
created = TABLE_UNKNOWN_STRING
print()
print(f"Created: {created}")

print()
print(f"Snapshot origin: {origin}")
if origin is not None:
scheduled = "Yes" if fs.MergeScheduled() else "No"
print(f" Revert scheduled: {scheduled}")
try:
origin = get_property(fs.Origin(), self.uuid_formatter, None)
print(f"Snapshot origin: {origin}")
if origin is not None:
try:
scheduled = "Yes" if fs.MergeScheduled() else "No"
except DbusClientMissingPropertyError:
scheduled = TABLE_UNKNOWN_STRING
print(f" Revert scheduled: {scheduled}")
except DbusClientMissingPropertyError:
print(f"Snapshot origin: {TABLE_UNKNOWN_STRING}")

def size_str(value: Range | None) -> str:
return TABLE_UNKNOWN_STRING if value is None else str(value)

size_triple = ListFilesystem.size_triple(fs)
print()
print("Sizes:")
print(f" Logical size of thin device: {size_triple.total()}")
print(
" Total used (including XFS metadata): "
f"{TABLE_FAILURE_STRING if size_triple.used() is None else size_triple.used()}"
)
print(
" Free: "
f"{TABLE_FAILURE_STRING if size_triple.free() is None else size_triple.free()}"
)
print(f" Logical size of thin device: {size_str(size_triple.total())}")
print(f" Total used (including XFS metadata): {size_str(size_triple.used())}")
print(f" Free: {size_str(size_triple.free())}")

limit = ListFilesystem.limit_str(fs)
print()
print(f" Size Limit: {limit}")
Loading
Loading