Merge pull request #460 from katlapinka/init-tests

Add init test with different runlevels and test for management device status

test/functional/api/cas/cas_module.py

@@ -4,7 +4,7 @@
 #
 
 from aenum import Enum
-
+from core.test_run import TestRun
 from test_utils import os_utils
 from test_utils.os_utils import ModuleRemoveMethod
 
@@ -24,3 +24,7 @@ def unload_all_cas_modules():
                                   os_utils.ModuleRemoveMethod.rmmod)
     os_utils.unload_kernel_module(CasModule.disk.value,
                                   os_utils.ModuleRemoveMethod.rmmod)
+
+
+def is_cas_management_dev_present():
+    return TestRun.executor.run("test -c /dev/cas_ctrl").exit_code == 0

test/functional/tests/initialize/test_initialize_runlevel.py

@@ -0,0 +1,89 @@
+#
+# Copyright(c) 2020 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause-Clear
+#
+
+import os
+import pytest
+import time
+from api.cas import casadm, casadm_parser
+from api.cas.cache_config import CacheMode
+from api.cas.init_config import InitConfig
+from core.test_run import TestRun
+from storage_devices.disk import DiskTypeLowerThan, DiskTypeSet, DiskType
+from test_tools.disk_utils import Filesystem
+from test_tools.fio.fio import Fio
+from test_tools.fio.fio_param import ReadWrite, IoEngine
+from test_utils import os_utils
+from test_utils.os_utils import Runlevel
+from test_utils.size import Size, Unit
+
+
+mount_point = "/mnt/test"
+
+
+@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
+@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
+@pytest.mark.parametrize("runlevel", [Runlevel.runlevel5, Runlevel.runlevel3])
+@pytest.mark.parametrize("cache_mode", CacheMode)
+def test_init_reboot_runlevels(runlevel, cache_mode):
+    """
+        title: Initialize CAS devices after reboot
+        description: |
+          Verify that CAS init script starts cache properly after reboot in different runlevels.
+        pass_criteria:
+          - Cache should be loaded successfully after reboot.
+    """
+    with TestRun.step(f"Set runlevel to {runlevel.value}."):
+        os_utils.change_runlevel(runlevel)
+
+    with TestRun.step("Prepare CAS device."):
+        cache_disk = TestRun.disks['cache']
+        cache_disk.create_partitions([Size(2, Unit.GibiByte)])
+        cache_dev = cache_disk.partitions[0]
+        core_disk = TestRun.disks['core']
+        core_disk.create_partitions([Size(1, Unit.GibiByte)])
+        core_dev = core_disk.partitions[0]
+
+        cache = casadm.start_cache(cache_dev, cache_mode, force=True)
+        core = cache.add_core(core_dev)
+
+    with TestRun.step("Create CAS init config based on running configuration."):
+        InitConfig.create_init_config_from_running_configuration()
+
+    with TestRun.step("Make filesystem on CAS device and mount it."):
+        core.create_filesystem(Filesystem.xfs)
+        core.mount(mount_point)
+
+    with TestRun.step("Start writing file to CAS."):
+        fio = Fio().create_command()\
+            .file_name(os.path.join(mount_point, "test_file"))\
+            .read_write(ReadWrite.randwrite)\
+            .io_engine(IoEngine.sync)\
+            .num_jobs(1).direct()\
+            .file_size(Size(30, Unit.GibiByte))
+
+        fio.run_in_background()
+        os_utils.sync()
+        os_utils.drop_caches()
+
+        time.sleep(10)
+        TestRun.executor.run_expect_success("pgrep fio")
+
+    with TestRun.step("Reboot machine during writing a file."):
+        TestRun.executor.reboot()
+
+    with TestRun.step("Check if cache was properly started at boot time"):
+        # Wait for CAS to load after boot
+        time.sleep(60)
+        caches = casadm_parser.get_caches()
+        if len(caches) == 1:
+            TestRun.LOGGER.info("Cache started properly at boot time.")
+        else:
+            TestRun.LOGGER.error("Cache did not start properly at boot time.")
+
+    with TestRun.step("Stop cache and set default runlevel."):
+        if len(caches) != 0:
+            casadm.stop_all_caches()
+        os_utils.change_runlevel(Runlevel.runlevel3)
+        TestRun.executor.reboot()

test/functional/tests/initialize/test_initialize_status.py

@@ -0,0 +1,43 @@
+#
+# Copyright(c) 2020 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause-Clear
+#
+
+from api.cas import cas_module, casctl
+from api.cas.cas_module import CasModule
+from core.test_run import TestRun
+from test_utils import os_utils
+
+
+def test_init_status():
+    """
+        title: CAS management device status
+        description: |
+          Verify that CAS management device is present in OS only when CAS modules are loaded.
+        pass_criteria:
+          - CAS management device present in OS when CAS modules are loaded.
+          - CAS management device not present in OS when CAS modules are not loaded.
+    """
+    with TestRun.step("Check if CAS management device is present in OS."):
+        if cas_module.is_cas_management_dev_present():
+            TestRun.LOGGER.info("CAS management device is present in OS when CAS module is loaded.")
+        else:
+            TestRun.fail("CAS management device is not present in OS when CAS module is loaded.")
+
+    with TestRun.step("Remove CAS module."):
+        cas_module.unload_all_cas_modules()
+
+    with TestRun.step("Stop CAS service."):
+        casctl.stop()
+
+    with TestRun.step("Check if CAS management device is not present in OS."):
+        if not cas_module.is_cas_management_dev_present():
+            TestRun.LOGGER.info(
+                "CAS management device is not present in OS when CAS module is not loaded.")
+        else:
+            TestRun.fail("CAS management device is present in OS when CAS module is not loaded.")
+
+    with TestRun.step("Load CAS modules and start CAS service."):
+        os_utils.load_kernel_module(CasModule.cache.value)
+        os_utils.load_kernel_module(CasModule.disk.value)
+        casctl.start()