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 →
3.2 KiB88 linesraw
1"""A Python crash injects exactly one reset notice without corrupting history; a
2daemon crash leaves the detached kernel alive, so it injects none."""
3
4import json
5import urllib.error
6import unittest
7
8from harness import latest_user
9from harness import Albedo, Provider, exclusive, python, text
10
11NOTICE = (
12    "<system-note>The python kernel got reset and all variables are lost</system-note>"
13)
14
15
16# exclusive: crashes and restarts the daemon
17@exclusive
18class KernelResetTests(unittest.TestCase):
19    def setUp(self):
20        def script(request):
21            if request["input"][-1].get("role") == "user":
22                prompt = latest_user(request)
23                if prompt == "lose kernel":
24                    return python("import os; os._exit(1)")
25                if prompt == "recover kernel" + NOTICE:
26                    return python("print('kernel-ready')")
27            return text("ok")
28
29        self.provider = Provider(script)
30        self.addCleanup(self.provider.close)
31        self.app = Albedo(self.provider, protocol="responses")
32        self.app.__enter__()
33        self.addCleanup(self.app.__exit__, None, None, None)
34
35    def turn(self, session, prompt):
36        self.app.prompt(session, prompt).close()
37        self.app.idle(session)
38        return latest_user(self.provider.requests[-1]["request"])
39
40    def test_kernel_crash_emits_one_shot_reset_notice_and_daemon_crash_none(self):
41        app = self.app
42        session = app.session()
43        self.assertEqual(self.turn(session, "first prompt"), "first prompt")
44        unused = app.session()
45        app.restart(crash=True)
46        self.assertEqual(
47            self.turn(unused, "unused session's first prompt"),
48            "unused session's first prompt",
49        )
50        with self.assertRaises(urllib.error.HTTPError) as rejected:
51            app.prompt(session, "   ").close()
52        self.assertEqual(rejected.exception.code, 400)
53        self.assertEqual(self.turn(session, "after restart"), "after restart")
54        self.assertEqual(self.turn(session, "ordinary turn"), "ordinary turn")
55        self.assertEqual(self.turn(session, "lose kernel"), "lose kernel")
56        self.assertEqual(
57            self.turn(session, "recover kernel"), "recover kernel" + NOTICE
58        )
59        self.assertTrue(
60            any(
61                "kernel-ready" in item.get("output", "")
62                for item in self.provider.requests[-1]["request"]["input"]
63            )
64        )
65        self.assertEqual(self.turn(session, "still alive"), "still alive")
66        users = [
67            part["text"]
68            for entry in app.history(session)["items"]
69            if entry["kind"] == "user"
70            for part in entry["content"]
71            if part["kind"] == "text"
72        ]
73        self.assertEqual(
74            users,
75            [
76                "first prompt",
77                "after restart",
78                "ordinary turn",
79                "lose kernel",
80                "recover kernel",
81                "still alive",
82            ],
83        )
84        with app.api("/sessions") as response:
85            info = next(
86                item for item in json.load(response)["items"] if item["id"] == session
87            )
88        self.assertEqual(info["name"], "still alive")