wehub-resource-sync
a7d6d88f6f
CI / changes (push) Has been cancelled
CI / cd libs/checkpoint (push) Has been cancelled
CI / cd libs/checkpoint-conformance (push) Has been cancelled
CI / cd libs/checkpoint-postgres (push) Has been cancelled
CI / cd libs/checkpoint-sqlite (push) Has been cancelled
CI / cd libs/cli (push) Has been cancelled
CI / cd libs/prebuilt (push) Has been cancelled
CI / cd libs/sdk-py (push) Has been cancelled
CI / cd libs/langgraph (push) Has been cancelled
CI / Check SDK methods matching (push) Has been cancelled
CI / Check CLI schema hasn't changed #3.13 (push) Has been cancelled
CI / CLI integration test (push) Has been cancelled
CI / sdk-py integration test (push) Has been cancelled
CI / CI Success (push) Has been cancelled
baseline / benchmark (push) Has been cancelled
Deploy Redirects to GitHub Pages / deploy (push) Has been cancelled
36 lines
1.1 KiB
Python
36 lines
1.1 KiB
Python
"""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")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delta_channel_conformance():
|
|
from langgraph.checkpoint.conformance import validate
|
|
from langgraph.checkpoint.conformance.initializer import checkpointer_test
|
|
|
|
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
|
|
|
|
@checkpointer_test(name="AsyncSqliteSaver")
|
|
async def sqlite_saver():
|
|
async with AsyncSqliteSaver.from_conn_string(":memory:") as saver:
|
|
yield saver
|
|
|
|
report = await validate(
|
|
sqlite_saver,
|
|
capabilities={
|
|
"delta_channel_history",
|
|
},
|
|
)
|
|
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}")
|