Coverage for Users / vladimirpavlov / PycharmProjects / parameterizable / tests / test_single_thread_enforcer.py: 96%

1import threading 

2import pytest 

3import mixinforge.single_thread_enforcer_mixin as ste 

4from mixinforge.single_thread_enforcer_mixin import ( 

5 _restrict_to_single_thread, 

6 _reset_thread_ownership, 

7) 

8 

9def setup_function(): 

10 """Reset the enforcer before each test.""" 

11 _reset_thread_ownership() 

12 

13def teardown_function(): 

14 """Reset the enforcer after each test.""" 

15 _reset_thread_ownership() 

16 

17def test_single_thread_success(): 

18 """Test that the first thread to access can access repeatedly.""" 

19 # First access 

20 _restrict_to_single_thread() 

21 assert ste._owner_thread_native_id == threading.get_native_id() 

22 

23 # Second access from same thread 

24 _restrict_to_single_thread() 

25 assert ste._owner_thread_native_id == threading.get_native_id() 

26 

27def test_multi_thread_failure(): 

28 """Test that a second thread raises RuntimeError.""" 

29 # Main thread claims ownership 

30 _restrict_to_single_thread() 

31 

32 exception_caught = False 

33 

34 def intruder_thread(): 

35 nonlocal exception_caught 

36 try: 

37 _restrict_to_single_thread() 

38 except RuntimeError as e: 

39 if "This object is restricted to single-threaded execution" in str(e): 

40 exception_caught = True 

41 

42 t = threading.Thread(target=intruder_thread, name="Intruder") 

43 t.start() 

44 t.join() 

45 

46 assert exception_caught, "Secondary thread should have raised RuntimeError" 

47 

48def test_reset_allows_new_owner(): 

49 """Test that resetting allows a new thread to become the owner.""" 

50 # Main thread claims ownership 

51 _restrict_to_single_thread() 

52 

53 # Reset 

54 _reset_thread_ownership() 

55 assert ste._owner_thread_native_id is None 

56 

57 # New thread attempts to claim ownership 

58 exception_caught = False 

59 

60 def new_owner_thread(): 

61 nonlocal exception_caught 

62 try: 

63 _restrict_to_single_thread() 

64 except Exception: 

65 exception_caught = True 

66 

67 t = threading.Thread(target=new_owner_thread, name="NewOwner") 

68 t.start() 

69 t.join() 

70 

71 assert not exception_caught, "New thread should succeed after reset" 

72 

73 # Now main thread should fail because "NewOwner" claimed it 

74 # Note: "NewOwner" thread finished, but the enforcer remembers the ID. 

75 # The ID might be reused by OS, but likely not immediately. 

76 # Even if "NewOwner" is dead, the enforcer still holds its ID. 

77 

78 # However, if threading.get_native_id() of the main thread is different 

79 # from the stored ID (which was NewOwner's ID), it should raise RuntimeError. 

80 

81 with pytest.raises(RuntimeError) as excinfo: 

82 _restrict_to_single_thread() 

83 

84 assert "This object is restricted to single-threaded execution" in str(excinfo.value) 

85 

86def test_pid_change_resets_ownership(): 

87 """Test that PID change (simulating fork) resets ownership.""" 

88 import os 

89 

90 # Main thread claims ownership 

91 _restrict_to_single_thread() 

92 original_native_id = ste._owner_thread_native_id 

93 original_pid = ste._owner_process_id 

94 

95 assert original_native_id == threading.get_native_id() 

96 assert original_pid == os.getpid() 

97 

98 # Simulate a PID change (as would happen after fork) 

99 # We can't actually fork in pytest easily, so we manually change the PID 

100 fake_new_pid = original_pid + 99999 

101 ste._owner_process_id = fake_new_pid 

102 

103 # Now when we call _restrict_to_single_thread, it should detect PID mismatch 

104 # and reset ownership, allowing the current thread to claim it 

105 _restrict_to_single_thread() 

106 

107 # Ownership should be reset to current thread and PID 

108 assert ste._owner_thread_native_id == threading.get_native_id() 

109 assert ste._owner_process_id == os.getpid() 

110 assert ste._owner_process_id != fake_new_pid