diff --git a/libs/checkpoint-sqlite/tests/test_conformance_delta.py b/libs/checkpoint-sqlite/tests/test_conformance_delta.py new file mode 100644 index 000000000..b20cdf787 --- /dev/null +++ b/libs/checkpoint-sqlite/tests/test_conformance_delta.py @@ -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}") diff --git a/libs/checkpoint/tests/test_conformance_delta.py b/libs/checkpoint/tests/test_conformance_delta.py new file mode 100644 index 000000000..3385b0c43 --- /dev/null +++ b/libs/checkpoint/tests/test_conformance_delta.py @@ -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}")