mgr/dashboard: add deprecation warning on ns add nvme cli command

Fixes: https://tracker.ceph.com/issues/78228
Signed-off-by: Tomer Haskalovitch <tomer.haska@ibm.com>
This commit is contained in:
Tomer Haskalovitch 2026-07-14 01:34:38 +03:00
parent 55ccaac10e
commit c47b6f4e8d
3 changed files with 528 additions and 3 deletions

View File

@ -1263,7 +1263,8 @@ else:
"nvmeof namespace add",
model.NamespaceCreation,
alias="nvmeof ns add",
success_message_template="Adding namespace {nsid} to {nqn}: Successful"
success_message_template="Adding namespace {nsid} to {nqn}: Successful",
deprecated_params={"size": "--size is deprecated, please use --rbd-image-size"}
)
@EndpointDoc(
"Create a new NVMeoF namespace.",

View File

@ -349,7 +349,8 @@ class NvmeofCLICommand(DBCLICommand):
poll: bool = False,
success_message_template: Optional[str] = None,
success_message_map: Optional[Dict[str, Any]] = None,
success_message_fn: Optional[Callable[[Dict[str, Any]], str]] = None):
success_message_fn: Optional[Callable[[Dict[str, Any]], str]] = None,
deprecated_params: Optional[Dict[str, str]] = None):
super().__init__(prefix, perm, poll)
self._output_formatter = AnnotatedDataTextOutputFormatter()
self._model = model
@ -360,6 +361,7 @@ class NvmeofCLICommand(DBCLICommand):
self._success_message_map = success_message_map or {}
self._success_message_fn = success_message_fn
self._func_defaults: Dict[str, Any] = {}
self._deprecated_params: Dict[str, str] = deprecated_params or {}
def __call__(self, func):
resp = super().__call__(func)
@ -372,6 +374,7 @@ class NvmeofCLICommand(DBCLICommand):
success_message_template=self._success_message_template,
success_message_map=self._success_message_map,
success_message_fn=self._success_message_fn,
deprecated_params=self._deprecated_params,
)
self._alias_cmd(func)
self._alias_cmd._func_defaults = self._alias_cmd._compute_func_defaults()
@ -495,9 +498,16 @@ class NvmeofCLICommand(DBCLICommand):
mgr: Any,
cmd_dict: Dict[str, Any],
inbuf: Optional[str] = None) -> HandleCommandResult:
deprecated_warnings = ''
try:
out_format = cmd_dict.get('format')
args_map = self._args_map_from_argspec(cmd_dict, inbuf)
if out_format == 'plain' or not out_format:
for param, msg in self._deprecated_params.items():
if args_map.get(param) is not None:
deprecated_warnings += f"\nWarning: {msg}"
ret = super().call(mgr, cmd_dict, inbuf)
if ret is None:
ret = {}
@ -519,6 +529,7 @@ class NvmeofCLICommand(DBCLICommand):
wrn_msg = ret.get('error_message', '') if isinstance(ret, dict) else ''
if wrn_msg:
out += f"\nWarning: {wrn_msg}"
out += deprecated_warnings
elif out_format == 'json':
out = json.dumps(ret, indent=4)
@ -532,4 +543,4 @@ class NvmeofCLICommand(DBCLICommand):
# pylint: disable=broad-except
except Exception as e:
return HandleCommandResult(-errno.EINVAL, '', str(e))
return HandleCommandResult(-errno.EINVAL, '', str(e) + deprecated_warnings)

View File

@ -788,6 +788,519 @@ class TestNvmeofCLICommandSuccessMessage:
assert test_cmd not in NvmeofCLICommand.COMMANDS
class TestNvmeofCLICommandDeprecatedParams: # pylint: disable=too-many-public-methods
@staticmethod
def _make_cmd(test_cmd, deprecated_params=None, alias=None):
class Model(NamedTuple):
status: str
kwargs = {}
if deprecated_params is not None:
kwargs['deprecated_params'] = deprecated_params
if alias is not None:
kwargs['alias'] = alias
@NvmeofCLICommand(test_cmd, Model, **kwargs)
def fn(self, new_param: str, # pylint: disable=unused-variable,unused-argument
old_param: Optional[str] = None): # pylint: disable=unused-argument
return {'status': 'ok'}
return test_cmd
@staticmethod
def _cleanup(*cmds):
for cmd in cmds:
NvmeofCLICommand.COMMANDS.pop(cmd, None)
def test_deprecated_params_stored_on_instance(self):
test_cmd = "test deprecated store"
mapping = {"old_param": "old_param is deprecated, use new_param"}
self._make_cmd(test_cmd, deprecated_params=mapping)
try:
cmd = NvmeofCLICommand.COMMANDS[test_cmd]
assert cmd._deprecated_params == mapping # pylint: disable=protected-access
finally:
self._cleanup(test_cmd)
def test_no_deprecated_params_defaults_to_empty_dict(self):
test_cmd = "test deprecated empty"
self._make_cmd(test_cmd)
try:
cmd = NvmeofCLICommand.COMMANDS[test_cmd]
assert cmd._deprecated_params == {} # pylint: disable=protected-access
finally:
self._cleanup(test_cmd)
def test_none_deprecated_params_defaults_to_empty_dict(self):
test_cmd = "test deprecated none"
self._make_cmd(test_cmd, deprecated_params=None)
try:
cmd = NvmeofCLICommand.COMMANDS[test_cmd]
assert cmd._deprecated_params == {} # pylint: disable=protected-access
finally:
self._cleanup(test_cmd)
def test_alias_inherits_deprecated_params(self):
test_cmd = "test deprecated alias main"
test_alias = "test deprecated alias alias"
mapping = {"old_param": "old_param is deprecated, use new_param"}
self._make_cmd(test_cmd, deprecated_params=mapping, alias=test_alias)
try:
cmd = NvmeofCLICommand.COMMANDS[test_alias]
assert cmd._deprecated_params == mapping # pylint: disable=protected-access
finally:
self._cleanup(test_cmd, test_alias)
def test_alias_without_deprecated_params_has_empty_dict(self):
test_cmd = "test no deprecated alias main"
test_alias = "test no deprecated alias alias"
self._make_cmd(test_cmd, alias=test_alias)
try:
cmd = NvmeofCLICommand.COMMANDS[test_alias]
assert cmd._deprecated_params == {} # pylint: disable=protected-access
finally:
self._cleanup(test_cmd, test_alias)
def test_warning_emitted_when_deprecated_param_is_supplied(self):
test_cmd = "test deprecated warn supplied"
self._make_cmd(
test_cmd,
deprecated_params={"old_param": "--old-param is deprecated, please use --new-param"}
)
try:
result = NvmeofCLICommand.COMMANDS[test_cmd].call(
MagicMock(),
{"new_param": "foo", "old_param": "bar"}
)
assert result.retval == 0
assert "\nWarning: --old-param is deprecated, please use --new-param" in result.stdout
finally:
self._cleanup(test_cmd)
def test_no_warning_when_deprecated_param_is_absent(self):
test_cmd = "test deprecated warn absent"
self._make_cmd(
test_cmd,
deprecated_params={"old_param": "--old-param is deprecated, please use --new-param"}
)
try:
result = NvmeofCLICommand.COMMANDS[test_cmd].call(
MagicMock(),
{"new_param": "foo"}
)
assert result.retval == 0
assert "Warning" not in result.stdout
finally:
self._cleanup(test_cmd)
def test_no_warning_when_deprecated_params_is_empty(self):
test_cmd = "test no deprecated params"
self._make_cmd(test_cmd)
try:
result = NvmeofCLICommand.COMMANDS[test_cmd].call(
MagicMock(),
{"new_param": "foo", "old_param": "bar"}
)
assert result.retval == 0
assert "Warning" not in result.stdout
finally:
self._cleanup(test_cmd)
def test_multiple_deprecated_params_each_emit_warning(self):
test_cmd = "test deprecated multi warn"
class Model(NamedTuple):
status: str
@NvmeofCLICommand(
test_cmd,
Model,
deprecated_params={
"old_a": "--old-a is deprecated, use --new-a",
"old_b": "--old-b is deprecated, use --new-b",
}
) # pylint: disable=unused-variable
def fn(self, new_a: str, # pylint: disable=unused-argument
old_a: Optional[str] = None, # pylint: disable=unused-argument
old_b: Optional[str] = None): # pylint: disable=unused-argument
return {'status': 'ok'}
try:
result = NvmeofCLICommand.COMMANDS[test_cmd].call(
MagicMock(),
{"new_a": "x", "old_a": "y", "old_b": "z"}
)
assert result.retval == 0
assert "\nWarning: --old-a is deprecated, use --new-a" in result.stdout
assert "\nWarning: --old-b is deprecated, use --new-b" in result.stdout
finally:
self._cleanup(test_cmd)
def test_only_supplied_deprecated_param_warns(self):
test_cmd = "test deprecated partial warn"
class Model(NamedTuple):
status: str
@NvmeofCLICommand(
test_cmd,
Model,
deprecated_params={
"old_a": "--old-a is deprecated, use --new-a",
"old_b": "--old-b is deprecated, use --new-b",
}
) # pylint: disable=unused-variable
def fn(self, new_a: str, # pylint: disable=unused-argument
old_a: Optional[str] = None, # pylint: disable=unused-argument
old_b: Optional[str] = None): # pylint: disable=unused-argument
return {'status': 'ok'}
try:
result = NvmeofCLICommand.COMMANDS[test_cmd].call(
MagicMock(),
{"new_a": "x", "old_a": "y"}
)
assert result.retval == 0
assert "\nWarning: --old-a is deprecated, use --new-a" in result.stdout
assert "--old-b" not in result.stdout
finally:
self._cleanup(test_cmd)
def test_warning_not_emitted_for_json_format(self):
test_cmd = "test deprecated json no warn"
self._make_cmd(
test_cmd,
deprecated_params={"old_param": "--old-param is deprecated, please use --new-param"}
)
try:
result = NvmeofCLICommand.COMMANDS[test_cmd].call(
MagicMock(),
{"format": "json", "new_param": "foo", "old_param": "bar"}
)
assert result.retval == 0
assert "Warning" not in result.stdout
parsed = json.loads(result.stdout)
assert "Warning" not in str(parsed)
finally:
self._cleanup(test_cmd)
def test_warning_not_emitted_for_yaml_format(self):
test_cmd = "test deprecated yaml no warn"
self._make_cmd(
test_cmd,
deprecated_params={"old_param": "--old-param is deprecated, please use --new-param"}
)
try:
result = NvmeofCLICommand.COMMANDS[test_cmd].call(
MagicMock(),
{"format": "yaml", "new_param": "foo", "old_param": "bar"}
)
assert result.retval == 0
assert "Warning" not in result.stdout
finally:
self._cleanup(test_cmd)
def test_warning_emitted_for_default_format(self):
"""Default format (no 'format' key) also emits warnings."""
test_cmd = "test deprecated default format warn"
self._make_cmd(
test_cmd,
deprecated_params={"old_param": "--old-param is deprecated, please use --new-param"}
)
try:
result = NvmeofCLICommand.COMMANDS[test_cmd].call(
MagicMock(),
{"new_param": "foo", "old_param": "bar"}
)
assert result.retval == 0
assert "\nWarning: --old-param is deprecated, please use --new-param" in result.stdout
finally:
self._cleanup(test_cmd)
def test_warning_emitted_for_explicit_plain_format(self):
test_cmd = "test deprecated plain format warn"
self._make_cmd(
test_cmd,
deprecated_params={"old_param": "--old-param is deprecated, please use --new-param"}
)
try:
result = NvmeofCLICommand.COMMANDS[test_cmd].call(
MagicMock(),
{"format": "plain", "new_param": "foo", "old_param": "bar"}
)
assert result.retval == 0
assert "\nWarning: --old-param is deprecated, please use --new-param" in result.stdout
finally:
self._cleanup(test_cmd)
def test_deprecated_warning_and_error_message_warning_both_appear(self):
test_cmd = "test deprecated coexist error message"
class Model(NamedTuple):
status: str
error_message: str
@NvmeofCLICommand(
test_cmd,
Model,
deprecated_params={"old_param": "--old-param is deprecated, please use --new-param"}
) # pylint: disable=unused-variable
def fn(self, new_param: str, # pylint: disable=unused-argument
old_param: Optional[str] = None): # pylint: disable=unused-argument
return {'status': 'ok', 'error_message': 'something to note from gRPC'}
try:
result = NvmeofCLICommand.COMMANDS[test_cmd].call(
MagicMock(),
{"new_param": "foo", "old_param": "bar"}
)
assert result.retval == 0
assert "\nWarning: something to note from gRPC" in result.stdout
assert "\nWarning: --old-param is deprecated, please use --new-param" in result.stdout
finally:
self._cleanup(test_cmd)
def test_error_message_warning_without_deprecated_param(self):
test_cmd = "test error message no deprecated"
class Model(NamedTuple):
status: str
error_message: str
@NvmeofCLICommand(test_cmd, Model)
def fn(self, new_param: str): # pylint: disable=unused-variable,unused-argument
return {'status': 'ok', 'error_message': 'note from gRPC'}
try:
result = NvmeofCLICommand.COMMANDS[test_cmd].call(
MagicMock(),
{"new_param": "foo"}
)
assert result.retval == 0
assert "\nWarning: note from gRPC" in result.stdout
finally:
self._cleanup(test_cmd)
def test_alias_emits_same_warning_as_main_command(self):
test_cmd = "test deprecated alias warn main"
test_alias = "test deprecated alias warn alias"
self._make_cmd(
test_cmd,
deprecated_params={"old_param": "--old-param is deprecated, please use --new-param"},
alias=test_alias
)
try:
for cmd_key in (test_cmd, test_alias):
result = NvmeofCLICommand.COMMANDS[cmd_key].call(
MagicMock(),
{"new_param": "foo", "old_param": "bar"}
)
assert result.retval == 0, f"failed for {cmd_key}"
assert "\nWarning: --old-param is deprecated, please use --new-param" \
in result.stdout, f"warning missing for {cmd_key}"
finally:
self._cleanup(test_cmd, test_alias)
def test_alias_also_suppresses_warning_when_param_absent(self):
test_cmd = "test deprecated alias no warn main"
test_alias = "test deprecated alias no warn alias"
self._make_cmd(
test_cmd,
deprecated_params={"old_param": "--old-param is deprecated, please use --new-param"},
alias=test_alias
)
try:
result = NvmeofCLICommand.COMMANDS[test_alias].call(
MagicMock(),
{"new_param": "foo"}
)
assert result.retval == 0
assert "Warning" not in result.stdout
finally:
self._cleanup(test_cmd, test_alias)
def test_warning_emitted_on_failure_when_deprecated_param_supplied(self):
test_cmd = "test deprecated warn on failure"
class Model(NamedTuple):
status: str
@NvmeofCLICommand(
test_cmd,
Model,
deprecated_params={"old_param": "--old-param is deprecated, please use --new-param"}
) # pylint: disable=unused-variable
def fn(self, new_param: str, # pylint: disable=unused-argument
old_param: Optional[str] = None): # pylint: disable=unused-argument
raise DashboardException(msg="something went wrong", component="nvmeof",
http_status_code=500)
try:
result = NvmeofCLICommand.COMMANDS[test_cmd].call(
MagicMock(),
{"new_param": "foo", "old_param": "bar"}
)
assert result.retval == -errno.EINVAL
assert result.stdout == ''
assert "\nWarning: --old-param is deprecated, please use --new-param" in result.stderr
finally:
self._cleanup(test_cmd)
def test_no_warning_on_failure_when_deprecated_param_absent(self):
test_cmd = "test deprecated no warn on failure absent"
class Model(NamedTuple):
status: str
@NvmeofCLICommand(
test_cmd,
Model,
deprecated_params={"old_param": "--old-param is deprecated, please use --new-param"}
) # pylint: disable=unused-variable
def fn(self, new_param: str, # pylint: disable=unused-argument
old_param: Optional[str] = None): # pylint: disable=unused-argument
raise DashboardException(msg="something went wrong", component="nvmeof",
http_status_code=500)
try:
result = NvmeofCLICommand.COMMANDS[test_cmd].call(
MagicMock(),
{"new_param": "foo"}
)
assert result.retval == -errno.EINVAL
assert "Warning" not in result.stderr
finally:
self._cleanup(test_cmd)
def test_warning_in_stdout_on_success_and_stderr_on_failure(self):
test_cmd_ok = "test deprecated path ok"
test_cmd_fail = "test deprecated path fail"
class Model(NamedTuple):
status: str
@NvmeofCLICommand(
test_cmd_ok,
Model,
deprecated_params={"old_param": "--old-param is deprecated, please use --new-param"}
) # pylint: disable=unused-variable
def fn_ok(self, new_param: str, # pylint: disable=unused-argument
old_param: Optional[str] = None): # pylint: disable=unused-argument
return {'status': 'ok'}
@NvmeofCLICommand(
test_cmd_fail,
Model,
deprecated_params={"old_param": "--old-param is deprecated, please use --new-param"}
) # pylint: disable=unused-variable
def fn_fail(self, new_param: str, # pylint: disable=unused-argument
old_param: Optional[str] = None): # pylint: disable=unused-argument
raise DashboardException(msg="boom", component="nvmeof", http_status_code=500)
try:
ok_result = NvmeofCLICommand.COMMANDS[test_cmd_ok].call(
MagicMock(),
{"new_param": "foo", "old_param": "bar"}
)
assert ok_result.retval == 0
assert "\nWarning: --old-param is deprecated, please use --new-param" \
in ok_result.stdout
assert ok_result.stderr == ''
fail_result = NvmeofCLICommand.COMMANDS[test_cmd_fail].call(
MagicMock(),
{"new_param": "foo", "old_param": "bar"}
)
assert fail_result.retval == -errno.EINVAL
assert fail_result.stdout == ''
assert "\nWarning: --old-param is deprecated, please use --new-param" \
in fail_result.stderr
finally:
self._cleanup(test_cmd_ok, test_cmd_fail)
def test_deprecated_warning_appended_after_success_message(self):
test_cmd = "test deprecated with success msg"
class Model(NamedTuple):
status: str
@NvmeofCLICommand(
test_cmd,
Model,
success_message_template="Done: {new_param}",
deprecated_params={"old_param": "--old-param is deprecated, please use --new-param"}
) # pylint: disable=unused-variable
def fn(self, new_param: str, # pylint: disable=unused-argument
old_param: Optional[str] = None): # pylint: disable=unused-argument
return {'status': 'ok'}
try:
result = NvmeofCLICommand.COMMANDS[test_cmd].call(
MagicMock(),
{"new_param": "foo", "old_param": "bar"}
)
assert result.retval == 0
assert result.stdout.startswith("Done: foo")
assert "\nWarning: --old-param is deprecated, please use --new-param" in result.stdout
finally:
self._cleanup(test_cmd)
def test_no_warning_appended_to_success_message_when_param_absent(self):
test_cmd = "test deprecated with success msg no warn"
class Model(NamedTuple):
status: str
@NvmeofCLICommand(
test_cmd,
Model,
success_message_template="Done: {new_param}",
deprecated_params={"old_param": "--old-param is deprecated, please use --new-param"}
) # pylint: disable=unused-variable
def fn(self, new_param: str, # pylint: disable=unused-argument
old_param: Optional[str] = None): # pylint: disable=unused-argument
return {'status': 'ok'}
try:
result = NvmeofCLICommand.COMMANDS[test_cmd].call(
MagicMock(),
{"new_param": "foo"}
)
assert result.retval == 0
assert result.stdout == "Done: foo"
finally:
self._cleanup(test_cmd)
def test_malformed_arg_in_args_map_returns_handle_command_result_not_uncaught_exception(self):
test_cmd = "test argspec throws inside try"
class Model(NamedTuple):
status: str
@NvmeofCLICommand(
test_cmd,
Model,
deprecated_params={"old_param": "--old-param is deprecated, please use --new-param"}
) # pylint: disable=unused-variable
def fn(self, count: int, # pylint: disable=unused-argument
old_param: Optional[str] = None): # pylint: disable=unused-argument
return {'status': 'ok'}
try:
# 'count' is typed as int but we pass a string value to it so
# CephArgtype.cast_to raises ValueError inside _args_map_from_argspec
result = NvmeofCLICommand.COMMANDS[test_cmd].call(
MagicMock(),
{"count": "not-a-number", "old_param": "bar"}
)
assert isinstance(result, HandleCommandResult)
assert result.retval == -errno.EINVAL
assert result.stdout == ''
assert result.stderr != ''
finally:
self._cleanup(test_cmd)
class TestNVMeoFConfCLI(unittest.TestCase, CLICommandTestMixin):
def setUp(self):
self.mock_kv_store()