Quick start
Register your checkpointer with @checkpointer_test and
run validate():
import asyncio
from langgraph.checkpoint.conformance import checkpointer_test, validate
@checkpointer_test(name="MyCheckpointer")
async def my_checkpointer():
saver = MyCheckpointer(...)
yield saver
# cleanup runs after yield
async def main():
report = await validate(my_checkpointer)
report.print_report()
assert report.passed_all_base()
asyncio.run(main())
Or in a pytest test:
import pytest
from langgraph.checkpoint.conformance import checkpointer_test, validate
@checkpointer_test(name="MyCheckpointer")
async def my_checkpointer():
yield MyCheckpointer(...)
@pytest.mark.asyncio
async def test_conformance():
report = await validate(my_checkpointer)
report.print_report()
assert report.passed_all_base()