pyrtan / sourcev0.14.0 · main

pyrtan / tests / test_watch_grow.py

20,033 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"""Watches (a tree that runs when something speaks) and growing a tree from
2a gap (the branch drafted by rules for a person to accept)."""
3
4from __future__ import annotations
5
6import json
7import os
8import threading
9import time
10from pathlib import Path
11
12import pytest
13
14from pyrtan import Registry
15from pyrtan.cli import main
16from pyrtan.grow import Draft, Unanswered, apply_draft, draft_branch
17from pyrtan.service import Server, Service, http_json
18from pyrtan.tree import TreeError, check, load_tree, load_tree_text
19from pyrtan.watch import Watcher
20from test_adapters import Broker
21
22LINES = '''
23name = "lines"
24description = "counts the lines a watch brought"
25paths = ["{root}"]
26start = "count"
27
28[inputs.text]
29type = "str"
30
31[inputs.n]
32type = "int"
33default = 0
34
35[[watches]]
36name = "log"
37file = "{log}"
38inputs = {{ text = "{{{{ event.body | strip }}}}", n = "{{{{ event.lines | length }}}}" }}
39
40[nodes.count]
41kind = "decide"
42on = "inputs.n"
43branches = [
44 {{ when = {{ eq = 1 }}, then = "one" }},
45 {{ when = {{ gt = 1 }}, then = "many" }},
46 {{ else = "none" }},
47]
48
49[nodes.one]
50kind = "done"
51say = "one line: {{{{ inputs.text }}}}"
52
53[nodes.many]
54kind = "done"
55say = "{{{{ inputs.n }}}} lines"
56
57[nodes.none]
58kind = "fail"
59reason = "nothing"
60say = "nothing"
61'''
62
63TICK = '''
64name = "tick"
65description = "reads a file on a clock; the path changes when the file does"
66paths = ["{root}"]
67start = "read"
68
69[[watches]]
70name = "clock"
71every = 1
72
73[nodes.read]
74kind = "observe"
75file = {{ path = "{flag}" }}
76into = "f"
77next = "there"
78
79[nodes.there]
80kind = "decide"
81on = "f.exists"
82branches = [{{ when = {{ eq = true }}, then = "yes" }}, {{ else = "no" }}]
83
84[nodes.yes]
85kind = "done"
86say = "there"
87
88[nodes.no]
89kind = "done"
90say = "gone"
91'''
92
93
94def wait_for(cond, timeout=10.0):
95 deadline = time.time() + timeout
96 while time.time() < deadline:
97 if cond():
98 return True
99 time.sleep(0.05)
100 return False
101
102
103def registry(tmp_path: Path, **trees: str) -> Registry:
104 d = tmp_path / "trees"
105 d.mkdir(parents=True, exist_ok=True)
106 for name, text in trees.items():
107 (d / f"{name}.toml").write_text(text)
108 reg = Registry.from_paths([d])
109 assert not reg.errors, reg.errors
110 return reg
111
112
113# ---------------------------------------------------------------- watches
114
115
116def test_watches_are_checked():
117 base = LINES.format(root="/tmp", log="/tmp/x.log")
118 tree = load_tree_text(base)
119 assert tree.watch_kinds() == ["file"] and tree.capability()["watches"] == [{"kind": "file", "what": "/tmp/x.log"}]
120 with pytest.raises(TreeError, match="exactly one of every, file, queue, serial"):
121 load_tree_text(base.replace('file = "/tmp/x.log"', 'file = "/tmp/x.log"\nevery = "5m"'))
122 with pytest.raises(TreeError, match="not under this tree's paths"):
123 load_tree_text(base.replace('file = "/tmp/x.log"', 'file = "/etc/x.log"'))
124 with pytest.raises(TreeError, match="literal path"):
125 load_tree_text(base.replace('file = "/tmp/x.log"', 'file = "/tmp/{{ env.x }}"'))
126 with pytest.raises(TreeError, match="template root 'state' is not one of event, env, now"):
127 load_tree_text(base.replace("{{ event.body | strip }}", "{{ state.body }}"))
128 with pytest.raises(TreeError, match="input 'zzz' is not declared"):
129 load_tree_text(base.replace("n = \"{{ event.lines | length }}\"", "n = \"{{ event.lines | length }}\", zzz = 1"))
130 with pytest.raises(TreeError, match="input 'text' is required and the watch does not give it"):
131 load_tree_text(base.replace('inputs = { text = "{{ event.body | strip }}", n = "{{ event.lines | length }}" }', 'inputs = { n = 1 }'))
132 with pytest.raises(TreeError, match="'every' is a duration"):
133 load_tree_text(base.replace('file = "/tmp/x.log"', 'every = "soon"').replace('paths = ["/tmp"]', ""))
134 ok = load_tree_text(base.replace('file = "/tmp/x.log"', 'every = "1 hour and a half"').replace('paths = ["/tmp"]', ""))
135 assert ok.capability()["watches"] == [{"kind": "every", "what": "1 hour and a half"}]
136 with pytest.raises(TreeError, match="not under this tree's topics"):
137 load_tree_text(base.replace('file = "/tmp/x.log"', 'queue = { url = "mqtt://b", topic = "x/y" }').replace('paths = ["/tmp"]', 'hosts = ["b"]\ntopics = ["a/#"]'))
138 with pytest.raises(TreeError, match="host 'c' is not in this tree's hosts"):
139 load_tree_text(base.replace('file = "/tmp/x.log"', 'queue = { url = "mqtt://c", topic = "a/y" }').replace('paths = ["/tmp"]', 'hosts = ["b"]\ntopics = ["a/#"]'))
140 with pytest.raises(TreeError, match="not in this tree's devices"):
141 load_tree_text(base.replace('file = "/tmp/x.log"', 'serial = { device = "/dev/ttyACM0" }').replace('paths = ["/tmp"]', 'devices = ["/dev/ttyUSB*"]'))
142 # A watch on a source counts as using it: the list is required.
143 with pytest.raises(TreeError, match="must list the paths"):
144 load_tree_text(base.replace('paths = ["/tmp"]', ""))
145
146
147def test_file_watch_runs_the_tree_for_appended_lines(tmp_path):
148 log = tmp_path / "app.log"
149 log.write_text("old line\n")
150 reg = registry(tmp_path, lines=LINES.format(root=str(tmp_path), log=str(log)))
151 svc = Service(reg, name="w")
152 fired: list = []
153 watcher = Watcher(reg, svc.run_watched, after=lambda w, e, o: fired.append((w.name, e, o))).start()
154 try:
155 assert len(watcher) == 1 and watcher.status()["watches"][0]["what"] == str(log)
156 time.sleep(0.3) # the watch starts at the end: the old line is not an event
157 with log.open("a") as f:
158 f.write("hello\n")
159 assert wait_for(lambda: len(fired) == 1)
160 name, event, out = fired[0]
161 assert name == "log" and event["lines"] == ["hello"] and out.status == "done" and out.say == "one line: hello"
162 assert out.trace.steps[0].kind == "watch" and out.trace.steps[0].node == "watch" and out.trace.steps[0].next == "count"
163 assert "hello" in out.trace.steps[0].summary or "1 new line" in out.trace.steps[0].summary
164 with log.open("a") as f:
165 f.write("a\nb\nc\n")
166 assert wait_for(lambda: len(fired) == 2) and fired[1][2].say == "3 lines"
167 # Truncation (log rotation) is read from the start again.
168 log.write_text("fresh\n")
169 assert wait_for(lambda: len(fired) == 3) and fired[2][2].say == "one line: fresh"
170 st = watcher.status()["watches"][0]
171 assert st["fires"] == 3 and st["last"]["status"] == "done" and st["path"] == ["watch", "count", "one"] and st["alive"]
172 # The path changed between runs 2 and 3 (many -> one), and the service counted the runs.
173 assert st["path_changed_at"] is not None and st["previous_path"] == ["watch", "count", "many"]
174 assert svc.runs == 3
175 finally:
176 watcher.stop()
177
178
179def test_clock_watch_notices_the_path_change_and_sync_follows_the_files(tmp_path):
180 flag = tmp_path / "flag"
181 reg = registry(tmp_path, tick=TICK.format(root=str(tmp_path), flag=str(flag)))
182 svc = Service(reg, name="w")
183 fired: list = []
184 watcher = Watcher(reg, svc.run_watched, after=lambda w, e, o: fired.append(o)).start()
185 try:
186 assert wait_for(lambda: len(fired) >= 1) and fired[0].say == "gone" and fired[0].trace.steps[0].summary == "every 1 s, run 1"
187 flag.write_text("x")
188 assert wait_for(lambda: any(o.say == "there" for o in fired), 5)
189 st = watcher.status()["watches"][0]
190 assert st["path_changed_at"] and st["previous_path"] == ["watch", "read", "there", "no"] and st["path"] == ["watch", "read", "there", "yes"]
191 # A tree that vanishes stops its watch; one that appears starts one.
192 (tmp_path / "trees" / "tick.toml").unlink()
193 assert watcher.sync() == (0, 1) and len(watcher) == 0
194 (tmp_path / "trees" / "tick.toml").write_text(TICK.format(root=str(tmp_path), flag=str(flag)))
195 assert watcher.sync() == (1, 0) and len(watcher) == 1
196 finally:
197 watcher.stop()
198
199
200def test_queue_watch_runs_per_message_and_serial_watch_per_line(tmp_path):
201 b = Broker(message=None)
202 tree = f'''
203name = "sensor"
204description = "x"
205hosts = ["127.0.0.1"]
206topics = ["home/#"]
207start = "say"
208
209[inputs.reading]
210type = "str"
211
212[[watches]]
213queue = {{ url = "{b.url}", topic = "home/+/temperature" }}
214inputs = {{ reading = "{{{{ event.topic }}}}={{{{ event.json.temperature }}}}" }}
215
216[nodes.say]
217kind = "done"
218say = "{{{{ inputs.reading }}}}"
219'''
220 reg = registry(tmp_path, sensor=tree)
221 svc = Service(reg, name="w")
222 fired: list = []
223 watcher = Watcher(reg, svc.run_watched, after=lambda w, e, o: fired.append(o)).start()
224 try:
225 assert wait_for(lambda: b.subscribed == ["home/+/temperature"])
226 b.publish_to_subscribers("home/kitchen/temperature", b'{"temperature": 21}')
227 assert wait_for(lambda: len(fired) == 1) and fired[0].say == "home/kitchen/temperature=21"
228 b.publish_to_subscribers("home/attic/temperature", b'{"temperature": 30}')
229 assert wait_for(lambda: len(fired) == 2) and fired[1].say == "home/attic/temperature=30"
230 finally:
231 watcher.stop()
232 b.close()
233 # Serial: a pty, each line an event.
234 master, slave = os.openpty()
235 device = os.ttyname(slave)
236 stree = f'''
237name = "modem"
238description = "x"
239devices = ["/dev/pts/*"]
240start = "say"
241
242[inputs.line]
243type = "str"
244
245[[watches]]
246serial = {{ device = "{device}", baud = 9600, read_until = "\\n" }}
247inputs = {{ line = "{{{{ event.body | strip }}}}" }}
248
249[nodes.say]
250kind = "done"
251say = "line: {{{{ inputs.line }}}}"
252'''
253 reg2 = registry(tmp_path / "s", modem=stree)
254 svc2 = Service(reg2, name="w2")
255 fired2: list = []
256 watcher2 = Watcher(reg2, svc2.run_watched, after=lambda w, e, o: fired2.append(o)).start()
257 try:
258 time.sleep(0.3)
259 os.write(master, b"RING\r\nOK\r\n")
260 assert wait_for(lambda: len(fired2) == 2) and [o.say for o in fired2] == ["line: RING", "line: OK"]
261 finally:
262 watcher2.stop()
263 os.close(master)
264 os.close(slave)
265
266
267def test_hosted_service_watches_no_file_and_errors_are_shown(tmp_path):
268 log = tmp_path / "x.log"
269 log.write_text("")
270 reg = registry(tmp_path, lines=LINES.format(root=str(tmp_path), log=str(log)))
271 svc = Service(reg, name="w", egress="public")
272 notes: list[str] = []
273 watcher = Watcher(reg, svc.run_watched, egress="public", note=notes.append).start()
274 try:
275 assert wait_for(lambda: notes)
276 assert "watches no file" in notes[0]
277 st = watcher.status()["watches"][0]
278 assert st["errors"] == 1 and "watches no file" in st["last_error"]
279 finally:
280 watcher.stop()
281
282
283def test_watches_over_http_and_the_commands(tmp_path, capsys):
284 log = tmp_path / "x.log"
285 log.write_text("")
286 (tmp_path / "t").mkdir()
287 (tmp_path / "t" / "lines.toml").write_text(LINES.format(root=str(tmp_path), log=str(log)))
288 from pyrtan.hub import serve_hub
289 from pyrtan.service import serve
290
291 hub, hub_srv = serve_hub(port=0, quiet=True, poll_every=60)
292 svc, srv = serve([tmp_path / "t"], port=0, quiet=True, watch=True, hub=hub_srv.url)
293 try:
294 assert srv.url in {p.url for p in hub.peers.values()}
295 with log.open("a") as f:
296 f.write("ping\n")
297 assert wait_for(lambda: http_json("GET", f"{srv.url}/watches")[1]["watches"][0]["fires"] == 1)
298 _, w = http_json("GET", f"{srv.url}/watches")
299 assert w["count"] == 1 and w["watches"][0]["last"]["say"] == "one line: ping"
300 _, hw = http_json("GET", f"{hub_srv.url}/watches")
301 assert hw["count"] == 1 and hw["watches"][0]["peer"] == srv.url
302 assert "/watches" in http_json("GET", f"{srv.url}/api")[1]["endpoints"]
303 assert main(["watches", "--at", hub_srv.url]) == 0
304 out = capsys.readouterr().out
305 assert "log file " in out and "fired 1×" in out and "one line: ping" in out
306 assert main(["watches", "--at", hub_srv.url, "--json"]) == 0 and json.loads(capsys.readouterr().out)["watches"][0]["name"] == "log"
307 # A service without --watch watches nothing, and says so.
308 svc2, srv2 = serve([tmp_path / "t"], port=0, quiet=True)
309 try:
310 assert main(["watches", "--at", srv2.url]) == 0 and "watches nothing" in capsys.readouterr().out
311 finally:
312 srv2.stop()
313 finally:
314 svc.stop()
315 srv.stop()
316 hub.stop()
317 hub_srv.stop()
318
319
320def test_unanswered_survive_a_restart(tmp_path):
321 path = tmp_path / "unanswered.jsonl"
322 u = Unanswered(path=path)
323 u.record("top 3 stories on lobsters")
324 u.record("top 3 stories on lobsters", peer="x")
325 u.record("is the printer on")
326 again = Unanswered(path=path)
327 rows = again.to_list()
328 assert [(r["text"], r["count"]) for r in rows] == [("top 3 stories on lobsters", 2), ("is the printer on", 1)]
329 assert rows[0]["proposal"]["name"] == "top-stories-lobsters"
330 (tmp_path / "none").mkdir()
331 svc = Service(Registry.from_paths([tmp_path / "none"]), name="s", unanswered=path)
332 assert len(svc.unanswered) == 2
333
334
335# ---------------------------------------------------------------- growing from a gap
336
337
338WEATHERISH = '''
339name = "sky"
340description = "a code into a word"
341start = "pick"
342
343[inputs.code]
344type = "int"
345
346[nodes.pick]
347kind = "decide"
348on = "inputs.code"
349branches = [
350 { when = { eq = 0 }, then = "clear" },
351 { when = { between = [1, 3] }, then = "cloudy" },
352 { when = { between = [51, 67] }, then = "rain" },
353 { when = { between = [95, 99] }, then = "storm" },
354]
355
356[nodes.clear]
357kind = "done"
358say = "clear"
359
360[nodes.cloudy]
361kind = "done"
362say = "cloudy"
363
364[nodes.rain]
365kind = "done"
366say = "rain"
367
368[nodes.storm]
369kind = "done"
370say = "storm"
371'''
372
373
374def test_draft_branch_rules():
375 tree = load_tree_text(WEATHERISH)
376 # A hole between ranges: the range of the hole.
377 d = draft_branch(tree, {"tree": "sky", "node": "pick", "on": "inputs.code", "observed": 71})
378 assert d is not None and d.when == {"between": [68, 94]} and d.then == "pick-71" and "hole between 67 and 95" in d.reason
379 assert d.branch == '{ when = { between = [68, 94] }, then = "pick-71" },'
380 assert d.node_text.startswith("# Drafted by pyrtan grow") and '[nodes.pick-71]\nkind = "fail"\nreason = "71"' in d.node_text
381 # Beyond the last range: just the value.
382 assert draft_branch(tree, {"node": "pick", "on": "inputs.code", "observed": 120}).when == {"eq": 120}
383 # Covered already: nothing to draft.
384 assert draft_branch(tree, {"node": "pick", "on": "inputs.code", "observed": 2}) is None
385 # Not a decide, or no such node: nothing.
386 assert draft_branch(tree, {"node": "clear", "on": "x", "observed": 1}) is None and draft_branch(tree, {"node": "nope", "observed": 1}) is None
387 # Missing, strings, types from a report.
388 assert draft_branch(tree, {"node": "pick", "on": "inputs.code", "observed": None}).when == {"exists": False}
389 s = draft_branch(tree, {"node": "pick", "on": "inputs.code", "observed": "N/A"})
390 assert s.when == {"eq": "N/A"} and s.then == "pick-n-a" and s.branch == '{ when = { eq = "N/A" }, then = "pick-n-a" },'
391 assert draft_branch(tree, {"node": "pick", "on": "inputs.code", "observed": "<dict>"}).when == {"type": "dict"}
392 assert draft_branch(tree, {"node": "pick", "on": "inputs.code", "observed": True}).when == {"eq": True}
393 # A node name already taken gets a suffix.
394 taken = load_tree_text(WEATHERISH.replace('[nodes.storm]', '[nodes.pick-71]\nkind = "done"\nsay = "x"\n\n[nodes.storm]'))
395 assert draft_branch(taken, {"node": "pick", "on": "inputs.code", "observed": 71}).then == "pick-71-2"
396
397
398def test_apply_draft_edits_the_file_and_reverts_when_it_does_not_fit(tmp_path):
399 path = tmp_path / "sky.toml"
400 path.write_text(WEATHERISH)
401 tree = load_tree(path)
402 d = draft_branch(tree, {"node": "pick", "on": "inputs.code", "observed": 71})
403 text = apply_draft(path, d)
404 grown = load_tree(path)
405 assert "pick-71" in grown.nodes and grown.nodes["pick-71"].kind == "fail"
406 assert [b.get("then") for b in grown.nodes["pick"].get("branches")] == ["clear", "cloudy", "rain", "storm", "pick-71"]
407 assert " { when = { between = [68, 94] }, then = \"pick-71\" },\n]" in text
408 assert not [p for p in check(grown) if p.level == "error"]
409 # Before an else, and on a one-line list.
410 one = tmp_path / "one.toml"
411 one.write_text(WEATHERISH.replace(" { when = { between = [95, 99] }, then = \"storm\" },\n]", " { when = { between = [95, 99] }, then = \"storm\" },\n { else = \"storm\" },\n]"))
412 apply_draft(one, draft_branch(load_tree(one), {"node": "pick", "on": "inputs.code", "observed": 71}))
413 assert [b.get("then") or "else" for b in load_tree(one).nodes["pick"].get("branches")] == ["clear", "cloudy", "rain", "storm", "pick-71", "else"]
414 flat = tmp_path / "flat.toml"
415 flat.write_text('name = "f"\ndescription = "x"\nstart = "p"\n[inputs.v]\ntype = "str"\n[nodes.p]\nkind = "decide"\non = "inputs.v"\nbranches = [{ when = { eq = "a" }, then = "a" }, { else = "b" }]\n[nodes.a]\nkind = "done"\nsay = "a"\n[nodes.b]\nkind = "done"\nsay = "b"\n')
416 apply_draft(flat, draft_branch(load_tree(flat), {"node": "p", "on": "inputs.v", "observed": "zz"}))
417 assert [b.get("then") or "else" for b in load_tree(flat).nodes["p"].get("branches")] == ["a", "p-zz", "else"]
418 # A draft that cannot fit (a node the file has no table for) leaves the file as it was.
419 before = path.read_text()
420 with pytest.raises(TreeError, match="no \\[nodes.nope\\] table"):
421 apply_draft(path, Draft("sky", "nope", "x", 1, {"eq": 1}, "nope-1", '{ when = { eq = 1 }, then = "nope-1" },', '[nodes.nope-1]\nkind = "done"\n'))
422 assert path.read_text() == before
423 bad = Draft("sky", "pick", "inputs.code", 5, {"eq": 5}, "pick-5", '{ when = { eq = 5 }, then = "pick-5" },', '[nodes.pick-5]\nkind = "done"\nnext = "nowhere"\n')
424 with pytest.raises(TreeError, match="did not fit the file"):
425 apply_draft(path, bad)
426 assert path.read_text() == before
427
428
429def test_grow_command(tmp_path, capsys):
430 (tmp_path / "t").mkdir()
431 path = tmp_path / "t" / "sky.toml"
432 path.write_text(WEATHERISH)
433 gaps = tmp_path / "gaps.jsonl"
434 gaps.write_text(json.dumps({"tree": "sky", "node": "pick", "on": "inputs.code", "observed": 71, "count": 3}) + "\n" + json.dumps({"tree": "sky", "node": "pick", "on": "inputs.code", "observed": 2}) + "\n" + json.dumps({"tree": "other", "node": "x", "on": "y", "observed": 1}) + "\n")
435 assert main(["grow", "--from", str(gaps), "--trees", str(tmp_path / "t")]) == 0
436 out = capsys.readouterr().out
437 assert "skipped other: not among the trees here" in out and "is covered already" in out and "[nodes.pick-71]" in out and "nothing written" in out
438 assert load_tree(path).nodes.get("pick-71") is None
439 assert main(["grow", "--from", str(gaps), "--trees", str(tmp_path / "t"), "--apply"]) == 0
440 assert "written into" in capsys.readouterr().out and load_tree(path).nodes["pick-71"].kind == "fail"
441 assert main(["grow", "--from", str(gaps), "--trees", str(tmp_path / "t")]) == 1 and "every gap has its branch" in capsys.readouterr().out
442 # From a federation report: the id names the tree; from a running service: its /gaps.
443 reports = tmp_path / "reports.json"
444 reports.write_text(json.dumps({"reports": [{"id": "pyrtan/sky@1", "node": "pick", "on": "inputs.code", "observed": 120, "count": 2, "reporters": 1}]}))
445 assert main(["grow", "--reports", str(reports), "--trees", str(tmp_path / "t"), "--json"]) == 0
446 drafts = json.loads(capsys.readouterr().out)["drafts"]
447 assert drafts[0]["when"] == {"eq": 120} and drafts[0]["file"].endswith("sky.toml")
448 svc = Service(Registry.from_paths([tmp_path / "t"]), name="s")
449 srv = Server(svc.route, quiet=True).start()
450 try:
451 assert svc.run({"tree": "sky", "inputs": {"code": 42}})["reason"] == "gap"
452 assert main(["grow", "--at", srv.url, "--trees", str(tmp_path / "t")]) == 0 and "pick-42" in capsys.readouterr().out
453 finally:
454 srv.stop()
455 assert main(["grow", "--trees", str(tmp_path / "t")]) == 2