Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions code2schema/analyzer/events.py
Original file line number Diff line number Diff line change
Expand Up @@ -89,12 +89,14 @@ def _find_emitters(modules: List[ModuleIR]) -> List[DomainEvent]:

def _find_handlers(events: List[DomainEvent], modules: List[ModuleIR]) -> None:
"""Krok 2: dopasuj handlery do zdarzeń (mutuje events in-place)."""
# Event keywords are invariant for this pass; normalize each only once.
keywords = [(ev, ev.name.lower().replace("event", "").strip()) for ev in events]
for mod in modules:
for f in mod.functions:
if _EVENT_HANDLER_PATTERNS.search(f.name):
for ev in events:
keyword = ev.name.lower().replace("event", "").strip()
if keyword and keyword in f.name.lower():
handler_name = f.name.lower()
for ev, keyword in keywords:
if keyword and keyword in handler_name:
ev.handled_by.append(f.qualified_name)


Expand Down
30 changes: 30 additions & 0 deletions tests/test_event_handlers.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,30 @@
import copy
import re

from code2schema.analyzer.events import DomainEvent, _find_handlers
from code2schema.core.models import FunctionIR, ModuleIR


def test_handler_matching_preserves_order_duplicates_case_and_empty_keywords():
names = ["UserEvent", "UserEvent", "Event", "Entity1Event", "Entity10Event", "ŻółćEvent"]
events = [DomainEvent(name=n, emitted_by=str(i), handled_by=["existing"]) for i,n in enumerate(names)]
modules = [ModuleIR(name=m, path=m+".py", functions=[FunctionIR(name=n,module=m) for n in [
"handle_user", "ON_USER", "handle_entity10", "on_żółć", "plain_user", "consumer_user"]]) for m in ["a","b"]]
expected = copy.deepcopy(events)
for _ in range(2):
for mod in modules:
for f in mod.functions:
if re.search(r"(on_|handle_|listener|subscriber|consumer|process_|receive_)", f.name, re.I):
for ev in expected:
keyword = ev.name.lower().replace("event", "").strip()
if keyword and keyword in f.name.lower():
ev.handled_by.append(f.qualified_name)
_find_handlers(events, modules)
assert events == expected


def test_empty_events_and_modules_are_noops():
events = [DomainEvent(name="UserEvent", emitted_by="emit")]
_find_handlers(events, [])
_find_handlers([], [])
assert events[0].handled_by == []