Add some tests

Signed-off-by: mulhern <[email protected]>

tests/whitebox/_misc.py

@@ -24,15 +24,20 @@
 import sys
 import time
 import unittest
+from uuid import UUID
 
 # isort: THIRDPARTY
 import psutil
 
 # isort: LOCAL
-from stratis_cli import run
+from stratis_cli import StratisCliErrorCodes, run
+from stratis_cli._actions._connection import get_object
+from stratis_cli._actions._constants import TOP_OBJECT
 from stratis_cli._error_reporting import handle_error
 from stratis_cli._errors import StratisCliActionError
 
+_OK = StratisCliErrorCodes.OK
+
 
 def device_name_list(min_devices=0, max_devices=10, unique=False):
     """
@@ -270,3 +275,45 @@ def test_runner(command_line):
 
 
 TEST_RUNNER = test_runner
+
+
+def stop_pool(pool_name):
+    """
+    Stop a pool and return the UUID of the pool.
+    This method exists because it is the most direct way to get the UUID of
+    a pool that has just been stopped, for testing.
+
+    :param str pool_name: the name of the pool to stop
+
+    :returns: the UUID of the stopped pool
+    :raises: RuntimeError
+    """
+    # pylint: disable=import-outside-toplevel
+    # isort: LOCAL
+    from stratis_cli._actions._data import Manager, ObjectManager, pools
+
+    proxy = get_object(TOP_OBJECT)
+
+    managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {})
+    (pool_object_path, _) = next(
+        pools(props={"Name": pool_name})
+        .require_unique_match(True)
+        .search(managed_objects)
+    )
+
+    ((stopped, pool_uuid), return_code, message) = Manager.Methods.StopPool(
+        proxy, {"pool": pool_object_path}
+    )
+
+    if not return_code == _OK:
+        raise RuntimeError(
+            "Pool with name %s was not stopped: %s" % (pool_name, message)
+        )
+
+    if not stopped:
+        raise RuntimeError(
+            "Pool with name %s was supposed to have been started but was not"
+            % pool_name
+        )
+
+    return UUID(pool_uuid)

tests/whitebox/integration/pool/test_list.py

@@ -1,4 +1,4 @@
-# Copyright 2016 Red Hat, Inc.
+# Copyright 2022 Red Hat, Inc.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -15,10 +15,19 @@
 Test 'list'.
 """
 
-from .._misc import RUNNER, TEST_RUNNER, SimTestCase, device_name_list
+# isort: STDLIB
+from uuid import uuid4
+
+# isort: LOCAL
+from stratis_cli import StratisCliErrorCodes
+from stratis_cli._errors import StratisCliResourceNotFoundError
+
+from .._misc import RUNNER, TEST_RUNNER, SimTestCase, device_name_list, stop_pool
 
 _DEVICE_STRATEGY = device_name_list(1)
 
+_ERROR = StratisCliErrorCodes.ERROR
+
 
 class ListTestCase(SimTestCase):
     """
@@ -90,3 +99,51 @@ def test_list_with_cache(self):
         TEST_RUNNER(command_line)
         command_line = self._MENU
         TEST_RUNNER(command_line)
+
+
+class List3TestCase(SimTestCase):
+    """
+    Test listing stopped pools.
+    """
+
+    _MENU = ["--propagate", "pool", "list-stopped"]
+    _POOLNAME = "deadpool"
+
+    def setUp(self):
+        """
+        Start the stratisd daemon with the simulator. Create a pool.
+        """
+        super().setUp()
+        command_line = ["pool", "create", self._POOLNAME] + _DEVICE_STRATEGY()
+        RUNNER(command_line)
+
+    def test_list(self):
+        """
+        Test listing all with a stopped pool.
+        """
+        command_line = ["pool", "stop", self._POOLNAME]
+        RUNNER(command_line)
+        TEST_RUNNER(self._MENU)
+
+    def test_list_empty(self):
+        """
+        Test listing when there are no stopped pools.
+        """
+        TEST_RUNNER(self._MENU)
+
+    def test_list_specific(self):
+        """
+        Test listing a specific stopped pool.
+        """
+
+        pool_uuid = stop_pool(self._POOLNAME)
+
+        command_line = self._MENU + ["--uuid=%s" % pool_uuid]
+        TEST_RUNNER(command_line)
+
+    def test_list_bogus(self):
+        """
+        Test listing a bogus stopped pool.
+        """
+        command_line = self._MENU + ["--uuid=%s" % uuid4()]
+        self.check_error(StratisCliResourceNotFoundError, command_line, _ERROR)

tests/whitebox/integration/pool/test_start.py

@@ -0,0 +1,62 @@
+# Copyright 2022 Red Hat, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+Test 'start'.
+"""
+
+# isort: STDLIB
+from uuid import uuid4
+
+# isort: LOCAL
+from stratis_cli import StratisCliErrorCodes
+from stratis_cli._errors import StratisCliEngineError, StratisCliNoChangeError
+
+from .._misc import RUNNER, TEST_RUNNER, SimTestCase, device_name_list, stop_pool
+
+_ERROR = StratisCliErrorCodes.ERROR
+_DEVICE_STRATEGY = device_name_list(1, 1)
+
+
+class StartTestCase(SimTestCase):
+    """
+    Test 'start' on a sim pool.
+    """
+
+    _MENU = ["--propagate", "pool", "start"]
+    _POOLNAME = "poolname"
+
+    def setUp(self):
+        super().setUp()
+        command_line = ["pool", "create", self._POOLNAME] + _DEVICE_STRATEGY()
+        RUNNER(command_line)
+
+    def test_bad_uuid(self):
+        """
+        Test trying to start a pool with non-existent UUID.
+        """
+        command_line = ["pool", "stop", self._POOLNAME]
+        RUNNER(command_line)
+        command_line = self._MENU + [str(uuid4())]
+        self.check_error(StratisCliEngineError, command_line, _ERROR)
+
+    def test_good_uuid(self):
+        """
+        Test trying to start a pool with a good UUID.
+        """
+        pool_uuid = stop_pool(self._POOLNAME)
+
+        command_line = self._MENU + [str(pool_uuid)]
+        TEST_RUNNER(command_line)
+
+        self.check_error(StratisCliNoChangeError, command_line, _ERROR)

tests/whitebox/integration/pool/test_stop.py

@@ -0,0 +1,47 @@
+# Copyright 2022 Red Hat, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+Test 'stop'.
+"""
+
+# isort: LOCAL
+from stratis_cli import StratisCliErrorCodes
+
+from .._misc import RUNNER, TEST_RUNNER, SimTestCase, device_name_list
+
+_ERROR = StratisCliErrorCodes.ERROR
+_DEVICE_STRATEGY = device_name_list(1, 1)
+
+
+class StopTestCase(SimTestCase):
+    """
+    Test 'stop' on a sim pool.
+    """
+
+    _MENU = ["--propagate", "pool", "stop"]
+    _POOLNAME = "poolname"
+
+    def setUp(self):
+        super().setUp()
+        command_line = ["pool", "create", self._POOLNAME] + _DEVICE_STRATEGY()
+        RUNNER(command_line)
+
+    def test_stop(self):
+        """
+        Stopping with known name should always succeed.
+        """
+        command_line = self._MENU + [
+            self._POOLNAME,
+        ]
+        TEST_RUNNER(command_line)