chore(prebuilt): Remove dead code from prebuilt tests (#5555)

Fixes #5554

This PR removes unused utility classes and functions from the prebuilt
tests directory to clean up dead code.

Changes include:
- Removed unused classes from `libs/prebuilt/tests/any_str.py`:
  - Deleted FloatBetween, AnyDict, AnyVersion, and UnsortedSequence
  - Kept only AnyStr class

- Removed unused functions from `libs/prebuilt/tests/messages.py`:
  - Deleted _AnyIdDocument and _AnyIdAIMessageChunk
  - Kept _AnyIdHumanMessage and _AnyIdToolMessage

- Removed unused classes from `libs/prebuilt/tests/memory_assert.py`:
- Deleted NoopSerializer, MemorySaverAssertCheckpointMetadata, and
MemorySaverNoPending
  - Kept MemorySaverAssertImmutable

Verification:
- Manually checked for no remaining references to removed code
- Maintained existing import structures
- Preserved functionality of the prebuilt test suite

The changes reduce code complexity and remove unnecessary utility
classes that were not being used in the test suite.

---------

Co-authored-by: open-swe-dev[bot] <open-swe-dev@users.noreply.github.com>
Co-authored-by: Eugene Yurtsev <eyurtsev@gmail.com>
This commit is contained in:
open-swe[bot]
2025-07-17 18:45:10 +00:00
committed by GitHub
co-authored by open-swe-dev[bot] <open-swe-dev@users.noreply.github.com> Eugene Yurtsev
parent b4eb57da67
commit f87608a16f
4 changed files with 4 additions and 169 deletions
+1 -76
View File
@@ -1,31 +1,19 @@
import asyncio
import os
import tempfile
from collections import defaultdict
from functools import partial
from typing import Any, Optional
from langchain_core.runnables import RunnableConfig
from typing import Optional
from langgraph.checkpoint.base import (
ChannelVersions,
Checkpoint,
CheckpointMetadata,
CheckpointTuple,
SerializerProtocol,
)
from langgraph.checkpoint.memory import InMemorySaver, PersistentDict
from langgraph.pregel.checkpoint import copy_checkpoint
class NoopSerializer(SerializerProtocol):
def loads_typed(self, data: tuple[str, bytes]) -> Any:
return data[1]
def dumps_typed(self, obj: Any) -> tuple[str, bytes]:
return "type", obj
class MemorySaverAssertImmutable(InMemorySaver):
storage_for_copies: defaultdict[str, dict[str, dict[str, Checkpoint]]]
@@ -69,66 +57,3 @@ class MemorySaverAssertImmutable(InMemorySaver):
)
# call super to write checkpoint
return super().put(config, checkpoint, metadata, new_versions)
class MemorySaverAssertCheckpointMetadata(InMemorySaver):
"""This custom checkpointer is for verifying that a run's configurable
fields are merged with the previous checkpoint config for each step in
the run. This is the desired behavior. Because the checkpointer's (a)put()
method is called for each step, the implementation of this checkpointer
should produce a side effect that can be asserted.
"""
def put(
self,
config: RunnableConfig,
checkpoint: Checkpoint,
metadata: CheckpointMetadata,
new_versions: ChannelVersions,
) -> None:
"""The implementation of put() merges config["configurable"] (a run's
configurable fields) with the metadata field. The state of the
checkpoint metadata can be asserted to confirm that the run's
configurable fields were merged with the previous checkpoint config.
"""
configurable = config["configurable"].copy()
# remove checkpoint_id to make testing simpler
checkpoint_id = configurable.pop("checkpoint_id", None)
thread_id = config["configurable"]["thread_id"]
checkpoint_ns = config["configurable"]["checkpoint_ns"]
self.storage[thread_id][checkpoint_ns].update(
{
checkpoint["id"]: (
self.serde.dumps_typed(checkpoint),
# merge configurable fields and metadata
self.serde.dumps_typed({**configurable, **metadata}),
checkpoint_id,
)
}
)
return {
"configurable": {
"thread_id": config["configurable"]["thread_id"],
"checkpoint_id": checkpoint["id"],
}
}
async def aput(
self,
config: RunnableConfig,
checkpoint: Checkpoint,
metadata: CheckpointMetadata,
new_versions: ChannelVersions,
) -> RunnableConfig:
return await asyncio.get_running_loop().run_in_executor(
None, self.put, config, checkpoint, metadata, new_versions
)
class MemorySaverNoPending(InMemorySaver):
def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
result = super().get_tuple(config)
if result:
return CheckpointTuple(result.config, result.checkpoint, result.metadata)
return result