okami.mom/albedo

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

Aly Raffaufrefactor(tests): share repeated E2E fixtures5bdc0ee

main
branchesmainall refs →
2.5 KiB71 linesraw
1"""Shared editable daemon and model-driven cells for kernel E2E scenarios."""
2
3import json
4from pathlib import Path
5import shutil
6import tempfile
7
8import harness
9from harness import Albedo, Provider, latest_user
10
11MARKER = '\nMARKER = "new bundle"\n'
12
13
14def editable_daemon(root):
15    """A copy of the test daemon whose python bundle is ours to edit: every
16    package links back to the snapshot except albedo's priv/python."""
17    assert harness._executable is not None, "the runner snapshots the daemon first"
18    source = Path(harness._executable).parent
19    out = Path(tempfile.mkdtemp(prefix="daemon-", dir=root))
20    for package in source.iterdir():
21        if package.is_dir() and package.name != "albedo":
22            (out / package.name).symlink_to(package)
23    albedo = out / "albedo"
24    (albedo / "priv").mkdir(parents=True)
25    (albedo / "ebin").symlink_to(source / "albedo" / "ebin")
26    priv = (source / "albedo" / "priv").resolve()
27    for entry in priv.iterdir():
28        if entry.name == "python":
29            shutil.copytree(
30                entry,
31                albedo / "priv" / "python",
32                ignore=shutil.ignore_patterns("__pycache__"),
33            )
34        else:
35            (albedo / "priv" / entry.name).symlink_to(entry)
36    launcher = out / "albedo-daemon"
37    launcher.write_text(
38        "#!/bin/sh\n"
39        f"exec erl -pa {out}/*/ebin -eval 'albedo@@main:run(albedo)' -noshell -extra\n"
40    )
41    launcher.chmod(0o755)
42    return launcher, albedo / "priv" / "python"
43
44
45class ModelCells:
46    """Run queued Python cells through the fixture's provider and session."""
47
48    cells: list[str]
49    app: Albedo
50    provider: Provider
51    session: str
52
53    def cell(self, code):
54        """Run one cell through a model turn; its result and the turn's prompt."""
55        self.cells.append(code)
56        self.app.prompt(self.session, "run it").close()
57        self.app.idle(self.session)
58        results = [
59            json.loads(part["value"])
60            for entry in self.app.history(self.session)["items"]
61            if entry["kind"] == "tool_result" and entry["tool"]["name"] == "python"
62            for part in entry["content"]
63            if part["kind"] == "json" and part["field"] == "result"
64        ]
65        request = next(
66            r["request"]
67            for r in reversed(self.provider.requests)
68            if any(i.get("role") == "user" for i in r["request"]["input"])
69        )
70        prompt = latest_user(request)
71        return results[-1], prompt