from abc import ABC, abstractmethod
from enum import Enum
import errno
+import inspect
import json
import logging
from typing import Annotated, Any, Callable, Dict, List, NamedTuple, Optional, Type, \
self._success_message_template = success_message_template
self._success_message_fn = success_message_fn
+ self._func_defaults: Dict[str, Any] = {}
def _use_api_endpoint_desc_if_available(self, func):
if not self.desc and hasattr(func, 'doc_info'):
def __call__(self, func) -> HandlerFuncType: # type: ignore
resp = super().__call__(func)
-
+ self._func_defaults = self._compute_func_defaults()
if self._alias:
self._alias_cmd = NvmeofCLICommand(
self._alias,
)
assert self._alias_cmd is not None
self._alias_cmd(func)
+ self._alias_cmd._func_defaults = self._alias_cmd._compute_func_defaults()
self._use_api_endpoint_desc_if_available(func)
return resp
+
+ def _compute_func_defaults(self) -> Dict[str, Any]:
+ defaults: Dict[str, Any] = {}
+ sig = inspect.signature(self.func)
+
+ for name, param in sig.parameters.items():
+ if name in CLICommand.KNOWN_ARGS:
+ continue
+ if name not in self.arg_spec:
+ continue
+ if param.default is not inspect.Parameter.empty:
+ defaults[name] = param.default
+
+ return defaults
+
def _args_map_from_argspec(self,
cmd_dict: Dict[str, Any],
inbuf: Optional[str] = None) -> Dict[str, Any]:
- """
- Build a dict of param_name -> value using the same mechanism as CLICommand.call.
- This applies defaults and type casting via CephArgtype.cast_to.
- """
kwargs, specials = self._collect_args_by_argspec(cmd_dict)
if inbuf and 'inbuf' in specials:
kwargs['inbuf'] = inbuf
- return kwargs
+
+ return {**self._func_defaults, **kwargs}
def _stringify(self, value: Any) -> str:
)
assert res.retval == 0
assert res.stdout == "ns 42 hosts a,b"
+
+ 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.
+ 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
+ }
+
+ # Run command
+ result = cmd.call(mgr=None, cmd_dict=cmd_dict, inbuf=None)
+ assert result.rval == 0
+ assert result.err == ""
+ assert result.out == (
+ "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()
+
+ cmd_dict = {
+ "nqn": "nqn.2014-08.org.nvmexpress:uuid:abcd",
+ "host_name": "nvme-host-2",
+ "traddr": "192.168.1.10",
+ "trsvcid": 8009, # override default 4420
+ }
+
+ result = cmd.call(mgr=None, cmd_dict=cmd_dict, inbuf=None)
+
+ assert result.rval == 0
+ assert result.err == ""
+ assert result.out == (
+ "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,
+ traddr: str,
+ trsvcid: int = 4420,
+ gw_group: Optional[str] = None, # None default intentionally used
+ ):
+ return Model(status="ok")
+
+ cmd = NvmeofCLICommand(
+ "nvmeof listener add",
+ model=Model,
+ success_message_template="Adding {nqn} listener at {traddr}:{trsvcid} gw={gw_group}: Successful",
+ )
+ cmd(create_with_none)
+
+ cmd_dict = {
+ "nqn": "nqn.none.test",
+ "traddr": "127.0.0.1",
+ # 'gw_group' omitted; None default should be injected
+ }
+
+ 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 == (
+ "Adding nqn.none.test listener at 127.0.0.1:4420 gw=None: Successful"
+ )
class TestNVMeoFConfCLI(unittest.TestCase, CLICommandTestMixin):