1"""
2Helpers for writing pytest tests over SysMLv2/KerML models.
3
4This module provides small, dependency-light utilities for asserting
5properties about models loaded with :mod:`syside`:
6
7* :func:`describe` renders a human-readable identifier for a model element,
8 used in assertion failure messages.
9* :func:`assert_all`, :func:`assert_any` and :func:`assert_none` assert
10 quantified predicates over an iterable of elements and raise
11 ``AssertionError`` with a clear, multi-line message that names the
12 offending elements.
13* :func:`model_fixture` builds a pytest fixture that loads a model once per
14 scope (session by default), so a test suite does not reload it per test.
15
16The assertion helpers consume their input iterable exactly once and call the
17``describe`` callback only for the (capped) set of elements that actually
18appear in a failure message, so they stay cheap on large models and on the
19success path.
20"""
21
22from __future__ import annotations
23
24import os
25from collections.abc import Callable, Hashable, Iterable, Iterator
26from typing import Literal
27
28import syside
29
30__all__ = [
31 "describe",
32 "assert_all",
33 "assert_any",
34 "assert_none",
35 "assert_implies",
36 "assert_foreach_unique",
37 "assert_acyclic",
38 "assert_reachable",
39 "model_fixture",
40]
41
42# Pytest fixture scopes, narrowed from pytest's own ``_ScopeName``.
43Scope = Literal["session", "package", "module", "class", "function"]
44
45
[docs]
46def describe(item: object) -> str:
47 """
48 Return a default human-readable identifier for a model element.
49
50 If ``item`` has a non-``None`` ``qualified_name``, return
51 ``f"{type(item).__name__} {item.qualified_name}"`` (the qualified name
52 renders to e.g. ``Pkg::Sub::Name``). Otherwise, or if attribute access
53 raises, fall back to ``repr(item)``.
54
55 This function never raises.
56 """
57 try:
58 qualified_name = item.qualified_name # type: ignore[attr-defined]
59 if qualified_name is not None:
60 return f"{type(item).__name__} {qualified_name}"
61 except Exception:
62 pass
63 return repr(item)
64
65
66def _label_of[T](predicate: Callable[[T], object], label: str | None) -> str:
67 if label is not None:
68 return label
69 name = getattr(predicate, "__name__", None)
70 return name if isinstance(name, str) else repr(predicate)
71
72
73def _render[T](
74 offenders: list[T],
75 total_reported: int,
76 describe: Callable[[T], str],
77) -> str:
78 lines = [f" - {describe(item)}" for item in offenders]
79 remaining = total_reported - len(offenders)
80 if remaining > 0:
81 lines.append(f" … and {remaining} more")
82 return "\n".join(lines)
83
84
[docs]
85def assert_all[T](
86 items: Iterable[T],
87 predicate: Callable[[T], object],
88 *,
89 describe: Callable[[T], str] = describe,
90 label: str | None = None,
91 limit: int = 10,
92) -> None:
93 """
94 Assert that ``predicate`` is truthy for every item in ``items``.
95
96 Passes vacuously when ``items`` is empty. On failure, raises
97 ``AssertionError`` naming the items for which ``predicate`` was falsy
98 (the counterexamples), listing at most ``limit`` of them.
99
100 ``predicate`` is called exactly once per item. ``describe`` is called
101 only on the items that appear in the failure message.
102 """
103 total = 0
104 failed_count = 0
105 offenders: list[T] = []
106 for item in items:
107 total += 1
108 if not predicate(item):
109 failed_count += 1
110 if len(offenders) < limit:
111 offenders.append(item)
112 if failed_count == 0:
113 return
114 name = _label_of(predicate, label)
115 body = _render(offenders, failed_count, describe)
116 raise AssertionError(
117 f"assert_all {name}: {failed_count} of {total} items failed:\n{body}"
118 )
119
120
[docs]
121def assert_any[T](
122 items: Iterable[T],
123 predicate: Callable[[T], object],
124 *,
125 describe: Callable[[T], str] = describe,
126 label: str | None = None,
127 limit: int = 10,
128) -> None:
129 """
130 Assert that ``predicate`` is truthy for at least one item in ``items``.
131
132 Fails when ``items`` is empty. On failure, raises ``AssertionError``
133 stating that none of the examined items satisfied the predicate, listing
134 at most ``limit`` of the examined items.
135
136 ``predicate`` is called exactly once per item, and short-circuits as soon
137 as a truthy result is found. ``describe`` is called only on the items that
138 appear in the failure message.
139 """
140 total = 0
141 examined: list[T] = []
142 for item in items:
143 total += 1
144 if predicate(item):
145 return
146 if len(examined) < limit:
147 examined.append(item)
148 name = _label_of(predicate, label)
149 body = _render(examined, total, describe)
150 raise AssertionError(
151 f"assert_any {name}: none of {total} items satisfied it:\n{body}"
152 )
153
154
[docs]
155def assert_none[T](
156 items: Iterable[T],
157 predicate: Callable[[T], object],
158 *,
159 describe: Callable[[T], str] = describe,
160 label: str | None = None,
161 limit: int = 10,
162) -> None:
163 """
164 Assert that ``predicate`` is truthy for no item in ``items``.
165
166 Passes vacuously when ``items`` is empty. On failure, raises
167 ``AssertionError`` naming the items for which ``predicate`` was truthy,
168 listing at most ``limit`` of them.
169
170 ``predicate`` is called exactly once per item. ``describe`` is called
171 only on the items that appear in the failure message.
172 """
173 total = 0
174 matched_count = 0
175 offenders: list[T] = []
176 for item in items:
177 total += 1
178 if predicate(item):
179 matched_count += 1
180 if len(offenders) < limit:
181 offenders.append(item)
182 if matched_count == 0:
183 return
184 name = _label_of(predicate, label)
185 body = _render(offenders, matched_count, describe)
186 raise AssertionError(
187 f"assert_none {name}: {matched_count} of {total} items matched:\n{body}"
188 )
189
190
[docs]
191def assert_implies[T](
192 items: Iterable[T],
193 when: Callable[[T], object],
194 then: Callable[[T], object],
195 *,
196 describe: Callable[[T], str] = describe,
197 label: str | None = None,
198 limit: int = 10,
199) -> None:
200 """
201 Assert that ``when(item)`` truthy implies ``then(item)`` truthy.
202
203 For every item, if ``when`` is truthy then ``then`` must be truthy too.
204 Passes vacuously when ``items`` is empty (or no item satisfies ``when``).
205 On failure, raises ``AssertionError`` naming the items for which the
206 antecedent held but the consequent failed (the counterexamples), listing
207 at most ``limit`` of them.
208
209 ``when`` is called exactly once per item; ``then`` is called only for items
210 where ``when`` is truthy (short-circuited otherwise). ``describe`` is called
211 only on the items that appear in the failure message.
212 """
213 antecedent_count = 0
214 failed_count = 0
215 offenders: list[T] = []
216 for item in items:
217 if not when(item):
218 continue
219 antecedent_count += 1
220 if not then(item):
221 failed_count += 1
222 if len(offenders) < limit:
223 offenders.append(item)
224 if failed_count == 0:
225 return
226 if label is not None:
227 name = label
228 else:
229 name = f"{_callable_name(when)} ⇒ {_callable_name(then)}"
230 body = _render(offenders, failed_count, describe)
231 raise AssertionError(
232 f"assert_implies {name}: {failed_count} of {antecedent_count} items "
233 f"satisfying the antecedent failed the consequent:\n{body}"
234 )
235
236
[docs]
237def assert_foreach_unique[T](
238 items: Iterable[T],
239 target: Callable[[T], Hashable],
240 *,
241 describe: Callable[[T], str] = describe,
242 label: str | None = None,
243 limit: int = 10,
244) -> None:
245 """
246 Assert that ``target`` yields a distinct value for each item.
247
248 That is, ``target`` is injective over ``items``. Passes when ``items`` is
249 empty or all target values are distinct. On failure, raises
250 ``AssertionError`` naming the target values shared by more than one item,
251 listing at most ``limit`` colliding values; for each, the items that share
252 it are listed (also capped at ``limit``, with a ``…`` suffix).
253
254 ``target`` is called exactly once per item. ``describe`` is called only on
255 the items that appear in the reported collisions.
256 """
257 groups: dict[Hashable, list[T]] = {}
258 for item in items:
259 groups.setdefault(target(item), []).append(item)
260 collisions = [(value, group) for value, group in groups.items() if len(group) > 1]
261 if not collisions:
262 return
263 name = label if label is not None else _callable_name(target, fallback="target")
264 lines: list[str] = []
265 for value, group in collisions[:limit]:
266 shown = group[:limit]
267 rendered = ", ".join(describe(item) for item in shown)
268 if len(group) > len(shown):
269 rendered += ", …"
270 lines.append(f" - {value!r}: {rendered}")
271 remaining = len(collisions) - min(len(collisions), limit)
272 if remaining > 0:
273 lines.append(f" … and {remaining} more")
274 body = "\n".join(lines)
275 raise AssertionError(
276 f"assert_foreach_unique {name}: {len(collisions)} value(s) shared by "
277 f"multiple items:\n{body}"
278 )
279
280
[docs]
281def assert_acyclic[T](
282 nodes: Iterable[T],
283 successors: Callable[[T], Iterable[T]],
284 *,
285 describe: Callable[[T], str] = describe,
286 label: str | None = None,
287) -> None:
288 """
289 Assert that the directed graph over ``nodes`` has no directed cycle.
290
291 The graph has ``nodes`` as its (initial) vertices; any node returned by
292 ``successors`` is also part of the graph even if it is not in ``nodes``. A
293 node listed among its own ``successors`` is a length-1 cycle.
294
295 Traversal is an iterative depth-first search with an explicit stack (never
296 recursion), so graphs thousands of nodes deep do not hit the Python
297 recursion limit. Passes when ``nodes`` is empty or the graph is acyclic. On
298 failure, raises ``AssertionError`` naming one concrete cycle as a path.
299
300 ``describe`` is called only on the nodes of the reported cycle.
301 """
302 # Three-state marking by object identity: unvisited (absent),
303 # on-stack (False), done (True).
304 state: dict[int, bool] = {}
305 keep_alive: dict[int, T] = {}
306
307 for root in nodes:
308 root_id = id(root)
309 if state.get(root_id) is True:
310 continue
311 # Each stack frame: (node, node_id, iterator over its successors).
312 stack: list[tuple[T, int, Iterator[T]]] = [
313 (root, root_id, iter(successors(root)))
314 ]
315 state[root_id] = False
316 keep_alive[root_id] = root
317 path: list[T] = [root]
318 while stack:
319 _node, _node_id, children = stack[-1]
320 advanced = False
321 for child in children:
322 child_id = id(child)
323 marking = state.get(child_id)
324 if marking is False:
325 # Back edge to an on-stack node: cycle found.
326 cycle = _extract_cycle(path, child)
327 rendered = " → ".join(describe(n) for n in cycle)
328 name = label if label is not None else "graph"
329 raise AssertionError(
330 f"assert_acyclic {name}: cycle found:\n {rendered}"
331 )
332 if marking is True:
333 continue
334 state[child_id] = False
335 keep_alive[child_id] = child
336 stack.append((child, child_id, iter(successors(child))))
337 path.append(child)
338 advanced = True
339 break
340 if not advanced:
341 state[_node_id] = True
342 stack.pop()
343 path.pop()
344
345
[docs]
346def assert_reachable[T](
347 sources: Iterable[T],
348 targets: Iterable[T],
349 successors: Callable[[T], Iterable[T]],
350 *,
351 describe: Callable[[T], str] = describe,
352 label: str | None = None,
353 limit: int = 10,
354) -> None:
355 """
356 Assert that every source can reach at least one target.
357
358 Reachability follows ``successors`` forward. A source that is itself a
359 target reaches trivially (distance 0). Passes when ``sources`` is empty.
360 When ``sources`` is non-empty but ``targets`` is empty, every source fails.
361
362 On failure, raises ``AssertionError`` naming the sources that cannot reach
363 any target (the counterexamples), listing at most ``limit`` of them.
364
365 ``describe`` is called only on the unreachable sources that appear in the
366 failure message.
367 """
368 target_ids = {id(target) for target in targets}
369 total = 0
370 failed_count = 0
371 offenders: list[T] = []
372 for source in sources:
373 total += 1
374 if not _can_reach(source, target_ids, successors):
375 failed_count += 1
376 if len(offenders) < limit:
377 offenders.append(source)
378 if failed_count == 0:
379 return
380 name = label if label is not None else "reachable"
381 body = _render(offenders, failed_count, describe)
382 raise AssertionError(
383 f"assert_reachable {name}: {failed_count} of {total} sources cannot "
384 f"reach any target:\n{body}"
385 )
386
387
388def _can_reach[T](
389 source: T,
390 target_ids: set[int],
391 successors: Callable[[T], Iterable[T]],
392) -> bool:
393 """Return whether ``source`` reaches any node whose id is in ``target_ids``."""
394 visited: set[int] = {id(source)}
395 stack: list[T] = [source]
396 while stack:
397 node = stack.pop()
398 if id(node) in target_ids:
399 return True
400 for child in successors(node):
401 child_id = id(child)
402 if child_id not in visited:
403 visited.add(child_id)
404 stack.append(child)
405 return False
406
407
408def _extract_cycle[T](path: list[T], back_to: T) -> list[T]:
409 """
410 Return the cycle in ``path`` closing on ``back_to``.
411
412 ``path`` is the current DFS stack of nodes; ``back_to`` is an on-stack node
413 reached by a back edge. The returned list starts and ends at ``back_to``.
414 """
415 back_id = id(back_to)
416 start = next(i for i, n in enumerate(path) if id(n) == back_id)
417 return [*path[start:], back_to]
418
419
420def _callable_name(fn: Callable[..., object], *, fallback: str | None = None) -> str:
421 """Return ``fn.__name__`` if a string, else ``fallback`` or ``repr(fn)``."""
422 name = getattr(fn, "__name__", None)
423 if isinstance(name, str):
424 return name
425 return fallback if fallback is not None else repr(fn)
426
427
[docs]
428def model_fixture(
429 *paths: str | os.PathLike[str],
430 sysml_source: str | None = None,
431 kerml_source: str | None = None,
432 scope: Scope = "session",
433 warnings_as_errors: bool = False,
434) -> Callable[[], syside.Model]:
435 """
436 Build a pytest fixture that loads a model once per scope.
437
438 Provide the model through exactly one of :func:`syside.load_model`'s
439 entrypoints -- there is no guessing between a path and inline text:
440
441 * positional ``paths`` -- one or more file or directory paths, or
442 * ``sysml_source`` -- a single inline SysML v2 source string, or
443 * ``kerml_source`` -- a single inline KerML source string.
444
445 Passing none, or more than one, of these raises :class:`ValueError`
446 immediately (when the fixture is defined, not when a test runs).
447
448 The fixture loads the model once per ``scope`` (``"session"`` by default,
449 so the suite does not reload it per test) and returns the loaded
450 :class:`syside.Model`. If loading fails to compile -- either because
451 :func:`syside.load_model` raises :class:`syside.ModelError`, or because it
452 returns diagnostics for which ``diagnostics.contains_errors()`` is true --
453 the fixture fails with an ``AssertionError`` describing the diagnostics.
454 With ``warnings_as_errors=True``, diagnostics containing any warnings fail
455 the fixture the same way.
456
457 Intended use in a ``conftest.py``::
458
459 import syside.testing
460
461 model = syside.testing.model_fixture("path/to/models")
462
463 or with inline source for a focused test::
464
465 model = syside.testing.model_fixture(sysml_source="package P { part def A; }")
466
467 The returned object is a pytest fixture; reference it by the name you bind
468 it to (``model`` above) as a test argument.
469 """
470 import pytest
471
472 chosen = [
473 name
474 for name, given in (
475 ("paths", bool(paths)),
476 ("sysml_source", sysml_source is not None),
477 ("kerml_source", kerml_source is not None),
478 )
479 if given
480 ]
481 if len(chosen) != 1:
482 raise ValueError(
483 "model_fixture takes exactly one of: positional path(s), "
484 f"sysml_source=, or kerml_source= (got: {', '.join(chosen) or 'none'})"
485 )
486
487 @pytest.fixture(scope=scope)
488 def _model() -> syside.Model:
489 try:
490 # Exactly one entrypoint is chosen (validated above).
491 if paths:
492 model, diagnostics = syside.load_model(paths=list(paths))
493 elif sysml_source is not None:
494 model, diagnostics = syside.load_model(sysml_source=sysml_source)
495 else:
496 assert kerml_source is not None # the only remaining entrypoint
497 model, diagnostics = syside.load_model(kerml_source=kerml_source)
498 except syside.ModelError as error:
499 raise AssertionError(
500 f"model_fixture failed to load model: {error}"
501 ) from error
502 if diagnostics.contains_errors():
503 raise AssertionError(f"model_fixture failed to load model: {diagnostics}")
504 if warnings_as_errors and any(True for _ in diagnostics.warnings):
505 raise AssertionError(
506 "model_fixture loaded the model with warnings "
507 f"(warnings_as_errors=True): {diagnostics}"
508 )
509 return model
510
511 return _model
