| 31 | |
| 32 | |
| 33 | class AuditEvents: |
| 34 | def __init__(self) -> None: |
| 35 | self.events: dict[str, list[str]] = {} |
| 36 | self.sources: dict[str, set[tuple[str, str]]] = {} |
| 37 | |
| 38 | def __iter__(self) -> Iterator[tuple[str, list[str], tuple[str, str]]]: |
| 39 | for name, args in self.events.items(): |
| 40 | for source in self.sources[name]: |
| 41 | yield name, args, source |
| 42 | |
| 43 | def add_event( |
| 44 | self, name, args: list[str], source: tuple[str, str] |
| 45 | ) -> None: |
| 46 | if name in self.events: |
| 47 | self._check_args_match(name, args) |
| 48 | else: |
| 49 | self.events[name] = args |
| 50 | self.sources.setdefault(name, set()).add(source) |
| 51 | |
| 52 | def _check_args_match(self, name: str, args: list[str]) -> None: |
| 53 | current_args = self.events[name] |
| 54 | msg = ( |
| 55 | f"Mismatched arguments for audit-event {name}: " |
| 56 | f"{current_args!r} != {args!r}" |
| 57 | ) |
| 58 | if current_args == args: |
| 59 | return |
| 60 | if len(current_args) != len(args): |
| 61 | logger.warning(msg) |
| 62 | return |
| 63 | for a1, a2 in zip(current_args, args, strict=False): |
| 64 | if a1 == a2: |
| 65 | continue |
| 66 | if any(a1 in s and a2 in s for s in _SYNONYMS): |
| 67 | continue |
| 68 | logger.warning(msg) |
| 69 | return |
| 70 | |
| 71 | def id_for(self, name) -> str: |
| 72 | source_count = len(self.sources.get(name, set())) |
| 73 | name_clean = re.sub(r"\W", "_", name) |
| 74 | return f"audit_event_{name_clean}_{source_count}" |
| 75 | |
| 76 | def rows(self) -> Iterator[tuple[str, list[str], Set[tuple[str, str]]]]: |
| 77 | for name in sorted(self.events.keys()): |
| 78 | yield name, self.events[name], self.sources[name] |
| 79 | |
| 80 | |
| 81 | def initialise_audit_events(app: Sphinx) -> None: |
no outgoing calls
no test coverage detected
searching dependent graphs…