Merge pull request #532 from Ostrokrzew/update_tests

Update CLI tests

test/functional/api/cas/cli_messages.py

@@ -69,6 +69,10 @@ try_add_core_sector_size_mismatch = [
     r"or try other device with the same logical sector size as core device\."
 ]
 
+no_caches_running = [
+    r"No caches running"
+]
+
 
 def check_stderr_msg(output: Output, expected_messages):
     return __check_string_msg(output.stderr, expected_messages)

test/functional/tests/cli/test_cli_start_stop.py

@@ -3,8 +3,8 @@
 # SPDX-License-Identifier: BSD-3-Clause-Clear
 #
 
-
 import pytest
+from random import randint
 
 from api.cas import casadm, casadm_parser, cli_messages
 from api.cas.cli import start_cmd
@@ -12,79 +12,201 @@ from core.test_run import TestRun
 from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
 from test_utils.size import Unit, Size
 
+CACHE_ID_RANGE = (1, 16384)
+CORE_ID_RANGE = (0, 4095)
+
 
 @pytest.mark.require_disk("cache", DiskTypeSet([DiskType.nand, DiskType.optane]))
 @pytest.mark.parametrize("shortcut", [True, False])
-def test_cli_start_stop_default_value(shortcut):
+def test_cli_start_stop_default_id(shortcut):
-    with TestRun.LOGGER.step("Prepare devices"):
+    """
+        title: Test for starting a cache with a default ID - short and long command
+        description: |
+          Start a new cache with a default ID and then stop this cache.
+        pass_criteria:
+          - The cache has successfully started with default ID
+          - The cache has successfully stopped
+    """
+    with TestRun.step("Prepare the device for the cache."):
         cache_device = TestRun.disks['cache']
         cache_device.create_partitions([Size(500, Unit.MebiByte)])
         cache_device = cache_device.partitions[0]
 
-    with TestRun.LOGGER.step("Start cache"):
+    with TestRun.step("Start the cache."):
-        casadm.start_cache(cache_device, shortcut=shortcut, force=True)
+        cache = casadm.start_cache(cache_device, shortcut=shortcut, force=True)
 
-    with TestRun.LOGGER.step("Check if cache started successfully"):
+    with TestRun.step("Check if the cache has started successfully."):
         caches = casadm_parser.get_caches()
         if len(caches) != 1:
-            TestRun.LOGGER.error(f"There is wrong caches count found in OS: {len(caches)}")
+            TestRun.fail(f"There is a wrong number of caches found in the OS: {len(caches)}. "
-        if caches[0].cache_device.system_path != cache_device.system_path:
+                         f"Should be only 1.")
-            TestRun.LOGGER.error(f"Cache started using wrong device: "
+        if cache.cache_device.system_path != cache_device.system_path:
-                                 f"{caches[0].cache_device.system_path}. "
+            TestRun.fail(f"The cache has started using a wrong device:"
-                                 f"Should be {cache_device.system_path}")
+                         f" {cache.cache_device.system_path}."
+                         f"\nShould use {cache_device.system_path}.")
 
-    with TestRun.LOGGER.step("Stop cache"):
+    with TestRun.step("Stop the cache."):
-        casadm.stop_cache(cache_id=caches[0].cache_id, shortcut=shortcut)
+        casadm.stop_cache(cache.cache_id, shortcut=shortcut)
 
-    with TestRun.LOGGER.step("Check if cache stopped properly"):
+    with TestRun.step("Check if the cache has stopped properly."):
-        output = casadm.list_caches(shortcut=shortcut)
         caches = casadm_parser.get_caches()
         if len(caches) != 0:
-            TestRun.LOGGER.error(f"There is wrong caches count found in OS: {len(caches)}. "
+            TestRun.fail(f"There is a wrong number of caches found in the OS: {len(caches)}."
-                                 f"Should be 0.")
+                         f"\nNo cache should be present after stopping the cache.")
-        if output.stdout != "No caches running":
+        output = casadm.list_caches(shortcut=shortcut)
-            TestRun.LOGGER.error("There is no 'No caches running' info in casadm -L output")
+        cli_messages.check_stdout_msg(output, cli_messages.no_caches_running)
+
+
+@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.nand, DiskType.optane]))
+@pytest.mark.parametrize("shortcut", [True, False])
+def test_cli_start_stop_custom_id(shortcut):
+    """
+        title: Test for starting a cache with a custom ID - short and long command
+        description: |
+          Start a new cache with a random ID (from allowed pool) and then stop this cache.
+        pass_criteria:
+          - The cache has successfully started with a custom ID
+          - The cache has successfully stopped
+    """
+    with TestRun.step("Prepare the device for the cache."):
+        cache_device = TestRun.disks['cache']
+        cache_device.create_partitions([Size(500, Unit.MebiByte)])
+        cache_device = cache_device.partitions[0]
+
+    with TestRun.step("Start the cache with a random ID."):
+        cache_id = randint(*CACHE_ID_RANGE)
+        cache = casadm.start_cache(cache_device, cache_id=cache_id, shortcut=shortcut, force=True)
+        TestRun.LOGGER.info(f"Cache ID: {cache_id}")
+
+    with TestRun.step("Check if the cache has started successfully."):
+        caches = casadm_parser.get_caches()
+        if len(caches) != 1:
+            TestRun.fail(f"There is a wrong number of caches found in the OS: {len(caches)}. "
+                         f"Should be only 1.")
+        if cache.cache_device.system_path != cache_device.system_path:
+            TestRun.fail(f"The cache has started using a wrong device:"
+                         f" {cache.cache_device.system_path}."
+                         f"\nShould use {cache_device.system_path}.")
+
+    with TestRun.step("Stop the cache."):
+        casadm.stop_cache(cache.cache_id, shortcut=shortcut)
+
+    with TestRun.step("Check if the cache has stopped properly."):
+        caches = casadm_parser.get_caches()
+        if len(caches) != 0:
+            TestRun.fail(f"There is a wrong number of caches found in the OS: {len(caches)}."
+                         f"\nNo cache should be present after stopping the cache.")
+        output = casadm.list_caches(shortcut=shortcut)
+        cli_messages.check_stdout_msg(output, cli_messages.no_caches_running)
 
 
 @pytest.mark.require_disk("cache", DiskTypeSet([DiskType.nand, DiskType.optane]))
 @pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
 @pytest.mark.parametrize("shortcut", [True, False])
-def test_cli_add_remove_default_value(shortcut):
+def test_cli_add_remove_default_id(shortcut):
-    cache_device = TestRun.disks['cache']
+    """
-    cache_device.create_partitions([Size(50, Unit.MebiByte)])
+        title: Test for adding and removing a core with a default ID - short and long command
-    cache_device = cache_device.partitions[0]
+        description: |
-    cache = casadm.start_cache(cache_device, shortcut=shortcut, force=True)
+          Start a new cache and add a core to it without passing a core ID as an argument
+          and then remove this core from the cache.
+        pass_criteria:
+          - The core is added to the cache with a default ID
+          - The core is successfully removed from the cache
+    """
+    with TestRun.step("Prepare the devices."):
+        cache_disk = TestRun.disks['cache']
+        cache_disk.create_partitions([Size(50, Unit.MebiByte)])
+        cache_device = cache_disk.partitions[0]
+        core_device = TestRun.disks['core']
 
-    core_device = TestRun.disks['core']
+    with TestRun.step("Start the cache and add the core."):
+        cache = casadm.start_cache(cache_device, shortcut=shortcut, force=True)
+        core = casadm.add_core(cache, core_device, shortcut=shortcut)
 
-    casadm.add_core(cache, core_device, shortcut=shortcut)
+    with TestRun.step("Check if the core is added to the cache."):
+        caches = casadm_parser.get_caches()
+        if len(caches[0].get_core_devices()) != 1:
+            TestRun.fail("One core should be present in the cache.")
+        if caches[0].get_core_devices()[0].system_path != core.system_path:
+            TestRun.fail("The core path should be equal to the path of the core added.")
 
-    caches = casadm_parser.get_caches()
+    with TestRun.step("Remove the core from the cache."):
-    if len(caches[0].get_core_devices()) != 1:
+        casadm.remove_core(cache.cache_id, core.core_id, shortcut=shortcut)
-        TestRun.fail("One core should be present in cache")
+
-    if caches[0].get_core_devices()[0].core_device.system_path != core_device.system_path:
+    with TestRun.step("Check if the core is successfully removed from still running cache."):
-        TestRun.fail("Core path should equal to path of core added")
+        caches = casadm_parser.get_caches()
+        if len(caches) != 1:
+            TestRun.fail("One cache should be still present after removing the core.")
+        if len(caches[0].get_core_devices()) != 0:
+            TestRun.fail("No core device should be present after removing the core.")
+
+    with TestRun.step("Stop the cache."):
+        casadm.stop_cache(cache_id=cache.cache_id, shortcut=shortcut)
+
+    with TestRun.step("Check if the cache has successfully stopped."):
+        caches = casadm_parser.get_caches()
+        if len(caches) != 0:
+            TestRun.fail("No cache should be present after stopping the cache.")
+        output = casadm.list_caches(shortcut=shortcut)
+        cli_messages.check_stdout_msg(output, cli_messages.no_caches_running)
 
 
-    casadm.remove_core(cache.cache_id, 1, shortcut=shortcut)
+@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.nand, DiskType.optane]))
-    caches = casadm_parser.get_caches()
+@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
-    if len(caches) != 1:
+@pytest.mark.parametrize("shortcut", [True, False])
-        TestRun.fail("One cache should be present still after removing core")
+def test_cli_add_remove_custom_id(shortcut):
-    if len(caches[0].get_core_devices()) != 0:
+    """
-        TestRun.fail("No core devices should be present after removing core")
+        title: Test for adding and removing a core with a custom ID - short and long command
+        description: |
+          Start a new cache and add a core to it with passing a random core ID
+          (from allowed pool) as an argument and then remove this core from the cache.
+        pass_criteria:
+          - The core is added to the cache with a default ID
+          - The core is successfully removed from the cache
+    """
+    with TestRun.step("Prepare the devices."):
+        cache_disk = TestRun.disks['cache']
+        cache_disk.create_partitions([Size(50, Unit.MebiByte)])
+        cache_device = cache_disk.partitions[0]
+        core_device = TestRun.disks['core']
 
-    casadm.stop_cache(cache_id=cache.cache_id, shortcut=shortcut)
+    with TestRun.step("Start the cache and add the core with a random ID."):
+        core_id = randint(*CORE_ID_RANGE)
+        cache = casadm.start_cache(cache_device, shortcut=shortcut, force=True)
+        core = casadm.add_core(cache, core_device, core_id=core_id, shortcut=shortcut)
+        TestRun.LOGGER.info(f"Core ID: {core_id}")
 
-    output = casadm.list_caches(shortcut=shortcut)
+    with TestRun.step("Check if the core is added to the cache."):
-    caches = casadm_parser.get_caches()
+        caches = casadm_parser.get_caches()
-    if len(caches) != 0:
+        if len(caches[0].get_core_devices()) != 1:
-        TestRun.fail("No cache should be present after stopping the cache")
+            TestRun.fail("One core should be present in the cache.")
-    if output.stdout != "No caches running":
+        if caches[0].get_core_devices()[0].system_path != core.system_path:
-        TestRun.fail(f"Invalid message, expected 'No caches running', got {output.stdout}")
+            TestRun.fail("The core path should be equal to the path of the core added.")
+
+    with TestRun.step("Remove the core from the cache."):
+        casadm.remove_core(cache.cache_id, core.core_id, shortcut=shortcut)
+
+    with TestRun.step("Check if the core is successfully removed from still running cache."):
+        caches = casadm_parser.get_caches()
+        if len(caches) != 1:
+            TestRun.fail("One cache should be still present after removing the core.")
+        if len(caches[0].get_core_devices()) != 0:
+            TestRun.fail("No core device should be present after removing the core.")
+
+    with TestRun.step("Stop the cache."):
+        casadm.stop_cache(cache_id=cache.cache_id, shortcut=shortcut)
+
+    with TestRun.step("Check if the cache has successfully stopped."):
+        caches = casadm_parser.get_caches()
+        if len(caches) != 0:
+            TestRun.fail("No cache should be present after stopping the cache.")
+        output = casadm.list_caches(shortcut=shortcut)
+        cli_messages.check_stdout_msg(output, cli_messages.no_caches_running)
 
 
 @pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
-def test_cli_load_and_force():
+@pytest.mark.parametrize("shortcut", [True, False])
+def test_cli_load_and_force(shortcut):
     """
         title: Test if it is possible to use start command with 'load' and 'force' flag at once
         description: |
@@ -105,7 +227,7 @@ def test_cli_load_and_force():
 
     with TestRun.step("Try to load cache with 'force'."):
         output = TestRun.executor.run(
-            start_cmd(cache_dev=cache_device.system_path, force=True, load=True)
+            start_cmd(cache_dev=cache_device.system_path, force=True, load=True, shortcut=shortcut)
         )
         if output.exit_code == 0:
             TestRun.fail("Loading cache with 'force' option should fail.")