mgr/dashboard: dont show empty table when no data for read commands

Fixes: https://tracker.ceph.com/issues/77853
Signed-off-by: Tomer Haskalovitch <tomer.haska@ibm.com>

diff --git a/src/pybind/mgr/dashboard/model/nvmeof.py b/src/pybind/mgr/dashboard/model/nvmeof.py
index ec703740042002329a985d085e2b06bd72c2ec86..e89dde7eb6e0c191faa7164ad38ca8becfcaa5e2 100644 (file)
--- a/src/pybind/mgr/dashboard/model/nvmeof.py
+++ b/src/pybind/mgr/dashboard/model/nvmeof.py
@@ -25,6 +25,24 @@ class CliFieldTransformer:
         return self.func(data)
 
 
+class CliEmptyMessage:
+    """Annotation to specify message when EXCLUSIVE_LIST is empty.
+
+    Template variables available:
+     - Fields from the response dict / parent NamedTuple (e.g., {subsystem_nqn})
+     - CLI command arguments passed to NvmeofCLICommand (e.g., {nqn})
+
+    Example:
+        listeners: Annotated[
+            List[Listener],
+            CliFlags.EXCLUSIVE_LIST,
+            CliEmptyMessage("No listeners for {subsystem_nqn}")
+        ]
+    """
+    def __init__(self, template: str):
+        self.template = template
+
+
 class GatewayInfo(NamedTuple):
     bool_status: Annotated[bool, CliFlags.DROP]
     status: int
@@ -91,7 +109,8 @@ class Subsystem(NamedTuple):
 class SubsystemList(NamedTuple):
     status: int
     error_message: str
-    subsystems: Annotated[List[Subsystem], CliFlags.EXCLUSIVE_LIST]
+    subsystems: Annotated[List[Subsystem], CliFlags.EXCLUSIVE_LIST,
+                          CliEmptyMessage("No subsystems")]
 
 
 class SubsystemStatus(NamedTuple):
@@ -133,7 +152,8 @@ class ConnectionList(NamedTuple):
     status: int
     error_message: str
     subsystem_nqn: str
-    connections: Annotated[List[Connection], CliFlags.EXCLUSIVE_LIST]
+    connections: Annotated[List[Connection], CliFlags.EXCLUSIVE_LIST,
+                           CliEmptyMessage("No connections for {subsystem_nqn}")]
 
 
 class LatencyStats(NamedTuple):
@@ -280,7 +300,9 @@ class Listener(NamedTuple):
 class ListenerList(NamedTuple):
     status: int
     error_message: str
-    listeners: Annotated[List[Listener], CliFlags.EXCLUSIVE_LIST]
+    listeners: Annotated[List[Listener], CliFlags.EXCLUSIVE_LIST,
+                         CliEmptyMessage("No listeners for {nqn}")]
+    nqn: Annotated[str, CliFlags.DROP] = ""
 
 
 class Host(NamedTuple):
@@ -296,7 +318,8 @@ class HostsInfo(NamedTuple):
     error_message: str
     allow_any_host: bool
     subsystem_nqn: str
-    hosts: Annotated[List[Host], CliFlags.EXCLUSIVE_LIST]
+    hosts: Annotated[List[Host], CliFlags.EXCLUSIVE_LIST,
+                     CliEmptyMessage("No hosts are allowed to access {subsystem_nqn}")]
 
 
 class PollGroupTransportInfo(NamedTuple):

diff --git a/src/pybind/mgr/dashboard/services/nvmeof_cli.py b/src/pybind/mgr/dashboard/services/nvmeof_cli.py
index 5f454c3c24e9fb5b8e8941ee328e0455a8bfa869..98948932fd4bf3f0ae2534d087c0900a08f1ac84 100644 (file)
--- a/src/pybind/mgr/dashboard/services/nvmeof_cli.py
+++ b/src/pybind/mgr/dashboard/services/nvmeof_cli.py
@@ -15,7 +15,7 @@ from prettytable import PrettyTable
 
 from ..cli import DBCLICommand
 from ..exceptions import DashboardException
-from ..model.nvmeof import CliFieldTransformer, CliFlags, CliHeader
+from ..model.nvmeof import CliEmptyMessage, CliFieldTransformer, CliFlags, CliHeader
 from ..rest_client import RequestException
 from .nvmeof_conf import ManagedByOrchestratorException, \
     NvmeofGatewayAlreadyExists, NvmeofGatewaysConfig
@@ -171,7 +171,7 @@ def resolve_nvmeof_server_address(
 
 class OutputFormatter(ABC):
     @abstractmethod
-    def format_output(self, data, model):
+    def format_output(self, data, model, template_context: Optional[Dict] = None):
         """Format the given data for output."""
         raise NotImplementedError()
 
@@ -249,7 +249,8 @@ class AnnotatedDataTextOutputFormatter(OutputFormatter):
     # pylint: disable=too-many-branches, too-many-nested-blocks
     def process_dict(self, input_dict: dict,
                      nt_class: Type[NamedTuple],
-                     is_top_level: bool) -> Union[Dict, str, List]:
+                     is_top_level: bool,
+                     template_context: Optional[Dict] = None) -> Union[Dict, str, List]:
         result: Dict = {}
         if not input_dict:
             return result
@@ -266,6 +267,7 @@ class AnnotatedDataTextOutputFormatter(OutputFormatter):
             annotations = []
             output_name = field
             skip = False
+            empty_message_template = None
 
             if origin is Annotated:
                 actual_type, *annotations = get_args(type_hint)
@@ -275,20 +277,38 @@ class AnnotatedDataTextOutputFormatter(OutputFormatter):
                         break
                     if isinstance(annotation, CliHeader):
                         output_name = annotation.label
+                    if isinstance(annotation, CliEmptyMessage):
+                        empty_message_template = annotation.template
+
+                for annotation in annotations:
                     if isinstance(annotation, CliFieldTransformer):
                         value = annotation.transform(value)
                     if is_top_level and annotation == CliFlags.EXCLUSIVE_LIST:
                         assert get_origin(actual_type) == list
                         assert len(get_args(actual_type)) == 1
+                        if not value and empty_message_template:
+                            format_dict = {**input_dict}
+                            if template_context:
+                                format_dict.update(template_context)
+                            try:
+                                return empty_message_template.format(**format_dict)
+                            except KeyError as e:
+                                logger.warning(
+                                    "Missing template variable %s in empty message template: %s",
+                                    e, empty_message_template
+                                )
+                                # Fall back to returning the template as-is if formatting fails
+                                return empty_message_template
                         return [self.process_dict(item, get_args(actual_type)[0],
-                                                  False) for item in value]
+                                                  False, template_context) for item in value]
                     if is_top_level and annotation == CliFlags.EXCLUSIVE_RESULT:
                         return f"Failure: {input_dict.get('error_message')}" if bool(
                             input_dict[field]) else "Success"
                     if annotation == CliFlags.SIZE:
                         value = convert_from_bytes(int(input_dict[field]))
                     elif annotation == CliFlags.PROMOTE_INTERNAL_FIELDS:
-                        object_to_promote = self.process_dict(value, actual_type, False)
+                        object_to_promote = self.process_dict(
+                            value, actual_type, False, template_context)
                         if isinstance(object_to_promote, dict):
                             for field_name, value in object_to_promote.items():
                                 result[field_name] = value
@@ -308,14 +328,14 @@ class AnnotatedDataTextOutputFormatter(OutputFormatter):
 
         return result
 
-    def _convert_to_text_output(self, data, model):
-        data = self.process_dict(data, model, True)
+    def _convert_to_text_output(self, data, model, template_context: Optional[Dict] = None):
+        data = self.process_dict(data, model, True, template_context)
         if isinstance(data, str):
             return data
         return self._get_text_output(data)
 
-    def format_output(self, data, model):
-        return self._convert_to_text_output(data, model)
+    def format_output(self, data, model, template_context: Optional[Dict] = None):
+        return self._convert_to_text_output(data, model, template_context)
 
 
 class NvmeofCLICommand(DBCLICommand):
@@ -491,8 +511,12 @@ class NvmeofCLICommand(DBCLICommand):
                     logger.warning("Formatting of success message failed for %s",
                                    self.prefix, exc_info=True)
 
-                out = message if message else self._output_formatter.format_output(ret, self._model)
-                wrn_msg = ret.get('error_message', '')
+                # Pass args_map as template_context for variable substitution in empty messages
+                # This ensures parameters like 'nqn' are available without polluting the data
+                out = message if message else self._output_formatter.format_output(
+                    ret, self._model, template_context=args_map
+                )
+                wrn_msg = ret.get('error_message', '') if isinstance(ret, dict) else ''
                 if wrn_msg:
                     out += f"\nWarning: {wrn_msg}"
 

diff --git a/src/pybind/mgr/dashboard/tests/test_nvmeof_cli.py b/src/pybind/mgr/dashboard/tests/test_nvmeof_cli.py
index 690103969dfcaf25fbbed06023745bf7e6e62a79..0821da9fcce578552bf6efbe67af781ebb4dde15 100644 (file)
--- a/src/pybind/mgr/dashboard/tests/test_nvmeof_cli.py
+++ b/src/pybind/mgr/dashboard/tests/test_nvmeof_cli.py
@@ -11,7 +11,7 @@ from mgr_module import HandleCommandResult
 from ..cli import DBCLICommand
 from ..controllers import EndpointDoc
 from ..exceptions import DashboardException
-from ..model.nvmeof import CliFieldTransformer, CliFlags, CliHeader
+from ..model.nvmeof import CliEmptyMessage, CliFieldTransformer, CliFlags, CliHeader
 from ..services.nvmeof_cli import AnnotatedDataTextOutputFormatter, \
     NvmeofCLICommand, convert_from_bytes, convert_to_bytes, \
     format_host_updates, resolve_nvmeof_server_address
@@ -1338,3 +1338,435 @@ class TestResolveNvmeofServerAddress:
             traddr=None,
             require=False,
         ) is None
+
+
+class TestCliEmptyMessage:
+    def test_empty_list_with_message_returns_custom_text(self):
+        test_cmd = "nvmeof test empty list message"
+
+        class Item(NamedTuple):
+            name: str
+
+        class TestModel(NamedTuple):
+            status: int
+            error_message: str
+            items: Annotated[List[Item], CliFlags.EXCLUSIVE_LIST, CliEmptyMessage("No items found")]
+
+        @NvmeofCLICommand(test_cmd, TestModel)
+        def func(_):  # pylint: disable=unused-argument, unused-variable
+            return {'status': 0, 'error_message': '', 'items': []}
+        try:
+            result = NvmeofCLICommand.COMMANDS[test_cmd].call(MagicMock(), {})
+
+            assert isinstance(result, HandleCommandResult)
+            assert result.retval == 0
+            assert result.stdout == "No items found"
+            assert result.stderr == ''
+        finally:
+            del NvmeofCLICommand.COMMANDS[test_cmd]
+
+    def test_empty_list_with_template_substitution(self):
+        test_cmd = "nvmeof test empty list template"
+
+        class Item(NamedTuple):
+            name: str
+
+        class TestModel(NamedTuple):
+            status: int
+            error_message: str
+            items: Annotated[List[Item], CliFlags.EXCLUSIVE_LIST,
+                             CliEmptyMessage("No items for {context}")]
+            context: Annotated[str, CliFlags.DROP] = ""
+
+        @NvmeofCLICommand(test_cmd, TestModel)
+        def func(_):  # pylint: disable=unused-argument, unused-variable
+            return {'status': 0, 'error_message': '', 'items': [], 'context': 'test-subsystem'}
+        try:
+            result = NvmeofCLICommand.COMMANDS[test_cmd].call(MagicMock(), {})
+
+            assert isinstance(result, HandleCommandResult)
+            assert result.retval == 0
+            assert result.stdout == "No items for test-subsystem"
+            assert result.stderr == ''
+        finally:
+            del NvmeofCLICommand.COMMANDS[test_cmd]
+
+    def test_non_empty_list_returns_table(self):
+        test_cmd = "nvmeof test non empty list"
+
+        class Item(NamedTuple):
+            name: str
+            value: int
+
+        class TestModel(NamedTuple):
+            status: int
+            error_message: str
+            items: Annotated[List[Item], CliFlags.EXCLUSIVE_LIST, CliEmptyMessage("No items found")]
+
+        @NvmeofCLICommand(test_cmd, TestModel)
+        def func(_):  # pylint: disable=unused-argument, unused-variable
+            return {
+                'status': 0,
+                'error_message': '',
+                'items': [{'name': 'item1', 'value': 10}, {'name': 'item2', 'value': 20}]
+            }
+        try:
+            result = NvmeofCLICommand.COMMANDS[test_cmd].call(MagicMock(), {})
+
+            assert isinstance(result, HandleCommandResult)
+            assert result.retval == 0
+            assert 'Name' in result.stdout
+            assert 'Value' in result.stdout
+            assert 'item1' in result.stdout
+            assert 'item2' in result.stdout
+            assert result.stderr == ''
+        finally:
+            del NvmeofCLICommand.COMMANDS[test_cmd]
+
+    def test_empty_message_with_missing_template_var_fallback(self):
+        test_cmd = "nvmeof test missing template var"
+
+        class Item(NamedTuple):
+            name: str
+
+        class TestModel(NamedTuple):
+            status: int
+            error_message: str
+            items: Annotated[List[Item], CliFlags.EXCLUSIVE_LIST,
+                             CliEmptyMessage("No items for {missing_field}")]
+
+        @NvmeofCLICommand(test_cmd, TestModel)
+        def func(_):  # pylint: disable=unused-argument, unused-variable
+            return {'status': 0, 'error_message': '', 'items': []}
+        try:
+            result = NvmeofCLICommand.COMMANDS[test_cmd].call(MagicMock(), {})
+
+            assert isinstance(result, HandleCommandResult)
+            assert result.retval == 0
+            assert "No items for {missing_field}" in result.stdout
+            assert result.stderr == ''
+        finally:
+            del NvmeofCLICommand.COMMANDS[test_cmd]
+
+    def test_empty_message_json_format_returns_empty_list(self):
+        test_cmd = "nvmeof test empty json"
+
+        class Item(NamedTuple):
+            name: str
+
+        class TestModel(NamedTuple):
+            status: int
+            error_message: str
+            items: Annotated[List[Item], CliFlags.EXCLUSIVE_LIST, CliEmptyMessage("No items found")]
+
+        @NvmeofCLICommand(test_cmd, TestModel)
+        def func(_):  # pylint: disable=unused-argument, unused-variable
+            return {'status': 0, 'error_message': '', 'items': []}
+        try:
+            result = NvmeofCLICommand.COMMANDS[test_cmd].call(MagicMock(), {'format': 'json'})
+
+            assert isinstance(result, HandleCommandResult)
+            assert result.retval == 0
+            data = json.loads(result.stdout)
+            assert data == {'status': 0, 'error_message': '', 'items': []}
+            assert result.stderr == ''
+        finally:
+            del NvmeofCLICommand.COMMANDS[test_cmd]
+
+    def test_empty_message_yaml_format_returns_empty_list(self):
+        test_cmd = "nvmeof test empty yaml"
+
+        class Item(NamedTuple):
+            name: str
+
+        class TestModel(NamedTuple):
+            status: int
+            error_message: str
+            items: Annotated[List[Item], CliFlags.EXCLUSIVE_LIST, CliEmptyMessage("No items found")]
+
+        @NvmeofCLICommand(test_cmd, TestModel)
+        def func(_):  # pylint: disable=unused-argument, unused-variable
+            return {'status': 0, 'error_message': '', 'items': []}
+
+        try:
+            result = NvmeofCLICommand.COMMANDS[test_cmd].call(MagicMock(), {'format': 'yaml'})
+
+            assert isinstance(result, HandleCommandResult)
+            assert result.retval == 0
+            assert 'items: []' in result.stdout
+            assert result.stderr == ''
+        finally:
+            del NvmeofCLICommand.COMMANDS[test_cmd]
+
+    def test_multiple_template_variables(self):
+        test_cmd = "nvmeof test multiple vars"
+
+        class Item(NamedTuple):
+            name: str
+
+        class TestModel(NamedTuple):
+            status: int
+            error_message: str
+            items: Annotated[List[Item], CliFlags.EXCLUSIVE_LIST,
+                             CliEmptyMessage("No {item_type} for {subsystem} in {location}")]
+            item_type: Annotated[str, CliFlags.DROP] = ""
+            subsystem: Annotated[str, CliFlags.DROP] = ""
+            location: Annotated[str, CliFlags.DROP] = ""
+
+        @NvmeofCLICommand(test_cmd, TestModel)
+        def func(_):  # pylint: disable=unused-argument, unused-variable
+            return {
+                'status': 0,
+                'error_message': '',
+                'items': [],
+                'item_type': 'listeners',
+                'subsystem': 'nqn.test',
+                'location': 'gateway1'
+            }
+        try:
+            result = NvmeofCLICommand.COMMANDS[test_cmd].call(MagicMock(), {})
+
+            assert isinstance(result, HandleCommandResult)
+            assert result.retval == 0
+            assert result.stdout == "No listeners for nqn.test in gateway1"
+            assert result.stderr == ''
+        finally:
+            del NvmeofCLICommand.COMMANDS[test_cmd]
+
+    def test_empty_message_with_special_characters(self):
+        test_cmd = "nvmeof test special chars"
+
+        class Item(NamedTuple):
+            name: str
+
+        class TestModel(NamedTuple):
+            status: int
+            error_message: str
+            items: Annotated[List[Item], CliFlags.EXCLUSIVE_LIST,
+                             CliEmptyMessage("No items found! (subsystem: {nqn})")]
+            nqn: Annotated[str, CliFlags.DROP] = ""
+
+        @NvmeofCLICommand(test_cmd, TestModel)
+        def func(_):  # pylint: disable=unused-argument, unused-variable
+            return {
+                'status': 0,
+                'error_message': '',
+                'items': [],
+                'nqn': 'test-nqn'
+            }
+
+        try:
+            result = NvmeofCLICommand.COMMANDS[test_cmd].call(MagicMock(), {})
+            assert isinstance(result, HandleCommandResult)
+            assert result.retval == 0
+            assert "No items found! (subsystem: test-nqn)" in result.stdout
+        finally:
+            del NvmeofCLICommand.COMMANDS[test_cmd]
+
+
+class TestCliEmptyMessageForAllListCommands:
+    def test_subsystem_list_empty_message(self):
+        from ..model.nvmeof import SubsystemList
+
+        test_cmd = "nvmeof test subsystem list empty"
+
+        @NvmeofCLICommand(test_cmd, SubsystemList)
+        def func(_):  # pylint: disable=unused-argument, unused-variable
+            return {'status': 0, 'error_message': '', 'subsystems': []}
+        try:
+            result = NvmeofCLICommand.COMMANDS[test_cmd].call(MagicMock(), {})
+
+            assert isinstance(result, HandleCommandResult)
+            assert result.retval == 0
+            assert result.stdout == "No subsystems"
+            assert result.stderr == ''
+        finally:
+            del NvmeofCLICommand.COMMANDS[test_cmd]
+
+    def test_subsystem_list_non_empty_returns_table(self):
+        from ..model.nvmeof import SubsystemList
+
+        test_cmd = "nvmeof test subsystem list non empty"
+
+        @NvmeofCLICommand(test_cmd, SubsystemList)
+        def func(_):  # pylint: disable=unused-argument, unused-variable
+            return {
+                'status': 0,
+                'error_message': '',
+                'subsystems': [{
+                    'nqn': 'nqn.2016-06.io.spdk:cnode1',
+                    'enable_ha': True,
+                    'serial_number': 'SPDK00000000000001',
+                    'model_number': 'SPDK bdev Controller',
+                    'min_cntlid': 1,
+                    'max_cntlid': 65519,
+                    'namespace_count': 0,
+                    'subtype': 'NVMe',
+                    'max_namespaces': 256,
+                    'has_dhchap_key': False,
+                    'allow_any_host': False,
+                    'created_without_key': False,
+                    'network_mask': []
+                }]
+            }
+        try:
+            result = NvmeofCLICommand.COMMANDS[test_cmd].call(MagicMock(), {})
+
+            assert isinstance(result, HandleCommandResult)
+            assert result.retval == 0
+            assert 'nqn.2016-06.io.spdk:cnode1' in result.stdout
+            assert 'SPDK00000000000001' in result.stdout
+        finally:
+            del NvmeofCLICommand.COMMANDS[test_cmd]
+
+    def test_connection_list_empty_message_with_subsystem(self):
+        from ..model.nvmeof import ConnectionList
+
+        test_cmd = "nvmeof test connection list empty"
+
+        @NvmeofCLICommand(test_cmd, ConnectionList)
+        def func(_):  # pylint: disable=unused-argument, unused-variable
+            return {
+                'status': 0,
+                'error_message': '',
+                'subsystem_nqn': 'nqn.2016-06.io.spdk:cnode1',
+                'connections': []
+            }
+        try:
+            result = NvmeofCLICommand.COMMANDS[test_cmd].call(MagicMock(), {})
+
+            assert isinstance(result, HandleCommandResult)
+            assert result.retval == 0
+            assert result.stdout == "No connections for nqn.2016-06.io.spdk:cnode1"
+            assert result.stderr == ''
+        finally:
+            del NvmeofCLICommand.COMMANDS[test_cmd]
+
+    def test_connection_list_non_empty_returns_table(self):
+        from ..model.nvmeof import ConnectionList
+
+        test_cmd = "nvmeof test connection list non empty"
+
+        @NvmeofCLICommand(test_cmd, ConnectionList)
+        def func(_):  # pylint: disable=unused-argument, unused-variable
+            return {
+                'status': 0,
+                'error_message': '',
+                'subsystem_nqn': 'nqn.2016-06.io.spdk:cnode1',
+                'connections': [{
+                    'nqn': 'nqn.2014-08.org.nvmexpress:uuid:12345',
+                    'traddr': '192.168.1.100',
+                    'trsvcid': 4420,
+                    'trtype': 'TCP',
+                    'adrfam': 0,
+                    'connected': True,
+                    'qpairs_count': 2,
+                    'controller_id': 1,
+                    'use_psk': False,
+                    'use_dhchap': False,
+                    'dhchap_controller_origin': None,
+                    'subsystem': 'nqn.2016-06.io.spdk:cnode1',
+                    'disconnected_due_to_keepalive_timeout': False
+                }]
+            }
+        try:
+            result = NvmeofCLICommand.COMMANDS[test_cmd].call(MagicMock(), {})
+
+            assert isinstance(result, HandleCommandResult)
+            assert result.retval == 0
+            assert 'nqn.2014-08.org.nvmexpress:uuid:12345' in result.stdout
+            assert '192.168.1.100' in result.stdout
+        finally:
+            del NvmeofCLICommand.COMMANDS[test_cmd]
+
+    def test_hosts_info_empty_message_with_subsystem(self):
+        from ..model.nvmeof import HostsInfo
+
+        test_cmd = "nvmeof test hosts info empty"
+
+        @NvmeofCLICommand(test_cmd, HostsInfo)
+        def func(_):  # pylint: disable=unused-argument, unused-variable
+            return {
+                'status': 0,
+                'error_message': '',
+                'allow_any_host': False,
+                'subsystem_nqn': 'nqn.2016-06.io.spdk:cnode1',
+                'hosts': []
+            }
+        try:
+            result = NvmeofCLICommand.COMMANDS[test_cmd].call(MagicMock(), {})
+
+            assert isinstance(result, HandleCommandResult)
+            assert result.retval == 0
+            assert result.stdout == "No hosts are allowed to access nqn.2016-06.io.spdk:cnode1"
+            assert result.stderr == ''
+        finally:
+            del NvmeofCLICommand.COMMANDS[test_cmd]
+
+    def test_hosts_info_non_empty_returns_table(self):
+        from ..model.nvmeof import HostsInfo
+
+        test_cmd = "nvmeof test hosts info non empty"
+
+        @NvmeofCLICommand(test_cmd, HostsInfo)
+        def func(_):  # pylint: disable=unused-argument, unused-variable
+            return {
+                'status': 0,
+                'error_message': '',
+                'allow_any_host': False,
+                'subsystem_nqn': 'nqn.2016-06.io.spdk:cnode1',
+                'hosts': [{
+                    'nqn': 'nqn.2014-08.org.nvmexpress:uuid:host1',
+                    'use_psk': False,
+                    'use_dhchap': True,
+                    'dhchap_controller_origin': 'controller',
+                    'disconnected_due_to_keepalive_timeout': False
+                }]
+            }
+        try:
+            result = NvmeofCLICommand.COMMANDS[test_cmd].call(MagicMock(), {})
+
+            assert isinstance(result, HandleCommandResult)
+            assert result.retval == 0
+            assert 'nqn.2014-08.org.nvmexpress:uuid:host1' in result.stdout
+        finally:
+            del NvmeofCLICommand.COMMANDS[test_cmd]
+
+    def test_all_empty_messages_work_with_json_format(self):
+        from ..model.nvmeof import ConnectionList, HostsInfo, SubsystemList
+
+        test_cmd_subsys = "nvmeof test subsystem json"
+
+        @NvmeofCLICommand(test_cmd_subsys, SubsystemList)
+        def func_subsys(_):  # pylint: disable=unused-argument, unused-variable
+            return {'status': 0, 'error_message': '', 'subsystems': []}
+        try:
+            result = NvmeofCLICommand.COMMANDS[test_cmd_subsys].call(
+                MagicMock(), {'format': 'json'})
+            assert '"subsystems": []' in result.stdout
+        finally:
+            del NvmeofCLICommand.COMMANDS[test_cmd_subsys]
+
+        test_cmd_conn = "nvmeof test connection json"
+
+        @NvmeofCLICommand(test_cmd_conn, ConnectionList)
+        def func_conn(_):  # pylint: disable=unused-argument, unused-variable
+            return {'status': 0, 'error_message': '',
+                    'subsystem_nqn': 'test', 'connections': []}
+        try:
+            result = NvmeofCLICommand.COMMANDS[test_cmd_conn].call(MagicMock(), {'format': 'json'})
+            assert '"connections": []' in result.stdout
+        finally:
+            del NvmeofCLICommand.COMMANDS[test_cmd_conn]
+
+        test_cmd_hosts = "nvmeof test hosts json"
+
+        @NvmeofCLICommand(test_cmd_hosts, HostsInfo)
+        def func_hosts(_):  # pylint: disable=unused-argument, unused-variable
+            return {'status': 0, 'error_message': '', 'allow_any_host': False,
+                    'subsystem_nqn': 'test', 'hosts': []}
+        try:
+            result = NvmeofCLICommand.COMMANDS[test_cmd_hosts].call(MagicMock(), {'format': 'json'})
+            assert '"hosts": []' in result.stdout
+        finally:
+            del NvmeofCLICommand.COMMANDS[test_cmd_hosts]