okami.mom/albedo

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

Aly Raffaufrefactor(tests): share fixtures and move workspace checks to E2Eeaef427

main
branchesmainall refs →
4.3 KiB99 linesraw
1"""Python work routes must isolate ledger reads and writes by session workspace."""
2
3import json
4import unittest
5from urllib.parse import urlencode
6
7from harness import Albedo, Provider, python, text
8
9
10class WorkTests(unittest.TestCase):
11    def test_python_work_is_scoped_to_session_workspace(self):
12        provider = Provider(
13            lambda request: (
14                python(
15                    "import os\n"
16                    "name = os.path.basename(os.getcwd())\n"
17                    "await work.create(title=name)\n"
18                    "print([item['title'] for item in await work.list()])"
19                )
20                if request["messages"][-1].get("role") == "user"
21                else text("done")
22            )
23        )
24        self.addCleanup(provider.close)
25        with Albedo(provider) as app:
26            outputs = []
27            for name in ("work-first", "work-second"):
28                workspace = app.workspace / name
29                workspace.mkdir()
30                session = app.session(workspace)
31                app.prompt(session, "use work").close()
32                app.idle(session)
33                results = [
34                    json.loads(part["value"])
35                    for entry in app.history(session)["items"]
36                    if entry["kind"] == "tool_result"
37                    and entry["tool"]["name"] == "python"
38                    for part in entry["content"]
39                    if part["kind"] == "json" and part["field"] == "result"
40                ]
41                self.assertEqual(len(results), 1)
42                self.assertEqual(results[0]["status"], "ok", results[0])
43                outputs.append(results[0]["output"])
44            self.assertIn("work-first", outputs[0])
45            self.assertNotIn("work-second", outputs[0])
46            self.assertIn("work-second", outputs[1])
47            self.assertNotIn("work-first", outputs[1])
48
49    def test_foreign_ids_cannot_be_read_changed_deleted_or_parented(self):
50        foreign = {}
51
52        def script(request):
53            if request["messages"][-1].get("role") != "user":
54                return text("done")
55            item_id, revision = int(foreign["id"]), int(foreign["revision"])
56            return python(
57                f"operations = [lambda: work.get({item_id}), "
58                f"lambda: work.update({item_id}, revision={revision}, title='changed'), "
59                f"lambda: work.delete({item_id}, revision={revision}), "
60                f"lambda: work.create(title='child', parent={item_id})]\n"
61                "for operation in operations:\n"
62                "    try:\n"
63                "        await operation()\n"
64                "    except WorkError:\n"
65                "        print('REFUSED')\n"
66                "    else:\n"
67                "        raise AssertionError('foreign item accepted')\n"
68                "print('EMPTY', await work.list())"
69            )
70
71        provider = Provider(script)
72        self.addCleanup(provider.close)
73        with Albedo(provider) as app:
74            query = urlencode({"workspace": str(app.workspace)})
75            with app.api(
76                f"/extensions/work/items?{query}",
77                {"title": "original"},
78                method="POST",
79            ) as response:
80                foreign.update(json.load(response)["resource"]["value"])
81            other = app.workspace / "other-workspace"
82            other.mkdir()
83            session = app.session(other)
84            app.prompt(session, "try foreign IDs").close()
85            app.idle(session)
86            results = [
87                json.loads(part["value"])
88                for entry in app.history(session)["items"]
89                if entry["kind"] == "tool_result" and entry["tool"]["name"] == "python"
90                for part in entry["content"]
91                if part["kind"] == "json" and part["field"] == "result"
92            ]
93            self.assertEqual(len(results), 1)
94            self.assertEqual(results[0]["status"], "ok", results[0])
95            self.assertEqual(results[0]["output"].count("REFUSED"), 4)
96            self.assertIn("EMPTY []", results[0]["output"])
97            with app.api(f"/extensions/work/items?{query}") as response:
98                items = json.load(response)["items"]
99            self.assertEqual([item["value"]["title"] for item in items], ["original"])