ret_addr = f"[{addr}]"
return ret_addr
- def build_listener_del_success_message(args: Dict[str, Any]) -> str:
+ def build_listener_del_success_message(args: Dict[str, Any], _) -> str:
traddr = args.get('traddr')
trsvcid = args.get('trsvcid')
subsystem = args.get('nqn')
f"{host_msg}: Successful"
)
- def build_ns_change_visibility_success_message(args: Dict[str, Any]) -> str:
+ def build_ns_change_visibility_success_message(args: Dict[str, Any], _) -> str:
nsid = args.get('nsid')
subsystem = args.get('nqn')
auto_visible_val = args.get('auto_visible')
f"to {vis_text}: Successful"
)
- def build_ns_set_auto_resize_success_message(args: Dict[str, Any]) -> str:
+ def build_ns_set_auto_resize_success_message(args: Dict[str, Any], _) -> str:
nsid = args.get('nsid')
subsystem = args.get('nqn')
auto_resize_enabled = args.get('auto_resize_enabled')
f"{subsystem} to {auto_resize_text}: Successful"
)
- def build_ns_set_rbd_trash_image_success_message(args: Dict[str, Any]) -> str:
+ def build_ns_set_rbd_trash_image_success_message(args: Dict[str, Any], _) -> str:
nsid = args.get('nsid')
subsystem = args.get('nqn')
rbd_trash_image_on_delete = args.get('rbd_trash_image_on_delete')
f"{subsystem} to {trash_text}: Successful"
)
- def build_host_add_success_message(args: Dict[str, Any]) -> str:
+ def build_host_add_success_message(args: Dict[str, Any], _) -> str:
subsystem = args.get('nqn')
host_nqn_list: List[str] = args.get('host_nqn') or []
messages.append(f"Adding host {one_host_nqn} to {subsystem}: Successful")
return "\n".join(messages)
- def build_host_del_success_message(args: Dict[str, Any]) -> str:
+ def build_host_del_success_message(args: Dict[str, Any], _) -> str:
subsystem = args.get('nqn')
host_nqn_list: List[str] = args.get('host_nqn') or []
return str(value)
- def _format_success_message_from_args(self, args_map: Dict[str, Any]) -> Optional[str]:
+ def _format_success_message_from_args(self,
+ args_map: Dict[str, Any],
+ response: Dict[str, Any]) -> Optional[str]:
if not self._success_message_template and not self._success_message_fn:
return None
if self._success_message_fn:
try:
- msg = self._success_message_fn(args_map)
+ msg = self._success_message_fn(args_map, response)
if msg:
return msg
except Exception:
if self._success_message_template:
try:
- str_map = {k: self._stringify(v) for k, v in args_map.items()}
+ fields_dict = {**args_map, **response}
+ str_map = {k: self._stringify(v) for k, v in fields_dict.items()}
return self._success_message_template.format(**str_map)
except Exception:
logger.warning("Success message template failed for %s", self.prefix, exc_info=True)
if out_format == 'plain' or not out_format:
message: Optional[str] = None
try:
- message = self._format_success_message_from_args(args_map)
+ message = self._format_success_message_from_args(args_map, ret)
except Exception:
logger.warning("Formatting of success message failed for %s",
self.prefix, exc_info=True)
def set_log_level(self, log_level: str, gw_group: Optional[str] = None, traddr: Optional[str] = None): # noqa
return {"status": 0}
- result_default = NvmeofCLICommand.COMMANDS[test_cmd].call(MagicMock(),
- {"log_level": "info"})
+ result_default = NvmeofCLICommand.COMMANDS[test_cmd].call(
+ MagicMock(),
+ {"log_level": "info"}
+ )
assert isinstance(result_default, HandleCommandResult)
assert result_default.retval == 0
assert result_default.stdout == "set log level to info"
def set_log_level(self, a: str): # noqa
return {"a": "b"}
- result_plain = NvmeofCLICommand.COMMANDS[test_cmd].call(MagicMock(), {"format": "plain"})
+ result_plain = NvmeofCLICommand.COMMANDS[test_cmd].call(
+ MagicMock(),
+ {"format": "plain"}
+ )
assert isinstance(result_plain, HandleCommandResult)
assert result_plain.retval == 0
assert result_plain.stdout == (
@NvmeofCLICommand(
test_cmd,
Model,
- success_message_fn=lambda args: (
+ success_message_fn=lambda args, response: (
f"set log level to {args.get('log_level', '')}"
+ (" for all hosts" if args.get('all_hosts') else "")
)
del NvmeofCLICommand.COMMANDS[test_cmd]
assert test_cmd not in NvmeofCLICommand.COMMANDS
-
-
+
def test_template_formats_int_and_list_without_failure(self):
class Model(NamedTuple):
status: str
success_message_template="ns {nsid} hosts {host_nqn}"
)
def fn(self, nsid: int, host_nqn: list[str]): # noqa
- return {"status": "ok"}
+ return {"status": 1}
res = NvmeofCLICommand.COMMANDS["nvmeof mixed params"].call(
MagicMock(),
)
assert res.retval == 0
assert res.stdout == "ns 42 hosts a,b"
-
+
+ del NvmeofCLICommand.COMMANDS["nvmeof mixed params"]
+ assert "nvmeof mixed params" not in NvmeofCLICommand.COMMANDS
+
def test_success_message_uses_default_when_cli_omits_param(self):
class Model(NamedTuple):
status: str
- def create(mgr,nqn: str,host_name: str, traddr: str, trsvcid: int = 4420, adrfam: int = 0, gw_group: Optional[str] = None):
- return Model(status="ok")
- cmd = NvmeofCLICommand("nvmeof listener add", model=Model, success_message_template="Adding {nqn} listener at {traddr}:{trsvcid}: Successful")
- # Simulate a CLI invocation without trsvcid (and without format flag).
- # CLICommand.call will use _collect_args_by_argspec; for this test
- # we pass the minimal dict that would be parsed from CLI switches.
+ def create(mgr, nqn: str, host_name: str, traddr: str,
+ trsvcid: int = 4420, adrfam: int = 0, gw_group: Optional[str] = None):
+ return dict(status=1)
+
+ cmd = NvmeofCLICommand(
+ "nvmeof listener add",
+ model=Model,
+ success_message_template="Adding {nqn} listener at {traddr}:{trsvcid}: Successful"
+ )
+ cmd(create)
+
cmd_dict = {
"nqn": "nqn.2014-08.org.nvmexpress:uuid:1234",
"host_name": "nvme-host-1",
"traddr": "10.0.0.5",
- # 'trsvcid' intentionally omitted
- # 'adrfam' intentionally omitted
+ # 'trsvcid' omitted
+ # 'adrfam' omitted
}
- # Run command
result = cmd.call(mgr=None, cmd_dict=cmd_dict, inbuf=None)
-
- assert result.rval == 0
- assert result.err == ""
- assert result.out == (
+ assert result.retval == 0
+ assert result.stderr == ""
+ assert result.stdout == (
"Adding nqn.2014-08.org.nvmexpress:uuid:1234 listener at 10.0.0.5:4420: Successful"
)
-
def test_success_message_cli_value_overrides_default(self):
class Model(NamedTuple):
status: str
- cmd = Model()
+
+ def create(mgr, nqn: str, host_name: str, traddr: str,
+ trsvcid: int = 4420, adrfam: int = 0, gw_group: Optional[str] = None):
+ return dict(status=1)
+
+ cmd = NvmeofCLICommand(
+ "nvmeof listener add",
+ model=Model,
+ success_message_template="Adding {nqn} listener at {traddr}:{trsvcid}: Successful"
+ )
+ cmd(create)
cmd_dict = {
"nqn": "nqn.2014-08.org.nvmexpress:uuid:abcd",
}
result = cmd.call(mgr=None, cmd_dict=cmd_dict, inbuf=None)
-
- assert result.rval == 0
- assert result.err == ""
- assert result.out == (
+ assert result.retval == 0
+ assert result.stderr == ""
+ assert result.stdout == (
"Adding nqn.2014-08.org.nvmexpress:uuid:abcd listener at 192.168.1.10:8009: Successful"
)
-
def test_defaults_allow_none_and_template_does_not_crash(self):
class Model(NamedTuple):
status: str
-
- # Define a handler with a None default referenced in the template
+
def create_with_none(
mgr,
nqn: str,
trsvcid: int = 4420,
gw_group: Optional[str] = None, # None default intentionally used
):
- return Model(status="ok")
+ return dict(status=1)
cmd = NvmeofCLICommand(
"nvmeof listener add",
}
result = cmd.call(mgr=None, cmd_dict=cmd_dict, inbuf=None)
-
- assert result.rval == 0
- assert result.err == ""
- # gw_group should stringify to 'None' under current _stringify implementation
- assert result.out == (
+ assert result.retval == 0
+ assert result.stderr == ""
+ assert result.stdout == (
"Adding nqn.none.test listener at 127.0.0.1:4420 gw=None: Successful"
)
+ def test_template_can_use_response_fields(self):
+ test_cmd = "nvmeof show op status"
+
+ class Model(NamedTuple):
+ status: str
+ message: str
+
+ @NvmeofCLICommand(
+ test_cmd,
+ Model,
+ success_message_template="operation {op} finished with status {message}"
+ )
+ def op(self, op: str):
+ return {"status": 1, "message": "done"}
+
+ res = NvmeofCLICommand.COMMANDS[test_cmd].call(
+ MagicMock(),
+ {"format": "plain", "op": "rebuild"}
+ )
+ assert res.retval == 0
+ assert res.stdout == "operation rebuild finished with status done"
+ assert res.stderr == ''
+
+ del NvmeofCLICommand.COMMANDS[test_cmd]
+ assert test_cmd not in NvmeofCLICommand.COMMANDS
+
+
class TestNVMeoFConfCLI(unittest.TestCase, CLICommandTestMixin):
def setUp(self):
projects / ceph-ci.git / commitdiff