mirror of
https://github.com/NCBM/plyngent.git
synced 2026-07-25 08:04:57 +08:00
49 lines
1.6 KiB
Python
49 lines
1.6 KiB
Python
|
"""PersistentDataView transaction behavior."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from plyngent.agent.todo_stack import TodoStack
|
||
|
|
from plyngent.tools.view import MemoryViewStore, PersistentDataView
|
||
|
|
|
||
|
|
|
||
|
|
async def test_view_txn_commit() -> None:
|
||
|
|
store = MemoryViewStore({})
|
||
|
|
root: PersistentDataView[dict[str, object]] = PersistentDataView(store, bound_type=dict)
|
||
|
|
async with root as data:
|
||
|
|
data["todo"].store({"groups": [], "next_id": 1})
|
||
|
|
loaded = await store.load()
|
||
|
|
assert isinstance(loaded, dict)
|
||
|
|
assert "todo" in loaded
|
||
|
|
|
||
|
|
|
||
|
|
async def test_view_rollback_on_error() -> None:
|
||
|
|
store = MemoryViewStore({"keep": 1})
|
||
|
|
root: PersistentDataView[dict[str, object]] = PersistentDataView(store)
|
||
|
|
with pytest.raises(RuntimeError, match="boom"):
|
||
|
|
async with root as data:
|
||
|
|
data.store({"keep": 2})
|
||
|
|
msg = "boom"
|
||
|
|
raise RuntimeError(msg)
|
||
|
|
assert await store.load() == {"keep": 1}
|
||
|
|
|
||
|
|
|
||
|
|
async def test_typed_todo_stack() -> None:
|
||
|
|
store = MemoryViewStore({})
|
||
|
|
root: PersistentDataView[dict[str, object]] = PersistentDataView(store, bound_type=dict)
|
||
|
|
async with root as data:
|
||
|
|
todo = data["todo"].typed(TodoStack)
|
||
|
|
_ = todo.push_group(["A", "B"])
|
||
|
|
assert todo.depth == 1
|
||
|
|
loaded = await store.load()
|
||
|
|
assert isinstance(loaded, dict)
|
||
|
|
assert isinstance(loaded.get("todo"), TodoStack)
|
||
|
|
|
||
|
|
|
||
|
|
async def test_mutate_outside_txn_errors() -> None:
|
||
|
|
store = MemoryViewStore({})
|
||
|
|
root: PersistentDataView[dict[str, object]] = PersistentDataView(store)
|
||
|
|
with pytest.raises(RuntimeError, match="transaction"):
|
||
|
|
root.store({"a": 1})
|