Coverage for test\test_event.py: 100%

1# test_events.py 

2import pytest 

3import asyncio 

4from unittest.mock import Mock, call 

5from nexios import NexiosApp, get_application 

6from nexios.events import ( 

7 EventEmitter, 

8 EventPriority, 

9 EventPhase, 

10) 

11 

12 

13# Fixture for app with event emitter, ensuring clean state 

14@pytest.fixture 

15def app_with_emitter(): 

16 app = get_application() 

17 app.emitter = EventEmitter() 

18 yield app 

19 app.emitter.remove_all_events() 

20 

21 

22# Fixture for a mock object to track listener calls 

23@pytest.fixture 

24def listener_mock(): 

25 return Mock() 

26 

27 

28# Test basic event registration and triggering 

29@pytest.mark.asyncio 

30async def test_app_event_registration_and_trigger(app_with_emitter, listener_mock): 

31 app = app_with_emitter 

32 

33 async def startup_handler(): 

34 listener_mock("started") 

35 

36 # Register listener explicitly 

37 app.emitter.event("app:startup").listen(startup_handler) 

38 

39 # Trigger event and ensure completion 

40 stats = app.emitter.emit("app:startup") 

41 

42 # Allow async tasks to complete 

43 await asyncio.sleep(0) # Yield to event loop 

44 

45 assert stats["listeners_executed"] == 1 

46 assert listener_mock.call_args_list == [call("started")] 

47 

48 

49# Test one-time listeners 

50@pytest.mark.asyncio 

51async def test_app_one_time_listener(app_with_emitter, listener_mock): 

52 app = app_with_emitter 

53 

54 async def request_handler(): 

55 listener_mock("handled") 

56 

57 # Register one-time listener 

58 app.emitter.event("app:request").once(request_handler) 

59 

60 # Trigger twice 

61 stats1 = app.emitter.emit("app:request") 

62 stats2 = app.emitter.emit("app:request") 

63 

64 # Allow async tasks to complete 

65 await asyncio.sleep(0) 

66 

67 assert stats1["listeners_executed"] == 1 

68 assert stats2["listeners_executed"] == 0 # Listener removed after first trigger 

69 assert listener_mock.call_args_list == [call("handled")] 

70 

71 

72# Test event priorities 

73@pytest.mark.asyncio 

74async def test_app_event_priorities(app_with_emitter, listener_mock): 

75 app = app_with_emitter 

76 

77 async def high_priority(): 

78 listener_mock("high") 

79 

80 async def low_priority(): 

81 listener_mock("low") 

82 

83 # Register listeners with different priorities 

84 app.emitter.event("app:process").listen(high_priority, priority=EventPriority.HIGH) 

85 app.emitter.event("app:process").listen(low_priority, priority=EventPriority.LOW) 

86 

87 # Trigger event 

88 stats = app.emitter.emit("app:process") 

89 

90 # Allow async tasks to complete 

91 await asyncio.sleep(0) 

92 

93 assert stats["listeners_executed"] == 2 

94 assert listener_mock.call_args_list == [call("high"), call("low")] 

95 

96 

97# Test event propagation (parent-child) 

98@pytest.mark.asyncio 

99async def test_app_event_propagation(app_with_emitter, listener_mock): 

100 app = app_with_emitter 

101 parent_event = app.emitter.event("app:parent") 

102 child_event = app.emitter.event("app:child") 

103 child_event.parent = parent_event 

104 

105 async def parent_handler(*args, phase=EventPhase.AT_TARGET): 

106 listener_mock(f"parent-{phase.name}") 

107 

108 async def child_handler(*args, phase=EventPhase.AT_TARGET): 

109 listener_mock(f"child-{phase.name}") 

110 

111 # Register listeners 

112 parent_event.listen(parent_handler) 

113 child_event.listen(child_handler) 

114 

115 # Trigger child event 

116 stats = app.emitter.emit("app:child") 

117 

118 # Allow async tasks to complete 

119 await asyncio.sleep(0) 

120 

121 assert stats["listeners_executed"] == 1 # Only child listeners at target phase