Add thread safety checks and fix field conflict test

Author: Евгений Блинов

tests/thread_safety/conftest.py

@@ -55,7 +55,10 @@ def acquire(self) -> None:
             self._owner_thread_id = thread_id
 
     def release(self) -> None:
+        thread_id = get_ident()
         with self._state_lock:
+            if self._owner_thread_id != thread_id:
+                raise RuntimeError('Release from a non-owner thread was detected.')
             self._owner_thread_id = None
         self._lock.release()
 
@@ -110,7 +113,10 @@ def callback(old_value: Any, new_value: Any, storage: Any) -> None:  # noqa: ARG
         entered_event.set()
         if exception is not None:
             raise exception
-        release_event.wait()
+        if not release_event.wait(DEFAULT_THREAD_TIMEOUT):
+            raise TimeoutError(
+                f'Gate callback "{label}" was not released within {DEFAULT_THREAD_TIMEOUT} seconds.',
+            )
         event_log.append(f'{label}:exit')
 
     return callback
@@ -121,14 +127,26 @@ def replace_field_lock_with_locklib_lock(
     field_name: str,
     expected_group_fields: List[str],
 ) -> LocklibDeadlockDetectingLock:
+    if field_name not in instance.__locks__:
+        raise RuntimeError(f'The "{field_name}" field is missing from instance.__locks__.')
+
     old_lock = instance.__locks__[field_name]
     new_lock = LocklibDeadlockDetectingLock()
+    replaced_fields = []
 
     for current_field_name, current_lock in list(instance.__locks__.items()):
         if current_lock is old_lock:
             instance.__locks__[current_field_name] = new_lock
+            replaced_fields.append(current_field_name)
+
+    if not replaced_fields:
+        raise RuntimeError(f'No fields were rebound for the "{field_name}" lock group.')
 
     for expected_field_name in expected_group_fields:
+        if expected_field_name not in instance.__locks__:
+            raise RuntimeError(
+                f'The expected field "{expected_field_name}" is missing from instance.__locks__.',
+            )
         if instance.__locks__[expected_field_name] is not new_lock:
             raise RuntimeError(
                 f'The "{expected_field_name}" field was not rebound to the expected lock group.',

tests/thread_safety/fields/test_base.py

@@ -336,6 +336,7 @@ def reader() -> None:
     while not read_values.empty():
         collected_reads.append(read_values.get_nowait())
 
+    assert len(collected_reads) == num_readers * iterations
     assert set(collected_reads).issubset({0, 1, 2, 3, 4, 5})
 
 

tests/thread_safety/test_storage.py

@@ -18,14 +18,22 @@ class SomeClass(Storage):
 
 
 def test_share_mutex_with_and_conflicts_merge_into_one_group():
+    def fields_conflict(
+        _old_first: int,
+        new_first: int,
+        _old_third: int,
+        new_third: int,
+    ) -> bool:
+        return new_first > 0 and new_third > 0
+
     class SomeClass(Storage):
         first_field = Field(
-            1,
+            0,
             share_mutex_with=['second_field'],
-            conflicts={'third_field': lambda *_: False},
+            conflicts={'third_field': fields_conflict},
         )
-        second_field = Field(2)
-        third_field = Field(3)
+        second_field = Field(0)
+        third_field = Field(0)
 
     instance = SomeClass()