okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

Aly Raffaufrefactor(test): share inspection and stream supportcba5ff8

main
branchesmainall refs →
3.3 KiB92 linesraw
1"""Bounded Erlang inspection transport shared by isolated daemon scenarios."""
2
3import json
4import re
5import subprocess
6import uuid
7
8
9class Inspection:
10    def __init__(self, app, source=None):
11        self.app = app
12        self.module = source.stem if source is not None else None
13        if source is not None:
14            result = subprocess.run(
15                ["erlc", "-Werror", "-o", str(app.root), str(source)],
16                capture_output=True,
17                text=True,
18                timeout=30,
19            )
20            if result.returncode != 0:
21                raise AssertionError(
22                    f"probe compilation failed: {source}\n{result.stdout}{result.stderr}"
23                )
24
25    def evaluate(self, expression, *, timeout=40):
26        app = self.app
27        log = (app.home / "daemon.log").read_text(errors="replace")
28        nodes = re.findall(rf"inspect: node (albedo_{app.daemon._pid}@[^\s]+)", log)
29        if not nodes:
30            raise AssertionError(
31                "daemon did not announce an inspection node; enable ALBEDO_INSPECT before boot"
32            )
33        cookie = (app.home / "inspect.cookie").read_text().strip()
34        node = nodes[-1]
35        prefix = (
36            f"Node = '{node}', "
37            "Call = fun(Module, Function, Arguments) -> "
38            f"case rpc:call(Node, Module, Function, Arguments, {int(timeout * 1000)}) of "
39            "{badrpc, Reason} -> error({rpc_failed, Module, Function, Reason}); "
40            "Value -> Value end end, "
41        )
42        if self.module is not None:
43            beam = app.root / (self.module + ".beam")
44            prefix += (
45                f"{{ok, Binary}} = file:read_file({json.dumps(str(beam))}), "
46                f"{{module, {self.module}}} = Call(code, load_binary, "
47                f'[{self.module}, "probe.erl", Binary]), '
48            )
49        try:
50            result = subprocess.run(
51                [
52                    "erl",
53                    "+S",
54                    "2:2",
55                    "-sname",
56                    f"probe_{uuid.uuid4().hex}",
57                    "-setcookie",
58                    cookie,
59                    "-noshell",
60                    "-eval",
61                    prefix + expression + ", halt().",
62                ],
63                cwd=app.root,
64                capture_output=True,
65                text=True,
66                timeout=timeout,
67            )
68        except subprocess.TimeoutExpired as error:
69            raise AssertionError(
70                f"inspection of {node} timed out after {timeout}s"
71            ) from error
72        if result.returncode != 0:
73            raise AssertionError(
74                f"inspection of {node} failed:\n{result.stdout}{result.stderr}"
75            )
76        return result.stdout
77
78    def call(self, function, arguments="[]"):
79        if self.module is None:
80            raise ValueError("calling a probe requires its source module")
81        return self.evaluate(
82            f"io:put_chars(Call({self.module}, {function}, {arguments}))"
83        )
84
85    def call_json(self, function, arguments="[]"):
86        output = self.call(function, arguments)
87        try:
88            return json.loads(output)
89        except ValueError as error:
90            raise AssertionError(
91                f"{self.module}:{function} returned invalid JSON: {output}"
92            ) from error