pyocf: Test parallelism improvements

tests/functional/pyocf/types/queue.py

@@ -1,11 +1,12 @@
 #
 # Copyright(c) 2019-2022 Intel Corporation
 # Copyright(c) 2024 Huawei Technologies
+# Copyright(c) 2026 Unvertical
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
 from ctypes import c_void_p, CFUNCTYPE, Structure, byref
-from threading import Thread, Condition, Event, Semaphore
+from threading import Thread, Condition, Event, Semaphore, current_thread
 import weakref
 
 from ..ocf import OcfLib
@@ -113,7 +114,8 @@ def stop(self):
             self.stop_event.set()
             self.kick_condition.notify_all()
 
-        self.thread.join()
+        if current_thread() is not self.thread:
+            self.thread.join()
 
     # settle - wait for OCF to finish execution within this queue context
     #

tests/functional/tests/conftest.py

@@ -1,5 +1,6 @@
 #
 # Copyright(c) 2019-2022 Intel Corporation
+# Copyright(c) 2026 Unvertical
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -27,21 +28,31 @@ def pytest_configure(config):
     sys.path.append(os.path.join(os.path.dirname(__file__), os.path.pardir))
 
 
+@pytest.hookimpl(tryfirst=True, hookwrapper=True)
+def pytest_runtest_makereport(item, call):
+    outcome = yield
+    rep = outcome.get_result()
+    if rep.when == "call":
+        item.test_failed = rep.failed
+
+
 @pytest.fixture()
-def pyocf_ctx():
+def pyocf_ctx(request):
     c = OcfCtx.with_defaults(DefaultLogger(LogLevel.WARN))
     for vol_type in default_registered_volumes:
         c.register_volume_type(vol_type)
     c.register_internal_volume_type_id(CVolume, get_composite_volume_type_id())
     yield c
     c.exit()
     gc.collect()
+    if getattr(request.node, "test_failed", False):
+        return
     if len(Volume._instances_) > 0:
         warnings.warn("Not all Volumes have been closed!!!")
 
 
 @pytest.fixture()
-def pyocf_ctx_log_buffer():
+def pyocf_ctx_log_buffer(request):
     logger = BufferLogger(LogLevel.DEBUG)
     c = OcfCtx.with_defaults(logger)
     for vol_type in default_registered_volumes:
@@ -50,12 +61,14 @@ def pyocf_ctx_log_buffer():
     yield logger
     c.exit()
     gc.collect()
+    if getattr(request.node, "test_failed", False):
+        return
     if len(Volume._instances_) > 0:
         warnings.warn("Not all Volumes have been closed!!!")
 
 
 @pytest.fixture()
-def pyocf_2_ctx():
+def pyocf_2_ctx(request):
     c1 = OcfCtx.with_defaults(DefaultLogger(LogLevel.WARN, "Ctx1"))
     c2 = OcfCtx.with_defaults(DefaultLogger(LogLevel.WARN, "Ctx2"))
     for vol_type in default_registered_volumes:
@@ -67,12 +80,14 @@ def pyocf_2_ctx():
     c1.exit()
     c2.exit()
     gc.collect()
+    if getattr(request.node, "test_failed", False):
+        return
     if len(Volume._instances_) > 0:
         warnings.warn("Not all Volumes have been closed!!!")
 
 
 @pytest.fixture()
-def pyocf_2_ctx_log_buffer():
+def pyocf_2_ctx_log_buffer(request):
     logger1 = BufferLogger(LogLevel.WARN, LogLevel.DEBUG, "Ctx1")
     logger2 = BufferLogger(LogLevel.WARN, LogLevel.DEBUG, "Ctx2")
     c1 = OcfCtx.with_defaults(logger1)
@@ -84,5 +99,7 @@ def pyocf_2_ctx_log_buffer():
     c1.exit()
     c2.exit()
     gc.collect()
+    if getattr(request.node, "test_failed", False):
+        return
     if len(Volume._instances_) > 0:
         warnings.warn("Not all Volumes have been closed!!!")

tests/functional/tests/engine/test_read.py

@@ -1,6 +1,7 @@
 #
 # Copyright(c) 2019-2022 Intel Corporation
 # Copyright(c) 2024 Huawei Technologies
+# Copyright(c) 2026 Unvertical
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -11,7 +12,7 @@
 import pytest
 import random
 from hashlib import md5
-from datetime import datetime
+from tests.utils.random import get_random_seed
 
 from pyocf.types.cache import Cache, CacheMode
 from pyocf.types.core import Core
@@ -185,8 +186,7 @@ def print_test_case(
 
 @pytest.mark.parametrize("cacheline_size", CacheLineSize)
 @pytest.mark.parametrize("cache_mode", CacheMode)
-@pytest.mark.parametrize("rand_seed", [datetime.now().timestamp()])
-def test_read_data_consistency(pyocf_ctx, cacheline_size, cache_mode, rand_seed):
+def test_read_data_consistency(pyocf_ctx, cacheline_size, cache_mode):
     CACHELINE_COUNT = 9
     SECTOR_SIZE = Size.from_sector(1).B
     CLS = cacheline_size // SECTOR_SIZE
@@ -195,7 +195,7 @@ def test_read_data_consistency(pyocf_ctx, cacheline_size, cache_mode, rand_seed)
     SECTOR_COUNT = int(WORKSET_SIZE / SECTOR_SIZE)
     ITRATION_COUNT = 50
 
-    random.seed(rand_seed)
+    random.seed(get_random_seed())
 
     # start sector for each region (positions of '*' on the above diagram)
     region_start = (

tests/functional/tests/engine/test_seq_cutoff.py

@@ -115,6 +115,7 @@ def test_seq_cutoff_max_streams(pyocf_ctx):
     io_size = threshold - smallest_io_size
     io_to_streams(vol, queue, streams, io_size)
 
+    queue.settle()
     stats = cache.get_stats()
     assert (
         stats["req"]["serviced"]["value"] == stats["req"]["total"]["value"] == len(streams)
@@ -129,6 +130,7 @@ def test_seq_cutoff_max_streams(pyocf_ctx):
     shuffle(streams)
     io_to_streams(vol, queue, streams, smallest_io_size)
 
+    queue.settle()
     stats = cache.get_stats()
     assert (
         stats["req"]["serviced"]["value"] == old_serviced
@@ -140,6 +142,7 @@ def test_seq_cutoff_max_streams(pyocf_ctx):
     # STEP 3
     io_to_streams(vol, queue, [non_active_stream], smallest_io_size)
 
+    queue.settle()
     stats = cache.get_stats()
     assert (
         stats["req"]["serviced"]["value"] == old_serviced + 1
@@ -149,6 +152,7 @@ def test_seq_cutoff_max_streams(pyocf_ctx):
     io_to_streams(vol, queue, [lru_stream], smallest_io_size)
 
     vol.close()
+    queue.settle()
     stats = cache.get_stats()
 
     assert (

tests/functional/tests/management/test_composite_volume.py

@@ -9,7 +9,8 @@
 import random
 import time
 from ctypes import POINTER, c_int, cast, c_void_p
-from datetime import datetime, timedelta
+from datetime import timedelta
+from tests.utils.random import get_random_seed
 from threading import Event
 from collections import namedtuple
 
@@ -623,8 +624,7 @@ def test_io_propagation_entire_dev(pyocf_ctx):
     cvol.destroy()
 
 
-@pytest.mark.parametrize("rand_seed", [datetime.now().timestamp()])
-def test_io_propagation_multiple_subvolumes(pyocf_ctx, rand_seed):
+def test_io_propagation_multiple_subvolumes(pyocf_ctx):
     """
     title: Perform multi-subvolume operations.
     description: |
@@ -646,7 +646,7 @@ def test_io_propagation_multiple_subvolumes(pyocf_ctx, rand_seed):
     requirements:
       - composite_volume::io_request_passing
     """
-    random.seed(rand_seed)
+    random.seed(get_random_seed())
     pyocf_ctx.register_volume_type(TraceDevice)
 
     vol_size = S.from_MiB(1)
@@ -731,8 +731,7 @@ def test_io_propagation_multiple_subvolumes(pyocf_ctx, rand_seed):
     cvol.destroy()
 
 
-@pytest.mark.parametrize("rand_seed", [datetime.now().timestamp()])
-def test_io_completion(pyocf_ctx, rand_seed):
+def test_io_completion(pyocf_ctx):
     """
     title: Composite volume completion order.
     description: |
@@ -754,7 +753,7 @@ def test_io_completion(pyocf_ctx, rand_seed):
     requirements:
       - composite_volume::io_request_completion
     """
-    random.seed(rand_seed)
+    random.seed(get_random_seed())
 
     class PendingIoVolume(RamVolume):
         def __init__(self, *args, **kwargs):
@@ -852,8 +851,7 @@ def resume_next(self):
     cvol.destroy()
 
 
-@pytest.mark.parametrize("rand_seed", [datetime.now().timestamp()])
-def test_io_error(pyocf_ctx, rand_seed):
+def test_io_error(pyocf_ctx):
     """
     title: Composite volume error propagation.
     description: |
@@ -876,7 +874,7 @@ def test_io_error(pyocf_ctx, rand_seed):
     requirements:
       - composite_volume::io_error_handling
     """
-    random.seed(rand_seed)
+    random.seed(get_random_seed())
     pyocf_ctx.register_volume_type(TraceDevice)
 
     vol_size = S.from_MiB(1)

tests/functional/tests/utils/random.py

@@ -1,5 +1,6 @@
 #
 # Copyright(c) 2019-2023 Intel Corporation
+# Copyright(c) 2026 Unvertical
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -10,6 +11,11 @@
 from ctypes import c_uint64, c_uint32, c_uint16, c_uint8, c_int, c_uint
 
 
+def get_random_seed():
+    with open("config/random.cfg") as f:
+        return int(f.read())
+
+
 class Range:
     def __init__(self, min_val, max_val):
         self.min = min_val
@@ -30,8 +36,7 @@ class DefaultRanges(Range, enum.Enum):
 
 class RandomGenerator:
     def __init__(self, base_range=DefaultRanges.INT, count=1000):
-        with open("config/random.cfg") as f:
-            self.random = random.Random(int(f.read()))
+        self.random = random.Random(get_random_seed())
         self.exclude = []
         self.range = base_range
         self.count = count
@@ -60,8 +65,7 @@ def __next__(self):
 
 class RandomStringGenerator:
     def __init__(self, len_range=Range(0, 20), count=700, extra_chars=[]):
-        with open("config/random.cfg") as f:
-            self.random = random.Random(int(f.read()))
+        self.random = random.Random(get_random_seed())
         self.generator = self.__string_generator(len_range, extra_chars)
         self.count = count
         self.n = 0