test_send_to_closed_stream() — langchain Function Reference

Architecture documentation for the test_send_to_closed_stream() function in test_memory_stream.py from the langchain codebase.

Function python CoreAbstractions Serialization

Entity Profile

CoreAbstractions→ Serialization→ test_send_to_closed_stream() — langchain Function Reference

Dependency Diagram

graph TD
  ae3112f9_75af_a5d0_94d6_9b3fe5ac0b01["test_send_to_closed_stream()"]
  f5c9b9cd_4c4b_26c8_2419_4e2f783c706c["test_memory_stream.py"]
  ae3112f9_75af_a5d0_94d6_9b3fe5ac0b01 -->|defined in| f5c9b9cd_4c4b_26c8_2419_4e2f783c706c
  style ae3112f9_75af_a5d0_94d6_9b3fe5ac0b01 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/tracers/test_memory_stream.py lines 115–130

def test_send_to_closed_stream() -> None:
    """Test that sending to a closed stream doesn't raise an error.

    We may want to handle this in a better way in the future.
    """
    event_loop = asyncio.new_event_loop()
    channel = _MemoryStream[str](event_loop)
    writer = channel.get_send_stream()
    # send with an open even loop
    writer.send_nowait("hello")
    event_loop.close()
    writer.send_nowait("hello")
    # now close the loop
    event_loop.close()
    writer.close()
    writer.send_nowait("hello")