test: add delta-channel conformance wrappers for InMemory and SQLite savers

Runs the three new delta-channel conformance capabilities
(delta_channel_history, delta_channel_keepset, delta_channel_reconstruction)
against InMemorySaver and AsyncSqliteSaver. Guarded by importorskip so
they're skipped gracefully when checkpoint-conformance isn't installed.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Quanzheng Long
2026-05-07 11:15:34 -07:00
co-authored by Cursor
parent d6c29f8157
commit 9dc68d6986
2 changed files with 74 additions and 0 deletions
@@ -0,0 +1,37 @@
"""Run delta-channel conformance capabilities against AsyncSqliteSaver."""
from __future__ import annotations
import pytest
pytest.importorskip(
"langgraph.checkpoint.conformance",
reason="langgraph-checkpoint-conformance not installed",
)
pytest.importorskip("aiosqlite", reason="aiosqlite not installed")
from langgraph.checkpoint.conformance import validate # noqa: E402
from langgraph.checkpoint.conformance.initializer import checkpointer_test # noqa: E402
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver # noqa: E402
@checkpointer_test(name="AsyncSqliteSaver")
async def _sqlite_saver():
async with AsyncSqliteSaver.from_conn_string(":memory:") as saver:
yield saver
@pytest.mark.asyncio
async def test_delta_channel_conformance():
report = await validate(
_sqlite_saver,
capabilities={
"delta_channel_history",
"delta_channel_keepset",
"delta_channel_reconstruction",
},
)
for cap, result in report.results.items():
if result.passed is False:
details = "\n".join(result.failures or [])
pytest.fail(f"Capability {cap} failed:\n{details}")
@@ -0,0 +1,37 @@
"""Run delta-channel conformance capabilities against InMemorySaver."""
from __future__ import annotations
import asyncio
import pytest
pytest.importorskip(
"langgraph.checkpoint.conformance",
reason="langgraph-checkpoint-conformance not installed",
)
from langgraph.checkpoint.conformance import validate # noqa: E402
from langgraph.checkpoint.conformance.initializer import checkpointer_test # noqa: E402
from langgraph.checkpoint.memory import InMemorySaver # noqa: E402
@checkpointer_test(name="InMemorySaver")
async def _mem_saver():
yield InMemorySaver()
@pytest.mark.asyncio
async def test_delta_channel_conformance():
report = await validate(
_mem_saver,
capabilities={
"delta_channel_history",
"delta_channel_keepset",
"delta_channel_reconstruction",
},
)
for cap, result in report.results.items():
if result.passed is False:
details = "\n".join(result.failures or [])
pytest.fail(f"Capability {cap} failed:\n{details}")