@APIRouter("/nvmeof/gateway", Scope.NVME_OF)
@APIDoc("NVMe-oF Gateway Management API", "NVMe-oF Gateway")
class NVMeoFGateway(RESTController):
-
- @NvmeofCLICommand("nvmeof gw info", model.GatewayInfo)
+ @NvmeofCLICommand(
+ "nvmeof gateway info", model.GatewayInfo, alias="nvmeof gw info"
+ )
@EndpointDoc("Get information about the NVMeoF gateway")
@convert_to_model(model.GatewayInfo)
@handle_nvmeof_error
@ReadPermission
@Endpoint('GET', '/version')
- @NvmeofCLICommand("nvmeof gw version", model.GatewayVersion)
+ @NvmeofCLICommand(
+ "nvmeof gateway version", model.GatewayVersion, alias="nvmeof gw version"
+ )
@EndpointDoc("Get the version of the NVMeoF gateway")
@convert_to_model(model.GatewayVersion)
@handle_nvmeof_error
@ReadPermission
@Endpoint('GET', '/log_level')
- @NvmeofCLICommand("nvmeof gw get_log_level", model.GatewayLogLevelInfo)
+ @NvmeofCLICommand(
+ "nvmeof gateway get_log_level", model.GatewayLogLevelInfo,
+ alias="nvmeof gw get_log_level"
+ )
@EndpointDoc("Get NVMeoF gateway log level information")
@convert_to_model(model.GatewayLogLevelInfo)
@handle_nvmeof_error
@ReadPermission
@Endpoint('PUT', '/log_level')
- @NvmeofCLICommand("nvmeof gw set_log_level", model.RequestStatus)
+ @NvmeofCLICommand(
+ "nvmeof gateway set_log_level", model.RequestStatus, alias="nvmeof gw set_log_level")
@EndpointDoc("Set NVMeoF gateway log levels")
@convert_to_model(model.RequestStatus)
@handle_nvmeof_error
)
@convert_to_model(model.SubsystemStatus)
@handle_nvmeof_error
- def create(self, nqn: str, enable_ha: bool = True, max_namespaces: int = 4096,
+ def create(self, nqn: str, enable_ha: Optional[bool] = True,
+ max_namespaces: Optional[int] = 4096, no_group_append: Optional[bool] = True,
+ serial_number: Optional[str] = None, dhchap_key: Optional[str] = None,
gw_group: Optional[str] = None, traddr: Optional[str] = None):
return NVMeoFClient(gw_group=gw_group, traddr=traddr).stub.create_subsystem(
NVMeoFClient.pb2.create_subsystem_req(
- subsystem_nqn=nqn, max_namespaces=max_namespaces, enable_ha=enable_ha
+ subsystem_nqn=nqn, serial_number=serial_number,
+ max_namespaces=max_namespaces, enable_ha=enable_ha,
+ no_group_append=no_group_append, dhchap_key=dhchap_key
)
)
@APIDoc("NVMe-oF Subsystem Namespace Management API", "NVMe-oF Subsystem Namespace")
class NVMeoFNamespace(RESTController):
@pick("namespaces")
- @NvmeofCLICommand("nvmeof ns list", model.NamespaceList)
+ @NvmeofCLICommand(
+ "nvmeof namespace list", model.NamespaceList, alias="nvmeof ns list"
+ )
@EndpointDoc(
"List all NVMeoF namespaces in a subsystem",
parameters={
)
@pick("namespaces", first=True)
- @NvmeofCLICommand("nvmeof ns get", model.NamespaceList)
+ @NvmeofCLICommand(
+ "nvmeof namespace get", model.NamespaceList, alias="nvmeof ns get")
@EndpointDoc(
"Get info from specified NVMeoF namespace",
parameters={
@ReadPermission
@Endpoint('GET', '{nsid}/io_stats')
- @NvmeofCLICommand("nvmeof ns get_io_stats", model.NamespaceIOStats)
+ @NvmeofCLICommand(
+ "nvmeof namespace get_io_stats", model.NamespaceIOStats,
+ alias="nvmeof ns get_io_stats"
+ )
@EndpointDoc(
"Get IO stats from specified NVMeoF namespace",
parameters={
subsystem_nqn=nqn, nsid=int(nsid))
)
- @NvmeofCLICommand("nvmeof ns add", model.NamespaceCreation)
+ @NvmeofCLICommand(
+ "nvmeof namespace add", model.NamespaceCreation, alias="nvmeof ns add"
+ )
@EndpointDoc(
"Create a new NVMeoF namespace",
parameters={
@ReadPermission
@Endpoint('PUT', '{nsid}/set_qos')
- @NvmeofCLICommand("nvmeof ns set_qos", model=model.RequestStatus)
+ @NvmeofCLICommand(
+ "nvmeof namespace set_qos", model=model.RequestStatus, alias="nvmeof ns set_qos")
@EndpointDoc(
"set QOS for specified NVMeoF namespace",
parameters={
@ReadPermission
@Endpoint('PUT', '{nsid}/change_load_balancing_group')
- @NvmeofCLICommand("nvmeof ns change_load_balancing_group", model=model.RequestStatus)
+ @NvmeofCLICommand(
+ "nvmeof namespace change_load_balancing_group", model=model.RequestStatus,
+ alias="nvmeof ns change_load_balancing_group"
+ )
@EndpointDoc(
"set the load balancing group for specified NVMeoF namespace",
parameters={
@ReadPermission
@Endpoint('PUT', '{nsid}/resize')
- @NvmeofCLICommand("nvmeof ns resize", model=model.RequestStatus)
+ @NvmeofCLICommand(
+ "nvmeof namespace resize", model=model.RequestStatus, alias="nvmeof ns resize"
+ )
@EndpointDoc(
"resize the specified NVMeoF namespace",
parameters={
@ReadPermission
@Endpoint('PUT', '{nsid}/add_host')
- @NvmeofCLICommand("nvmeof ns add_host", model=model.RequestStatus)
+ @NvmeofCLICommand(
+ "nvmeof namespace add_host", model=model.RequestStatus, alias="nvmeof ns add_host"
+ )
@EndpointDoc(
"Adds a host to the specified NVMeoF namespace",
parameters={
@ReadPermission
@Endpoint('PUT', '{nsid}/del_host')
- @NvmeofCLICommand("nvmeof ns del_host", model=model.RequestStatus)
+ @NvmeofCLICommand(
+ "nvmeof namespace del_host", model=model.RequestStatus, alias="nvmeof ns del_host"
+ )
@EndpointDoc(
"Removes a host from the specified NVMeoF namespace",
parameters={
@ReadPermission
@Endpoint('PUT', '{nsid}/change_visibility')
- @NvmeofCLICommand("nvmeof ns change_visibility", model=model.RequestStatus)
+ @NvmeofCLICommand(
+ "nvmeof namespace change_visibility", model=model.RequestStatus,
+ alias="nvmeof ns change_visibility"
+ )
@EndpointDoc(
"changes the visibility of the specified NVMeoF namespace to all or selected hosts",
parameters={
@ReadPermission
@Endpoint('PUT', '{nsid}/set_auto_resize')
- @NvmeofCLICommand("nvmeof ns set_auto_resize", model=model.RequestStatus)
+ @NvmeofCLICommand(
+ "nvmeof namespace set_auto_resize", model=model.RequestStatus,
+ alias="nvmeof ns set_auto_resize"
+ )
@EndpointDoc(
"Enable or disable namespace auto resize when RBD image is resized",
parameters={
@ReadPermission
@Endpoint('PUT', '{nsid}/set_rbd_trash_image')
- @NvmeofCLICommand("nvmeof ns set_rbd_trash_image", model=model.RequestStatus)
+ @NvmeofCLICommand(
+ "nvmeof namespace set_rbd_trash_image", model=model.RequestStatus,
+ alias="nvmeof ns set_rbd_trash_image"
+ )
@EndpointDoc(
"changes the trash image on delete of the specified NVMeoF \
namespace to all or selected hosts",
@ReadPermission
@Endpoint('PUT', '{nsid}/refresh_size')
- @NvmeofCLICommand("nvmeof ns refresh_size", model=model.RequestStatus)
+ @NvmeofCLICommand(
+ "nvmeof namespace refresh_size", model=model.RequestStatus,
+ alias="nvmeof ns refresh_size"
+ )
@EndpointDoc(
"refresh the specified NVMeoF namespace to current RBD image size",
parameters={
)
@pick("namespaces", first=True)
- @NvmeofCLICommand("nvmeof ns update", model.NamespaceList)
+ @NvmeofCLICommand(
+ "nvmeof namespace update", model.NamespaceList, alias="nvmeof ns update"
+ )
@EndpointDoc(
"Update an existing NVMeoF namespace",
parameters={
return response
@empty_response
- @NvmeofCLICommand("nvmeof ns del", model.RequestStatus)
+ @NvmeofCLICommand("nvmeof namespace del", model.RequestStatus, alias="nvmeof ns del")
@EndpointDoc(
"Delete an existing NVMeoF namespace",
parameters={
@convert_to_model(model.HostsInfo, finalize=_update_hosts)
@handle_nvmeof_error
def list(
- self, nqn: str, clear_alerts: Optional[bool],
+ self, nqn: str, clear_alerts: Optional[bool] = None,
gw_group: Optional[str] = None, traddr: Optional[str] = None
):
return NVMeoFClient(gw_group=gw_group, traddr=traddr).stub.list_hosts(
return self._get_list_text_output(data)
return self._get_object_text_output(data)
+ def _get_row(self, columns, data_obj):
+ row = []
+ for col in columns:
+ col_val = data_obj.get(col)
+ if col_val is None:
+ col_val = ''
+ row.append(str(col_val))
+ return row
+
def _get_list_text_output(self, data):
columns = list(dict.fromkeys([key for obj in data for key in obj.keys()]))
table = self._create_table(columns)
for d in data:
- row = []
- for col in columns:
- row.append(str(d.get(col)))
- table.add_row(row)
+ table.add_row(self._get_row(columns, d))
return table.get_string()
def _get_object_text_output(self, data):
columns = [k for k in data.keys() if k not in ["status", "error_message"]]
table = self._create_table(columns)
- row = []
- for col in columns:
- row.append(str(data.get(col)))
- table.add_row(row)
+ table.add_row(self._get_row(columns, data))
return table.get_string()
def _is_list_of_complex_type(self, value):
class NvmeofCLICommand(CLICommand):
desc: str
- def __init__(self, prefix, model: Type[NamedTuple], perm='rw', poll=False):
+ def __init__(self, prefix, model: Type[NamedTuple], alias=None, perm='rw', poll=False):
super().__init__(prefix, perm, poll)
self._output_formatter = AnnotatedDataTextOutputFormatter()
self._model = model
+ self._alias = alias
def _use_api_endpoint_desc_if_available(self, func):
if not self.desc and hasattr(func, 'doc_info'):
self.desc = func.doc_info.get('summary', '')
def __call__(self, func) -> HandlerFuncType: # type: ignore
- # pylint: disable=useless-super-delegation
- """
- This method is being overriden solely to be able to disable the linters checks for typing.
- The NvmeofCLICommand decorator assumes a different type returned from the
- function it wraps compared to CLICmmand, breaking a Liskov substitution principal,
- hence triggering linters alerts.
- """
- resp = super().__call__(func)
+ if self._alias:
+ NvmeofCLICommand(self._alias, model=self._model)._register_handler(func)
+ resp = super().__call__(func)
self._use_api_endpoint_desc_if_available(func)
-
return resp
def call(self,
del NvmeofCLICommand.COMMANDS[test_cmd]
assert test_cmd not in NvmeofCLICommand.COMMANDS
+ def test_command_alias_calls_command(self, base_call_mock):
+ test_cmd = "test command1"
+ test_alias = "test alias1"
+
+ class Model(NamedTuple):
+ a: str
+ b: int
+
+ @NvmeofCLICommand(test_cmd, Model, alias=test_alias)
+ def func(_): # noqa # pylint: disable=unused-variable
+ return {'a': '1', 'b': 2}
+
+ assert test_cmd in NvmeofCLICommand.COMMANDS
+ assert test_alias in NvmeofCLICommand.COMMANDS
+
+ result = NvmeofCLICommand.COMMANDS[test_cmd].call(MagicMock(), {})
+ assert result.retval == 0
+ assert result.stdout == (
+ "+-+\n"
+ "|A|\n"
+ "+-+\n"
+ "|b|\n"
+ "+-+"
+ )
+ assert result.stderr == ''
+ base_call_mock.assert_called_once()
+
+ result = NvmeofCLICommand.COMMANDS[test_alias].call(MagicMock(), {})
+ assert result.retval == 0
+ assert result.stdout == (
+ "+-+\n"
+ "|A|\n"
+ "+-+\n"
+ "|b|\n"
+ "+-+"
+ )
+ assert result.stderr == ''
+ assert base_call_mock.call_count == 2
+
+ del NvmeofCLICommand.COMMANDS[test_cmd]
+ del NvmeofCLICommand.COMMANDS[test_alias]
+ assert test_cmd not in NvmeofCLICommand.COMMANDS
+ assert test_alias not in NvmeofCLICommand.COMMANDS
+
class TestNVMeoFConfCLI(unittest.TestCase, CLICommandTestMixin):
def setUp(self):
class TestAnnotatedDataTextOutputFormatter():
+ def test_no_annotation(self):
+ class Sample(NamedTuple):
+ name: str
+ age: int
+ byte: int
+
+ data = {'name': 'Alice', 'age': 30, "byte": 20971520}
+
+ formatter = AnnotatedDataTextOutputFormatter()
+ output = formatter.format_output(data, Sample)
+ assert output == (
+ '+-----+---+--------+\n'
+ '|Name |Age|Byte |\n'
+ '+-----+---+--------+\n'
+ '|Alice|30 |20971520|\n'
+ '+-----+---+--------+'
+ )
+
+ def test_none_to_empty_str_annotation(self):
+ class Sample(NamedTuple):
+ name: str
+ age: int
+ byte: int
+
+ data = {'name': 'Alice', 'age': 30, "byte": None}
+
+ formatter = AnnotatedDataTextOutputFormatter()
+ output = formatter.format_output(data, Sample)
+ assert output == (
+ '+-----+---+----+\n'
+ '|Name |Age|Byte|\n'
+ '+-----+---+----+\n'
+ '|Alice|30 | |\n'
+ '+-----+---+----+'
+ )
+
def test_size_bytes_annotation(self):
class Sample(NamedTuple):
name: str
projects / ceph.git / commitdiff