okami.mom/albedo

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

nekomimi.petmake python kernel friendlier based on tracesd4842e6

main
branchesmainall refs →
3.5 KiB97 linesraw
1//// Partial repairs require permission, and reused call IDs need distinct rows.
2//// Direct calls pin compile state and stored call identities independently of turns.
3
4import albedo/harness/extensions/python/cells
5import albedo/harness/extensions/python/kernel as python
6import albedo/harness/runtime
7import gleam/string
8import gleeunit/should
9import harness/session_fixture
10
11pub fn partial_execution_requires_explicit_replay_permission_test() -> Nil {
12  let assert Ok(host) = runtime.start(":memory:")
13  session_fixture.initialise(host)
14  session_fixture.create(host, "a", "/tmp")
15  let assert Ok(session) = runtime.open_session(host, "a", "/tmp")
16  let assert Ok(failed) =
17    runtime.execute(
18      host,
19      session,
20      "counter = globals().get('counter', 0) + 1\nraise ValueError('bad')",
21      5000,
22    )
23  let assert Ok(original) = runtime.cell(host, failed.cell_id)
24  original.started |> should.be_true
25  let repair =
26    "await cells.run('"
27    <> failed.cell_id
28    <> "', replacements=[(\"raise ValueError('bad')\", 'counter')]"
29  let assert Ok(denied) = runtime.execute(host, session, repair <> ")", 5000)
30  let assert Ok(denied) = denied.result
31  denied.status |> should.equal(python.Failed)
32  denied.output |> string.contains("allow_partial=True") |> should.be_true
33  let assert Ok(counter) = runtime.execute(host, session, "counter", 5000)
34  let assert Ok(counter) = counter.result
35  counter.value |> should.equal("1")
36  let assert Ok(allowed) =
37    runtime.execute(host, session, repair <> ", allow_partial=True)", 5000)
38  let assert Ok(allowed) = allowed.result
39  allowed.value |> should.equal("2")
40  runtime.stop(host)
41}
42
43pub fn ambiguous_repairs_and_compile_errors_do_not_execute_test() -> Nil {
44  let assert Ok(host) = runtime.start(":memory:")
45  session_fixture.initialise(host)
46  session_fixture.create(host, "a", "/tmp")
47  let assert Ok(session) = runtime.open_session(host, "a", "/tmp")
48  let assert Ok(failed) =
49    runtime.execute(host, session, "state = 1\n(yield 1)", 5000)
50  let assert Ok(original) = runtime.cell(host, failed.cell_id)
51  original.started |> should.be_false
52  let assert Ok(denied) =
53    runtime.execute(
54      host,
55      session,
56      "await cells.run('" <> original.id <> "', replacements=[('1', '2')])",
57      5000,
58    )
59  let assert Ok(denied) = denied.result
60  denied.status |> should.equal(python.Failed)
61  denied.output |> string.contains("exactly once") |> should.be_true
62  let assert Ok(spaced) =
63    runtime.execute(
64      host,
65      session,
66      "await cells.run('"
67        <> original.id
68        <> "', replacements=[('state   =\\n1', '2')])",
69      5000,
70    )
71  let assert Ok(spaced) = spaced.result
72  spaced.output |> string.contains("whitespace differs") |> should.be_true
73  let assert Ok(state) =
74    runtime.execute(host, session, "'state' in globals()", 5000)
75  let assert Ok(state) = state.result
76  state.value |> should.equal("False")
77  runtime.stop(host)
78}
79
80pub fn reused_call_id_does_not_fail_unique_constraint_test() -> Nil {
81  let assert Ok(host) = runtime.start(":memory:")
82  let storage = runtime.ledger(host)
83  let session = "test-session"
84  let assert Ok(first_id) =
85    cells.begin_call(storage, session, "call_0", "x = 1")
86  first_id |> should.equal(session <> "/call_0")
87
88  let assert Ok(second_id) =
89    cells.begin_call(storage, session, "call_0", "x = 2")
90  string.starts_with(second_id, session <> "/call_0-") |> should.be_true
91  { first_id != second_id } |> should.be_true
92
93  let assert Ok(found) = cells.find_call(storage, session <> "/call_0")
94  found.id |> should.equal(second_id)
95
96  runtime.stop(host)
97}