Coverage for little_loops / testing.py: 100%

1"""Offline test harness for little-loops extensions. 

2 

3Provides :class:`LLTestBus` — a standalone replay engine that loads a recorded 

4``.events.jsonl`` file and dispatches events through registered 

5:class:`~little_loops.extension.LLExtension` instances without running a live loop. 

6 

7Example usage:: 

8 

9 from little_loops.testing import LLTestBus 

10 

11 bus = LLTestBus.from_jsonl("path/to/recorded.events.jsonl") 

12 bus.register(MyExtension()) 

13 bus.replay() 

14 assert len(bus.delivered_events) == 15 

15 assert bus.delivered_events[0].type == "loop_start" 

16""" 

17 

18from __future__ import annotations 

19 

20import fnmatch 

21from pathlib import Path 

22from typing import TYPE_CHECKING 

23 

24from little_loops.events import EventBus, LLEvent 

25 

26if TYPE_CHECKING: 

27 from little_loops.extension import LLExtension 

28 

29 

30class LLTestBus: 

31 """Offline event replay harness for testing :class:`~little_loops.extension.LLExtension` handlers. 

32 

33 Loads a pre-recorded ``.events.jsonl`` file, replays events through 

34 registered extensions offline (no live loop execution), and exposes 

35 :attr:`delivered_events` for assertions. 

36 

37 Attributes: 

38 delivered_events: Events actually delivered to at least one extension 

39 (i.e. events that passed the ``event_filter`` of any registered extension). 

40 """ 

41 

42 def __init__(self, events: list[LLEvent]) -> None: 

43 self._events: list[LLEvent] = events 

44 self._extensions: list[LLExtension] = [] 

45 self.delivered_events: list[LLEvent] = [] 

46 

47 @classmethod 

48 def from_jsonl(cls, path: str | Path) -> LLTestBus: 

49 """Create an :class:`LLTestBus` from a JSONL events file. 

50 

51 Args: 

52 path: Path to the ``.events.jsonl`` file. If the file does not exist, 

53 an empty bus is returned (no events, no error). 

54 

55 Returns: 

56 A new :class:`LLTestBus` instance loaded with events from the file. 

57 """ 

58 events = EventBus.read_events(Path(path)) 

59 return cls(events) 

60 

61 def register(self, ext: LLExtension) -> None: 

62 """Register an extension to receive events during :meth:`replay`. 

63 

64 Args: 

65 ext: An object implementing the :class:`~little_loops.extension.LLExtension` 

66 protocol (must have ``on_event`` and optionally ``event_filter``). 

67 """ 

68 self._extensions.append(ext) 

69 

70 def replay(self) -> None: 

71 """Replay all loaded events through registered extensions. 

72 

73 For each event, applies each extension's ``event_filter`` (if set) using 

74 glob matching, then calls ``ext.on_event(event)`` for matching events. 

75 Events that pass the filter for at least one extension are appended to 

76 :attr:`delivered_events`. 

77 

78 ``event_filter`` semantics mirror :class:`~little_loops.events.EventBus`: 

79 

80 - ``None`` (or absent): the extension receives every event. 

81 - ``str``: a single glob pattern matched against ``event.type``. 

82 - ``list[str]``: any matching pattern causes delivery. 

83 """ 

84 self.delivered_events = [] 

85 

86 for event in self._events: 

87 delivered = False 

88 for ext in self._extensions: 

89 ef = getattr(ext, "event_filter", None) 

90 patterns: list[str] | None = None 

91 if ef is not None: 

92 patterns = [ef] if isinstance(ef, str) else list(ef) 

93 

94 if patterns is not None and not any( 

95 fnmatch.fnmatch(event.type, p) for p in patterns 

96 ): 

97 continue 

98 

99 ext.on_event(event) 

100 delivered = True 

101 

102 if delivered: 

103 self.delivered_events.append(event)