pyrtan / sourcev0.19.0 · main

pyrtan / src / pyrtan / adapters.py

35,004 bytes · last changed 2026-10-26 in Watches and growing from a gap: trees that run when something speaks, and the branch a gap asks for drafted by rules; release 0.9.0

1"""The other forms of I/O: a browser, a file, a shell, a serial line, a queue.
2
3HTTP is the form the core spoke first (``transport.py``). Every other form
4arrives here by the same rule: a node names a *source* and what it wants
5from it, the tree lists what that source may reach, the runtime refuses
6anything outside the list, and the adapter moves the bytes. The trace reads
7the same whichever source a node used.
8
9 [nodes.df] observe shell = { command = "df", args = ["-P", "/"] }
10 [nodes.log] observe file = { path = "/var/log/syslog", tail = 200 }
11 [nodes.note] act file = { path = "./notes.txt", append = "..." }
12 [nodes.modem] observe serial = { device = "/dev/ttyUSB0", baud = 9600, write = "AT\\r\\n", read_until = "\\n" }
13 [nodes.sensor] observe queue = { url = "mqtt://broker.local", topic = "home/kitchen/temperature" }
14 [nodes.switch] act queue = { url = "mqtt://broker.local", topic = "home/kitchen/light", publish = "on" }
15 [nodes.page] observe browser = { url = "https://example.com/app", wait_for = "h1", actions = [{ click = "#more" }] }
16
17and at the top of the tree the allow-list of the same kind: ``hosts`` for
18``http``, ``browser`` and a queue's broker; ``paths`` for ``file``;
19``commands`` for ``shell``; ``devices`` for ``serial``; ``topics`` for
20``queue``. A tree without the list cannot use the source; ``pyrtan check``
21says so.
22
23What each source leaves in state (the node's ``into``):
24
25``file`` ``path``, ``exists``, ``size``, ``modified``, ``body``, ``json``, ``lines``;
26 after a write, ``written`` as well.
27``shell`` ``command``, ``args``, ``status`` (the exit code), ``ok``, ``stdout``,
28 ``stderr``, ``body`` (stdout again, so ``extract`` reads it by
29 default), ``json`` (stdout parsed), ``lines``.
30``serial`` ``device``, ``baud``, ``sent``, ``body``, ``bytes``, ``complete``, ``json``, ``lines``.
31``queue`` ``url``, ``topic``, ``received``, ``message`` (``topic`` and ``body``),
32 ``body``, ``json``; after a publish, ``published``.
33``browser`` ``url`` (where the page ended up), ``status``, ``ok``, ``title``,
34 ``body`` (the rendered document), ``elapsed_ms``.
35
36Files, shells and serial lines are on the box the service runs on. A
37service started with ``--egress public`` refuses them outright, the way it
38refuses private addresses: an operator hosting other people's trees does
39not lend them the box. A queue's broker and a browser's page go through the
40same egress check as an HTTP request.
41
42The browser adapter needs Playwright (``pip install 'pyrtan[browser]'``),
43the one optional dependency; everything else here is the standard library.
44A queue is MQTT 3.1.1 over a socket, the few packets a tree needs: connect,
45subscribe and wait for one message, or publish one; QoS 0.
46
47:class:`LocalAdapters` is the real thing. A recorded fixture (see
48``transport.RecordedTransport``) answers for every source as well, keyed by
49:func:`adapter_key`, so a tree over any of them is tested offline the same
50way an HTTP tree is.
51"""
52
53from __future__ import annotations
54
55import datetime as _dt
56import fnmatch
57import json
58import os
59import queue as _queue
60import select
61import socket
62import ssl
63import struct
64import subprocess
65import threading
66import time
67import urllib.parse
68from typing import Any, Callable
69
70from .transport import EgressRefused, TransportError, check_egress
71
72SOURCES = ("http", "browser", "file", "shell", "serial", "queue")
73LOCAL_SOURCES = ("file", "shell", "serial") # on the box itself
74ALLOW_LISTS = {"http": ("hosts",), "browser": ("hosts",), "file": ("paths",), "shell": ("commands",), "serial": ("devices",), "queue": ("hosts", "topics")}
75LIST_NAMES = ("hosts", "paths", "commands", "devices", "topics")
76MAX_READ = 8 * 1024 * 1024 # a file or a command's output larger than this is refused
77TAIL_WINDOW = 1024 * 1024 # how far back a tail looks
78MQTT_PORTS = {"mqtt": 1883, "mqtts": 8883}
79BROWSER_ACTIONS = ("click", "fill", "press", "select", "wait", "wait_for")
80
81Allowed = Callable[[str], bool]
82
83
84# ---------------------------------------------------------------- allow-lists
85
86
87def path_allowed(patterns: list[str], path: str) -> bool:
88 """``paths`` entries are absolute paths or globs; a directory covers
89 everything beneath it; ``*`` covers everything."""
90 target = absolute_path(path)
91 for pat in patterns:
92 if pat == "*":
93 return True
94 pat = absolute_path(pat)
95 if any(c in pat for c in "*?["):
96 if fnmatch.fnmatchcase(target, pat):
97 return True
98 elif target == pat or target.startswith(pat.rstrip("/") + "/"):
99 return True
100 return False
101
102
103def absolute_path(path: str) -> str:
104 return os.path.abspath(os.path.expanduser(str(path)))
105
106
107def command_allowed(commands: list[str], command: str) -> bool:
108 """``commands`` entries are program names (matched against the name the
109 node gives, or the last part of a path it gives) or full paths."""
110 command = str(command)
111 for entry in commands:
112 if entry == "*":
113 return True
114 if "/" in entry:
115 if absolute_path(command) == absolute_path(entry):
116 return True
117 elif command == entry or os.path.basename(command) == entry:
118 return True
119 return False
120
121
122def device_allowed(devices: list[str], device: str) -> bool:
123 device = str(device)
124 return any(d == "*" or fnmatch.fnmatchcase(device, d) for d in devices)
125
126
127def topic_allowed(topics: list[str], topic: str) -> bool:
128 """``topics`` entries are MQTT filters: ``+`` matches one level (so does
129 ``*``, for people who think in globs), ``#`` the rest."""
130 parts = str(topic).split("/")
131 for entry in topics:
132 if entry in ("*", "#"):
133 return True
134 pat = entry.split("/")
135 ok = True
136 for i, seg in enumerate(pat):
137 if seg == "#":
138 ok = ok and i <= len(parts)
139 break
140 if i >= len(parts) or (seg not in ("+", "*") and seg != parts[i]):
141 ok = False
142 break
143 else:
144 ok = ok and len(pat) == len(parts)
145 if ok:
146 return True
147 return False
148
149
150def broker(url: str) -> tuple[str, str, int]:
151 """``(scheme, host, port)`` of a queue URL, or a :class:`ValueError`."""
152 parsed = urllib.parse.urlsplit(str(url))
153 if parsed.scheme not in MQTT_PORTS or not parsed.hostname:
154 raise ValueError(f"{url!r} is not mqtt://host[:port] or mqtts://host[:port]")
155 return parsed.scheme, parsed.hostname, parsed.port or MQTT_PORTS[parsed.scheme]
156
157
158# ---------------------------------------------------------------- keys and words
159
160
161def adapter_key(source: str, spec: dict[str, Any]) -> str:
162 """How a fixture names one request to a source. The JavaScript runtime
163 builds the same key, so one fixture serves both."""
164 if source == "file":
165 mode = "write " if "write" in spec else "append " if "append" in spec else ""
166 return f"file {mode}{spec.get('path')}"
167 if source == "shell":
168 return "shell " + " ".join([str(spec.get("command"))] + [_arg(a) for a in (spec.get("args") or [])])
169 if source == "serial":
170 write = spec.get("write")
171 return f"serial {spec.get('device')}" + (f" {json.dumps(str(write))}" if write not in (None, "") else "")
172 if source == "queue":
173 mode = "publish " if "publish" in spec else ""
174 return f"queue {mode}{spec.get('url')} {spec.get('topic')}"
175 if source == "browser":
176 return f"browser {spec.get('url')}"
177 raise ValueError(source)
178
179
180def _arg(a: Any) -> str:
181 return a if isinstance(a, str) else json.dumps(a)
182
183
184def what(source: str, spec: dict[str, Any]) -> str:
185 """The words a trace uses for one request: what was reached."""
186 if source == "file":
187 return str(spec.get("path"))
188 if source == "shell":
189 return " ".join([str(spec.get("command"))] + [_arg(a) for a in (spec.get("args") or [])])
190 if source == "serial":
191 return str(spec.get("device"))
192 if source == "queue":
193 return f"{spec.get('topic')} at {spec.get('url')}"
194 return str(spec.get("url"))
195
196
197def brief_result(source: str, observed: dict[str, Any]) -> str:
198 """The words a trace uses for what came back."""
199 if source in ("http", "browser"):
200 return str(observed.get("status"))
201 if source == "file":
202 if "written" in observed:
203 return f"wrote {observed['written']} B"
204 return f"{observed.get('size')} B" if observed.get("exists") else "missing"
205 if source == "shell":
206 return f"exit {observed.get('status')}"
207 if source == "serial":
208 return f"{observed.get('bytes')} B"
209 if source == "queue":
210 if "published" in observed:
211 return f"published {observed['published']} B"
212 return "1 message" if observed.get("received") else "no message"
213 return "ok"
214
215
216def _json(text: str | None) -> Any:
217 if not text:
218 return None
219 try:
220 return json.loads(text)
221 except (json.JSONDecodeError, TypeError):
222 return None
223
224
225def _iso(ts: float) -> str:
226 return _dt.datetime.fromtimestamp(ts, _dt.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
227
228
229# ---------------------------------------------------------------- the real adapters
230
231
232class LocalAdapters:
233 """The sources as this box reaches them. ``egress`` is the service's
234 policy; ``browser`` an executable for Playwright's Chromium, if not the
235 one Playwright installed."""
236
237 def __init__(self, egress: str = "any", browser: str | None = None) -> None:
238 self.egress = egress
239 self._browser: Browser | None = None
240 self._browser_path = browser or os.environ.get("PYRTAN_BROWSER") or None
241 self._lock = threading.Lock()
242
243 def perform(self, source: str, spec: dict[str, Any], timeout: float, allowed: Allowed | None = None) -> dict[str, Any]:
244 if source in LOCAL_SOURCES and self.egress != "any":
245 raise EgressRefused(f"{source}: this service reaches only public addresses (--egress {self.egress}); files, shells and serial lines are its own box, and stay closed")
246 if source == "file":
247 return read_or_write_file(spec)
248 if source == "shell":
249 return run_command(spec, timeout)
250 if source == "serial":
251 return serial_exchange(spec, timeout)
252 if source == "queue":
253 return mqtt(spec, timeout, self.egress)
254 if source == "browser":
255 with self._lock:
256 if self._browser is None:
257 self._browser = Browser(self._browser_path)
258 return self._browser.perform(spec, timeout, allowed or (lambda url: True), self.egress)
259 raise TransportError(f"no adapter for {source!r}")
260
261 def close(self) -> None:
262 if self._browser is not None:
263 self._browser.close()
264
265
266# ---------------------------------------------------------------- file
267
268
269def read_or_write_file(spec: dict[str, Any]) -> dict[str, Any]:
270 path = absolute_path(spec.get("path", ""))
271 if "write" in spec or "append" in spec:
272 data = spec["write"] if "write" in spec else spec["append"]
273 if not isinstance(data, str):
274 data = json.dumps(data, ensure_ascii=False) + "\n"
275 if not os.path.isdir(os.path.dirname(path) or "."):
276 raise TransportError(f"{path}: the directory does not exist; a tree writes files, it does not make directories")
277 try:
278 with open(path, "a" if "append" in spec else "w", encoding="utf-8") as f:
279 f.write(data)
280 except OSError as e:
281 raise TransportError(f"{path}: {e.strerror or e}") from e
282 st = os.stat(path)
283 return {"path": path, "exists": True, "written": len(data.encode("utf-8")), "size": st.st_size, "modified": _iso(st.st_mtime)}
284 if not os.path.exists(path):
285 return {"path": path, "exists": False, "size": None, "modified": None, "body": None, "json": None, "lines": []}
286 try:
287 st = os.stat(path)
288 tail = spec.get("tail")
289 if tail is not None:
290 tail = int(tail)
291 with open(path, "rb") as f:
292 f.seek(max(0, st.st_size - TAIL_WINDOW))
293 raw = f.read()
294 lines = raw.decode("utf-8", errors="replace").splitlines()
295 if st.st_size > TAIL_WINDOW and lines:
296 lines = lines[1:] # the first line is a fragment
297 lines = lines[-tail:] if tail > 0 else []
298 text = "\n".join(lines)
299 else:
300 if st.st_size > MAX_READ:
301 raise TransportError(f"{path}: {st.st_size} bytes is more than a tree reads at once ({MAX_READ}); give 'tail'")
302 with open(path, "rb") as f:
303 text = f.read().decode("utf-8", errors="replace")
304 lines = text.splitlines()
305 except OSError as e:
306 raise TransportError(f"{path}: {e.strerror or e}") from e
307 return {"path": path, "exists": True, "size": st.st_size, "modified": _iso(st.st_mtime), "body": text, "json": _json(text), "lines": lines}
308
309
310# ---------------------------------------------------------------- shell
311
312
313def run_command(spec: dict[str, Any], timeout: float) -> dict[str, Any]:
314 """One program, by name, with its arguments as a list: no shell reads
315 the line, so nothing in an argument is ever a command."""
316 argv = [str(spec.get("command", ""))] + [_arg(a) for a in (spec.get("args") or [])]
317 stdin = spec.get("stdin")
318 if stdin is not None and not isinstance(stdin, str):
319 stdin = json.dumps(stdin)
320 env = dict(os.environ)
321 env.update({str(k): str(v) for k, v in (spec.get("env") or {}).items()})
322 cwd = spec.get("cwd")
323 started = time.monotonic()
324 try:
325 p = subprocess.run(argv, input=stdin, capture_output=True, text=True, timeout=timeout, cwd=str(cwd) if cwd else None, env=env, shell=False, errors="replace")
326 except FileNotFoundError:
327 raise TransportError(f"{argv[0]}: not found on this box") from None
328 except subprocess.TimeoutExpired:
329 raise TransportError(f"{argv[0]}: no answer within {timeout:g} s") from None
330 except OSError as e:
331 raise TransportError(f"{argv[0]}: {e.strerror or e}") from e
332 out, err = p.stdout[:MAX_READ], p.stderr[:MAX_READ]
333 return {"command": argv[0], "args": argv[1:], "status": p.returncode, "ok": p.returncode == 0, "stdout": out, "stderr": err, "body": out, "json": _json(out), "lines": out.splitlines(), "elapsed_ms": int((time.monotonic() - started) * 1000)}
334
335
336# ---------------------------------------------------------------- serial
337
338
339def serial_exchange(spec: dict[str, Any], timeout: float) -> dict[str, Any]:
340 """Open a serial device at a baud rate, write what the node says, read
341 until a terminator, a byte count or the timeout."""
342 try:
343 import termios
344 except ImportError: # pragma: no cover - Windows
345 raise TransportError("serial lines need a POSIX system") from None
346 device = str(spec.get("device", ""))
347 baud = int(spec.get("baud", 9600))
348 write = spec.get("write")
349 until = spec.get("read_until", "\n" if spec.get("read") is None else None)
350 count = int(spec.get("read") or 0)
351 speed = getattr(termios, f"B{baud}", None)
352 if speed is None:
353 raise TransportError(f"{device}: {baud} baud is not a rate this system knows")
354 try:
355 fd = os.open(device, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
356 except OSError as e:
357 raise TransportError(f"{device}: {e.strerror or e}") from e
358 sent = 0
359 buf = b""
360 try:
361 attrs = termios.tcgetattr(fd)
362 attrs[0] &= ~(termios.IXON | termios.IXOFF | termios.ICRNL | termios.INLCR | termios.IGNCR | termios.ISTRIP) # iflag
363 attrs[1] &= ~termios.OPOST # oflag
364 attrs[2] = (attrs[2] & ~(termios.PARENB | termios.CSTOPB | termios.CSIZE)) | termios.CS8 | termios.CLOCAL | termios.CREAD # cflag
365 attrs[3] &= ~(termios.ICANON | termios.ECHO | termios.ECHOE | termios.ISIG | termios.IEXTEN) # lflag
366 attrs[4] = attrs[5] = speed
367 termios.tcsetattr(fd, termios.TCSANOW, attrs)
368 if write not in (None, ""):
369 data = str(write).encode("utf-8")
370 while sent < len(data):
371 _, ready, _ = select.select([], [fd], [], timeout)
372 if not ready:
373 raise TransportError(f"{device}: could not write within {timeout:g} s")
374 sent += os.write(fd, data[sent:])
375 deadline = time.monotonic() + timeout
376 terminator = str(until).encode("utf-8") if until else None
377 complete = False
378 while True:
379 remaining = deadline - time.monotonic()
380 if remaining <= 0:
381 break
382 ready, _, _ = select.select([fd], [], [], remaining)
383 if not ready:
384 break
385 try:
386 chunk = os.read(fd, 4096)
387 except BlockingIOError:
388 continue
389 except OSError as e:
390 if buf:
391 break
392 raise TransportError(f"{device}: {e.strerror or e}") from e
393 if not chunk:
394 break
395 buf += chunk
396 if (terminator and terminator in buf) or (count and len(buf) >= count) or len(buf) >= MAX_READ:
397 complete = True
398 break
399 finally:
400 os.close(fd)
401 if not buf and (write not in (None, "") or terminator or count):
402 raise TransportError(f"{device}: no answer within {timeout:g} s")
403 text = buf.decode("utf-8", errors="replace")
404 return {"device": device, "baud": baud, "sent": sent, "body": text, "bytes": len(buf), "complete": complete, "json": _json(text.strip()), "lines": text.splitlines()}
405
406
407# ---------------------------------------------------------------- queue (MQTT 3.1.1)
408
409
410def _mqtt_str(s: str) -> bytes:
411 b = s.encode("utf-8")
412 return struct.pack(">H", len(b)) + b
413
414
415def _mqtt_packet(head: int, body: bytes) -> bytes:
416 n = len(body)
417 length = b""
418 while True:
419 byte = n % 128
420 n //= 128
421 length += bytes([byte | (0x80 if n else 0)])
422 if not n:
423 break
424 return bytes([head]) + length + body
425
426
427def _mqtt_read(sock: socket.socket, deadline: float) -> tuple[int, int, bytes] | None:
428 """One packet: ``(type, flags, body)``, or None at the deadline."""
429
430 def take(n: int) -> bytes | None:
431 out = b""
432 while len(out) < n:
433 remaining = deadline - time.monotonic()
434 if remaining <= 0:
435 return None
436 sock.settimeout(remaining)
437 try:
438 chunk = sock.recv(n - len(out))
439 except (socket.timeout, TimeoutError):
440 return None
441 if not chunk:
442 raise TransportError("the broker closed the connection")
443 out += chunk
444 return out
445
446 head = take(1)
447 if head is None:
448 return None
449 n = 0
450 shift = 1
451 while True:
452 b = take(1)
453 if b is None:
454 return None
455 n += (b[0] & 0x7F) * shift
456 shift *= 128
457 if not b[0] & 0x80:
458 break
459 body = take(n) if n else b""
460 if body is None:
461 return None
462 return head[0] >> 4, head[0] & 0x0F, body
463
464
465def mqtt(spec: dict[str, Any], timeout: float, egress: str = "any") -> dict[str, Any]:
466 """Connect, then either publish one message (``publish``) or subscribe
467 and wait for one. QoS 0, clean session, no retained state of our own."""
468 url = str(spec.get("url", ""))
469 topic = str(spec.get("topic", ""))
470 try:
471 scheme, host, port = broker(url)
472 except ValueError as e:
473 raise TransportError(str(e)) from None
474 check_egress(f"http://{host}:{port}/", egress)
475 deadline = time.monotonic() + timeout
476 try:
477 sock = socket.create_connection((host, port), timeout=timeout)
478 except OSError as e:
479 raise TransportError(f"{url}: {e.strerror or e}") from e
480 try:
481 if scheme == "mqtts":
482 sock = ssl.create_default_context().wrap_socket(sock, server_hostname=host)
483 client_id = str(spec.get("client_id") or f"pyrtan-{os.getpid()}-{int(time.time() * 1000) % 100000}")
484 user, password = spec.get("username"), spec.get("password")
485 flags = 0x02 | (0x80 if user else 0) | (0x40 if password else 0)
486 keepalive = min(65535, max(60, int(timeout) + 30))
487 body = _mqtt_str("MQTT") + bytes([4, flags]) + struct.pack(">H", keepalive) + _mqtt_str(client_id)
488 if user:
489 body += _mqtt_str(str(user))
490 if password:
491 body += _mqtt_str(str(password))
492 sock.sendall(_mqtt_packet(0x10, body))
493 ack = _mqtt_read(sock, deadline)
494 if ack is None or ack[0] != 2 or len(ack[2]) < 2:
495 raise TransportError(f"{url}: no CONNACK within {timeout:g} s")
496 if ack[2][1] != 0:
497 raise TransportError(f"{url}: connection refused (code {ack[2][1]})")
498 if "publish" in spec:
499 payload = spec["publish"]
500 if not isinstance(payload, str):
501 payload = json.dumps(payload, ensure_ascii=False)
502 data = payload.encode("utf-8")
503 sock.sendall(_mqtt_packet(0x30, _mqtt_str(topic) + data))
504 sock.sendall(_mqtt_packet(0xE0, b""))
505 return {"url": url, "topic": topic, "published": len(data), "body": payload}
506 sock.sendall(_mqtt_packet(0x82, struct.pack(">H", 1) + _mqtt_str(topic) + b"\x00"))
507 message = None
508 while True:
509 pkt = _mqtt_read(sock, deadline)
510 if pkt is None:
511 break
512 ptype, pflags, pbody = pkt
513 if ptype == 9: # SUBACK
514 if len(pbody) >= 3 and pbody[2] == 0x80:
515 raise TransportError(f"{url}: the broker refused the subscription to {topic!r}")
516 continue
517 if ptype == 3: # PUBLISH
518 tlen = struct.unpack(">H", pbody[:2])[0]
519 mtopic = pbody[2:2 + tlen].decode("utf-8", errors="replace")
520 rest = pbody[2 + tlen:]
521 qos = (pflags >> 1) & 3
522 if qos:
523 pid, rest = rest[:2], rest[2:]
524 if qos == 1:
525 sock.sendall(_mqtt_packet(0x40, pid))
526 text = rest.decode("utf-8", errors="replace")
527 message = {"topic": mtopic, "body": text, "json": _json(text)}
528 break
529 try:
530 sock.sendall(_mqtt_packet(0xE0, b""))
531 except OSError:
532 pass
533 return {"url": url, "topic": topic, "received": message is not None, "message": message, "body": message["body"] if message else None, "json": message["json"] if message else None}
534 finally:
535 try:
536 sock.close()
537 except OSError:
538 pass
539
540
541# ---------------------------------------------------------------- browser (Playwright)
542
543
544class Browser:
545 """A page rendered by Chromium, through Playwright. Playwright's sync
546 API lives on one thread, so one worker thread owns the browser and the
547 service's request threads queue jobs for it. The page may fetch what
548 it likes; it may not take the tree to another host: a top-level
549 navigation outside the tree's hosts is aborted."""
550
551 def __init__(self, executable: str | None = None) -> None:
552 self.executable = executable
553 self._jobs: _queue.Queue[tuple[Any, ...] | None] = _queue.Queue()
554 self._thread: threading.Thread | None = None
555 self._error: str | None = None
556 self._ready = threading.Event()
557 self._lock = threading.Lock()
558
559 def _start(self) -> None:
560 with self._lock:
561 if self._thread is None:
562 self._thread = threading.Thread(target=self._loop, name="pyrtan-browser", daemon=True)
563 self._thread.start()
564 self._ready.wait(120)
565 if self._error:
566 raise TransportError(self._error)
567
568 def _loop(self) -> None:
569 try:
570 from playwright.sync_api import sync_playwright
571 except ImportError:
572 self._error = "the browser adapter needs Playwright: pip install 'pyrtan[browser]' (and 'playwright install chromium', or set PYRTAN_BROWSER to a Chromium)"
573 self._ready.set()
574 return
575 try:
576 pw = sync_playwright().start()
577 browser = pw.chromium.launch(executable_path=self.executable, headless=True)
578 except Exception as e:
579 self._error = f"the browser could not start: {str(e).splitlines()[0]}"
580 self._ready.set()
581 return
582 self._ready.set()
583 while True:
584 job = self._jobs.get()
585 if job is None:
586 break
587 spec, timeout, allowed, done, box = job
588 try:
589 box["result"] = self._render(browser, spec, timeout, allowed)
590 except TransportError as e:
591 box["error"] = str(e)
592 except Exception as e: # Playwright's own errors
593 box["error"] = str(e).splitlines()[0]
594 finally:
595 done.set()
596 try:
597 browser.close()
598 pw.stop()
599 except Exception:
600 pass
601
602 def _render(self, browser: Any, spec: dict[str, Any], timeout: float, allowed: Allowed) -> dict[str, Any]:
603 url = str(spec.get("url", ""))
604 started = time.monotonic()
605 context = browser.new_context(user_agent=f"pyrtan/{_version()} (+https://github.com/b-i-t-8/pyrtan)")
606 try:
607 page = context.new_page()
608 page.set_default_timeout(timeout * 1000)
609 blocked: list[str] = []
610
611 def route(r: Any) -> None:
612 req = r.request
613 if req.resource_type == "document" and req.is_navigation_request() and not allowed(req.url):
614 blocked.append(req.url)
615 r.abort()
616 else:
617 r.continue_()
618
619 page.route("**/*", route)
620 try:
621 resp = page.goto(url, wait_until=str(spec.get("wait_until", "load")))
622 except Exception as e:
623 if blocked:
624 raise TransportError(f"{url} led to {blocked[-1]}, outside this tree's hosts; the navigation was refused") from None
625 raise TransportError(f"{url}: {str(e).splitlines()[0]}") from None
626 if blocked:
627 raise TransportError(f"{url} led to {blocked[-1]}, outside this tree's hosts; the navigation was refused")
628 wait_for = spec.get("wait_for")
629 if wait_for:
630 page.wait_for_selector(str(wait_for))
631 for action in spec.get("actions") or []:
632 self._act(page, action)
633 if blocked:
634 raise TransportError(f"{page.url} led to {blocked[-1]}, outside this tree's hosts; the navigation was refused")
635 status = resp.status if resp is not None else None
636 return {"url": page.url, "status": status, "ok": status is not None and 200 <= status < 300, "title": page.title(), "body": page.content(), "elapsed_ms": int((time.monotonic() - started) * 1000)}
637 finally:
638 context.close()
639
640 @staticmethod
641 def _act(page: Any, action: dict[str, Any]) -> None:
642 if "click" in action:
643 page.click(str(action["click"]))
644 elif "fill" in action:
645 page.fill(str(action["fill"]["selector"]), str(action["fill"]["value"]))
646 elif "press" in action:
647 page.press(str(action["press"]["selector"]), str(action["press"]["key"]))
648 elif "select" in action:
649 page.select_option(str(action["select"]["selector"]), str(action["select"]["value"]))
650 elif "wait_for" in action:
651 page.wait_for_selector(str(action["wait_for"]))
652 elif "wait" in action:
653 page.wait_for_timeout(float(action["wait"]))
654 else:
655 raise TransportError(f"a browser action is one of {', '.join(BROWSER_ACTIONS)}, not {sorted(action)}")
656
657 def perform(self, spec: dict[str, Any], timeout: float, allowed: Allowed, egress: str = "any") -> dict[str, Any]:
658 url = str(spec.get("url", ""))
659 check_egress(url, egress)
660 self._start()
661 done = threading.Event()
662 box: dict[str, Any] = {}
663 self._jobs.put((spec, timeout, allowed, done, box))
664 if not done.wait(timeout + 30):
665 raise TransportError(f"{url}: the browser did not answer within {timeout + 30:g} s")
666 if "error" in box:
667 raise TransportError(box["error"])
668 return box["result"]
669
670 def close(self) -> None:
671 if self._thread is not None:
672 self._jobs.put(None)
673 self._thread.join(10)
674 self._thread = None
675
676
677def _version() -> str:
678 from . import __version__
679
680 return __version__
681
682
683# ---------------------------------------------------------------- streams, for watches
684# A watch holds a line open and hands each message or line to a callback
685# until told to stop. Errors are reported through the callback's twin and
686# the caller reconnects; nothing here retries on its own.
687
688
689def mqtt_stream(spec: dict[str, Any], on_message: Callable[[dict[str, Any]], None], stop: threading.Event, egress: str = "any", timeout: float = 10.0) -> None:
690 """Subscribe to ``spec['topic']`` at ``spec['url']`` and call
691 ``on_message({"topic", "body", "json"})`` for each PUBLISH until
692 ``stop`` is set. Pings the broker at half the keepalive. Raises
693 :class:`TransportError` when the connection is lost."""
694 url = str(spec.get("url", ""))
695 topic = str(spec.get("topic", ""))
696 try:
697 scheme, host, port = broker(url)
698 except ValueError as e:
699 raise TransportError(str(e)) from None
700 check_egress(f"http://{host}:{port}/", egress)
701 keepalive = 60
702 try:
703 sock = socket.create_connection((host, port), timeout=timeout)
704 except OSError as e:
705 raise TransportError(f"{url}: {e.strerror or e}") from e
706 try:
707 if scheme == "mqtts":
708 sock = ssl.create_default_context().wrap_socket(sock, server_hostname=host)
709 client_id = str(spec.get("client_id") or f"pyrtan-watch-{os.getpid()}-{int(time.time() * 1000) % 100000}")
710 user, password = spec.get("username"), spec.get("password")
711 flags = 0x02 | (0x80 if user else 0) | (0x40 if password else 0)
712 body = _mqtt_str("MQTT") + bytes([4, flags]) + struct.pack(">H", keepalive) + _mqtt_str(client_id)
713 if user:
714 body += _mqtt_str(str(user))
715 if password:
716 body += _mqtt_str(str(password))
717 sock.sendall(_mqtt_packet(0x10, body))
718 ack = _mqtt_read(sock, time.monotonic() + timeout)
719 if ack is None or ack[0] != 2 or len(ack[2]) < 2 or ack[2][1] != 0:
720 raise TransportError(f"{url}: the broker did not accept the connection")
721 sock.sendall(_mqtt_packet(0x82, struct.pack(">H", 1) + _mqtt_str(topic) + b"\x00"))
722 last_ping = time.monotonic()
723 while not stop.is_set():
724 pkt = _mqtt_read(sock, time.monotonic() + 1.0)
725 if pkt is not None:
726 ptype, pflags, pbody = pkt
727 if ptype == 9 and len(pbody) >= 3 and pbody[2] == 0x80:
728 raise TransportError(f"{url}: the broker refused the subscription to {topic!r}")
729 if ptype == 3:
730 tlen = struct.unpack(">H", pbody[:2])[0]
731 mtopic = pbody[2:2 + tlen].decode("utf-8", errors="replace")
732 rest = pbody[2 + tlen:]
733 qos = (pflags >> 1) & 3
734 if qos:
735 pid, rest = rest[:2], rest[2:]
736 if qos == 1:
737 sock.sendall(_mqtt_packet(0x40, pid))
738 text = rest.decode("utf-8", errors="replace")
739 on_message({"topic": mtopic, "body": text, "json": _json(text)})
740 if time.monotonic() - last_ping > keepalive / 2:
741 sock.sendall(_mqtt_packet(0xC0, b""))
742 last_ping = time.monotonic()
743 try:
744 sock.sendall(_mqtt_packet(0xE0, b""))
745 except OSError:
746 pass
747 except OSError as e:
748 raise TransportError(f"{url}: {e.strerror or e}") from e
749 finally:
750 try:
751 sock.close()
752 except OSError:
753 pass
754
755
756def serial_lines(spec: dict[str, Any], on_line: Callable[[str], None], stop: threading.Event) -> None:
757 """Hold ``spec['device']`` open and call ``on_line`` for each unit
758 ending in ``read_until`` (default newline) until ``stop`` is set.
759 Raises :class:`TransportError` when the device fails."""
760 try:
761 import termios
762 except ImportError: # pragma: no cover
763 raise TransportError("serial lines need a POSIX system") from None
764 device = str(spec.get("device", ""))
765 baud = int(spec.get("baud", 9600))
766 terminator = str(spec.get("read_until", "\n")).encode("utf-8") or b"\n"
767 speed = getattr(termios, f"B{baud}", None)
768 if speed is None:
769 raise TransportError(f"{device}: {baud} baud is not a rate this system knows")
770 try:
771 fd = os.open(device, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
772 except OSError as e:
773 raise TransportError(f"{device}: {e.strerror or e}") from e
774 buf = b""
775 try:
776 attrs = termios.tcgetattr(fd)
777 attrs[0] &= ~(termios.IXON | termios.IXOFF | termios.ICRNL | termios.INLCR | termios.IGNCR | termios.ISTRIP)
778 attrs[1] &= ~termios.OPOST
779 attrs[2] = (attrs[2] & ~(termios.PARENB | termios.CSTOPB | termios.CSIZE)) | termios.CS8 | termios.CLOCAL | termios.CREAD
780 attrs[3] &= ~(termios.ICANON | termios.ECHO | termios.ECHOE | termios.ISIG | termios.IEXTEN)
781 attrs[4] = attrs[5] = speed
782 termios.tcsetattr(fd, termios.TCSANOW, attrs)
783 while not stop.is_set():
784 ready, _, _ = select.select([fd], [], [], 0.5)
785 if not ready:
786 continue
787 try:
788 chunk = os.read(fd, 4096)
789 except BlockingIOError:
790 continue
791 except OSError as e:
792 raise TransportError(f"{device}: {e.strerror or e}") from e
793 if not chunk:
794 raise TransportError(f"{device}: closed")
795 buf += chunk
796 while terminator in buf:
797 line, buf = buf.split(terminator, 1)
798 on_line(line.decode("utf-8", errors="replace"))
799 if len(buf) > MAX_READ:
800 buf = b""
801 finally:
802 os.close(fd)