conftest.py — langchain Source File

Source Code

import os
from collections.abc import AsyncIterator, Iterator
from uuid import UUID

import pytest
from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.store.base import BaseStore
from pytest_mock import MockerFixture

from tests.unit_tests.agents.conftest_checkpointer import (
    _checkpointer_memory,
    _checkpointer_postgres,
    _checkpointer_postgres_aio,
    _checkpointer_postgres_aio_pipe,
    _checkpointer_postgres_aio_pool,
    _checkpointer_postgres_pipe,
    _checkpointer_postgres_pool,
    _checkpointer_sqlite,
    _checkpointer_sqlite_aio,
)
from tests.unit_tests.agents.conftest_store import (
    _store_memory,
    _store_postgres,
    _store_postgres_aio,
    _store_postgres_aio_pipe,
    _store_postgres_aio_pool,
    _store_postgres_pipe,
    _store_postgres_pool,
)

# Global variables for checkpointer and store configurations
FAST_MODE = os.getenv("LANGGRAPH_TEST_FAST", "true").lower() in {"true", "1", "yes"}

SYNC_CHECKPOINTER_PARAMS = (
    ["memory"]
    if FAST_MODE
    else [
        "memory",
        "sqlite",
        "postgres",
        "postgres_pipe",
        "postgres_pool",
    ]
)

ASYNC_CHECKPOINTER_PARAMS = (
    ["memory"]
    if FAST_MODE
    else [
        "memory",
        "sqlite_aio",
        "postgres_aio",
        "postgres_aio_pipe",
        "postgres_aio_pool",
    ]
)

SYNC_STORE_PARAMS = (
    ["in_memory"]
    if FAST_MODE
// ... (135 more lines)