snappea

217    def __init__(self, target: SinkTarget):
218        self.target = target
219        #: Events dropped: queue full, a write failed, or a payload could not be encoded.
220        self.dropped = 0
221        self._write = self._writer(target)
222        self._lock = threading.Lock()
223        self._pid = -1
224        self._closed = False
225        self._queue: "queue.Queue[Optional[str]]" = queue.Queue(self.QUEUE_SIZE)

290    def flush(self, timeout: float = 5.0) -> bool:
291        """Waits up to `timeout` seconds for queued events to be written. Returns
292        whether they were."""
293        deadline = time.monotonic() + timeout
294        while self._queue.unfinished_tasks and self._pid == os.getpid():
295            if time.monotonic() >= deadline:
296                return False
297            time.sleep(0.005)
298        return True

300    def close(self, timeout: float = 5.0) -> None:
301        """Writes what is queued and stops the writer. Later events are ignored."""
302        with self._lock:
303            if self._closed:
304                return
305            self._closed = True
306            if self._pid == os.getpid():
307                try:
308                    self._queue.put_nowait(None)
309                except queue.Full:
310                    pass
311        self.flush(timeout)

69    def __init__(
70        self,
71        *specs: str,
72        spec_files: Optional[Sequence[str]] = None,
73        check_hot: Optional[bool] = None,
74        strict: Optional[bool] = None,
75        unknown_events: Optional[str] = None,
76        timeout: Optional[float] = None,
77        artifacts: Optional[str] = None,
78        liveness: Optional[bool] = None,
79    ):
80        self.recorder = Recorder(
81            specs,
82            spec_files=spec_files,
83            check_hot=check_hot_arg(check_hot, liveness),
84            strict=strict,
85            unknown_events=unknown_events,
86            timeout=timeout,
87            artifacts=artifacts,
88        )

116    def __init__(
117        self,
118        spec: str,
119        match: Optional[str] = None,
120        *,
121        spec_files: Optional[Sequence[str]] = None,
122        check_hot: Optional[bool] = None,
123        unknown_events: Optional[str] = None,
124        timeout: Optional[float] = None,
125        artifacts: Optional[str] = None,
126        liveness: Optional[bool] = None,
127    ):
128        self.spec = spec
129        self.match = match
130        self.recorder = Recorder(
131            [spec],
132            spec_files=spec_files,
133            check_hot=check_hot_arg(check_hot, liveness),
134            strict=False,  # the spec must break; whether it saw events is beside the point
135            unknown_events=unknown_events,
136            timeout=timeout,
137            artifacts=artifacts,
138        )
139        self.violation: Optional[SpecViolation] = None

154    def __init__(
155        self,
156        specs: Iterable[str] = (),
157        *,
158        spec_files: Optional[Sequence[str]] = None,
159        exclude: Optional[Iterable[str]] = None,
160        strict: Optional[bool] = None,
161        check_hot: Optional[bool] = None,
162        unknown_events: Optional[str] = None,
163        timeout: Optional[float] = None,
164        artifacts: Optional[str] = None,
165        liveness: Optional[bool] = None,
166        output_root: Optional[str] = None,
167    ):
168        if isinstance(specs, str):
169            specs = [specs]
170        self.specs: List[str] = list(specs)
171        self.spec_files = [os.fspath(f) for f in spec_files] if spec_files is not None else None
172        self.exclude: Optional[List[str]] = list(exclude) if exclude is not None else None
173        self.check_hot = check_hot_arg(check_hot, liveness)
174        self.strict = strict
175        if unknown_events is not None:
176            _config.check_unknown_events(unknown_events, "unknown_events")
177        self.unknown_events = unknown_events
178        self.timeout = timeout
179        if output_root is not None:
180            _deprecated("output_root=", "artifacts=")
181            artifacts = artifacts or output_root
182        self.artifacts = os.fspath(artifacts) if artifacts is not None else None
183        #: snappea's JSON result of the last check that ran (``specs[].coverage`` holds
184        #: each spec's monitor coverage); None before, or if snappea wrote none.
185        self.report: Optional[dict] = None
186        #: Specs whose violations this recorder ignores (broken on purpose inside
187        #: `expect_violation`).
188        self.ignored: set = set()
189        self._events: List[Event] = []
190        self._logs: List[Tuple[str, Optional[LogParser]]] = []
191        self._dirs: List[str] = []
192        self._owned: List[str] = []
193        self._active = False
194        self._prev_env: Optional[str] = None
195        self._child_dir: Optional[str] = None
196        # set by the pytest plugin: its configuration, and pytest's temp folder
197        self._cfg: Optional[_config.Config] = None
198        self._default_artifacts: Optional[str] = None

200    @property
201    def liveness(self) -> Optional[bool]:
202        """Deprecated: use :attr:`check_hot`."""
203        _deprecated("Recorder.liveness", "Recorder.check_hot")
204        return self.check_hot

208    def start(self) -> "Recorder":
209        """Starts collecting: events emitted in this process, and in processes started
210        from now on (through the `SNAPPEA_EVENTS_DIR` environment variable)."""
211        if self._active:
212            return self
213        events_dir = self.events_dir()
214        self._prev_env = os.environ.get(_emit.EVENTS_DIR_ENV)
215        os.environ[_emit.EVENTS_DIR_ENV] = events_dir
216        _emit._recorders.append(self)
217        self._active = True
218        return self

220    def stop(self) -> None:
221        """Stops collecting. Events child processes wrote are passed on to the recorders
222        still active, so an enclosing test-wide check sees them too."""
223        if not self._active:
224            return
225        self._active = False
226        if self in _emit._recorders:
227            _emit._recorders.remove(self)
228        if self._prev_env is None:
229            os.environ.pop(_emit.EVENTS_DIR_ENV, None)
230        else:
231            os.environ[_emit.EVENTS_DIR_ENV] = self._prev_env
232        if _emit._recorders and self._child_dir and self._child_dir in self._owned:
233            for outer in _emit._recorders:
234                outer._dirs.append(self._child_dir)
235            # the outermost recorder deletes the folder when it is done
236            _emit._recorders[0]._owned.append(self._child_dir)
237            self._owned.remove(self._child_dir)

248    def events_dir(self) -> str:
249        """This recording's events folder: the `SNAPPEA_EVENTS_DIR` of processes it starts.
250        Created on first use; deleted after the check."""
251        if self._child_dir is None:
252            self._child_dir = tempfile.mkdtemp(prefix="snappea-events-")
253            self._owned.append(self._child_dir)
254            self._dirs.append(self._child_dir)
255        return self._child_dir

257    def env(self) -> Dict[str, str]:
258        """The current environment plus `SNAPPEA_EVENTS_DIR`, for starting a child process:
259        ``subprocess.run(cmd, env=rec.env())``."""
260        return {**os.environ, _emit.EVENTS_DIR_ENV: self.events_dir()}

262    def add_log(self, path: str, parse: Optional[LogParser] = None) -> None:
263        """Also check the events in the log file at `path`, read when the check runs.
264
265        ``parse`` turns one line into an event or ``None``: a dict like ``emit``'s
266        arguments, ``{"event": "eEnter", "key": "L1", "at": when, "client": "c1"}``
267        (``at`` is required so events can be merged by time). Without ``parse`` the file
268        must hold snappea's JSON lines."""
269        self._logs.append((os.path.abspath(os.fspath(path)), parse))

271    def add_events_dir(self, path: str) -> None:
272        """Also check the JSON-lines files in folder `path`, e.g. one mounted into a
273        container whose service writes events there. The folder must exist at check time."""
274        self._dirs.append(os.path.abspath(os.fspath(path)))

278    @property
279    def events(self) -> List[Event]:
280        """Every event collected so far: emitted here, by child processes and in logs,
281        in time order."""
282        evs = list(self._events) + [e for _, e in self._external_lines()]
283        evs.sort(key=lambda e: e.time_ns)
284        return evs

286    def events_named(self, name: str) -> List[Any]:
287        """The payloads of the events called `name`, in order."""
288        return [e.payload for e in self.events if e.event == name]

290    def count(self, name: str) -> int:
291        return sum(1 for e in self.events if e.event == name)

319    def check(self) -> None:
320        """Runs snappea over the events; raises SpecViolation if a spec was broken, and
321        SetupError if the check could not run."""
322        __tracebackhide__ = True
323        violations = self.violations()
324        if violations:
325            first = violations[0]
326            first.others = violations[1:]
327            first.args = (first.report(),)
328            raise first

330    def violations(self) -> List[SpecViolation]:
331        """Runs snappea over the events and returns the violations found (empty if none)."""
332        __tracebackhide__ = True
333        try:
334            settings = self._settings()
335            if not self._events and not self._logs and not any(os.path.getsize(p) for p in self._dir_files()):
336                # nothing happened: every spec still holds, unless strict wants events
337                if not settings.strict:
338                    return []
339            return self._run(settings)
340        finally:
341            self._cleanup_owned()

53    def to_json(self) -> str:
54        """The JSON line `snappea observe` reads, plus fields used for reports."""
55        d: dict = {
56            "key": self.key,
57            "timestamp": iso_timestamp(self.time_ns),
58            "event": self.event,
59            "payload": _plain(self.payload),
60            "src": self.src,
61            "seq": self.seq,
62            "pid": self.pid,
63        }
64        if self.at:
65            d["at"] = True
66        return json.dumps(d, default=_json_default, separators=(",", ":"))

68    @staticmethod
69    def from_json(d: dict) -> "Event":
70        return Event(
71            event=d.get("event", ""),
72            payload=d.get("payload"),
73            key=_key_str(d.get("key")),
74            time_ns=0,
75            at=bool(d.get("at")),
76            src=d.get("src"),
77            seq=d.get("seq", 0) or 0,
78            pid=d.get("pid", 0) or 0,
79        )

43    def __init__(
44        self,
45        spec: str,
46        message: str,
47        key: str = "",
48        events: Sequence[Event] = (),
49        *,
50        location: Optional[str] = None,
51        liveness: bool = False,
52        earlier: int = 0,
53        output_dir: Optional[str] = None,
54        others: Sequence["SpecViolation"] = (),
55    ):
56        #: Name of the broken spec.
57        self.spec = spec
58        #: The spec's assert message (or, for liveness, which hot state it ended in).
59        self.message = message
60        #: Event key (stream) the violation happened on.
61        self.key = key
62        #: Events of that key up to and including the violating one (the last ones only).
63        self.events = list(events)
64        #: `file:line` of the failed assert in the P sources.
65        self.location = location
66        #: The spec ended in a hot state (found with `check_hot`).
67        self.liveness = liveness
68        #: Events of this key before `events` that are not shown.
69        self.earlier = earlier
70        #: Folder with the events and snappea's full output (the check's artifacts).
71        self.output_dir = output_dir
72        #: Further violations found by the same check.
73        self.others = list(others)
74        super().__init__(self.report())

 76    def report(self) -> str:
 77        lines = []
 78        if self.liveness:
 79            lines.append(f"P spec {self.spec} was still waiting when the events ended: {self.message}")
 80        else:
 81            lines.append(f"P spec {self.spec} was broken: {self.message}")
 82        lines.append("")
 83        lines.append(f'  Events (key "{self.key}"):' if self.key else "  Events:")
 84        if self.earlier:
 85            lines.append(f"     ... {self.earlier} earlier event(s)")
 86        rows = [(_payload_text(e.payload), e) for e in self.events]
 87        w_event = max([len(e.event) for _, e in rows] + [0])
 88        w_payload = min(max([len(p) for p, _ in rows] + [0]), 48)
 89        for i, (p, e) in enumerate(rows):
 90            n = self.earlier + i + 1
 91            row = f"    {n:>3}  {e.event:<{w_event}}  {p:<{w_payload}}  {e.src or ''}"
 92            if e.at:
 93                row += "  (time set with at=)"
 94            if i == len(rows) - 1:
 95                row += "  <- events ended here" if self.liveness else "  <- broke the spec here"
 96            lines.append(row.rstrip() if i < len(rows) - 1 else row)
 97        if not rows:
 98            lines.append("    (none)")
 99        lines.append("")
100        if self.location:
101            lines.append(f"  Spec: {self.location}")
102        if self.output_dir:
103            lines.append(f"  snappea output: {self.output_dir}")
104        if self.others:
105            lines.append("")
106            lines.append(f"  {len(self.others)} more violation(s):")
107            for o in self.others:
108                key = f' (key "{o.key}")' if o.key else ""
109                lines.append(f"    {o.spec}{key}: {o.message}")
110        return "\n".join(lines)

30    def __init__(self, message: str, kind: str = "config", *, output_dir: Optional[str] = None):
31        super().__init__(message)
32        #: One of :data:`KINDS` (or a newer kind the core reports).
33        self.kind = kind
34        #: The error text.
35        self.message = message
36        #: Folder with the events and snappea's output, if the check got that far.
37        self.output_dir = output_dir

49    def __init__(self, error: SetupError):
50        super().__init__(error.message)
51        self.error = error