Skip to content
Merged
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 15 additions & 4 deletions libs/langchain/tests/unit_tests/callbacks/test_file.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import pathlib
import re
from typing import Optional

from langchain_core.callbacks import CallbackManagerForChainRun
Expand Down Expand Up @@ -32,6 +33,17 @@
return {"bar": "bar"}


def strip_ansi(text: str) -> str:
"""
Removes ANSI escape sequences from a string.

Args:
text: The string potentially containing ANSI codes.
"""
ansi_escape = re.compile(r'\x1B\[[0-?]*[ -/]*[@-~]')

Check failure on line 43 in libs/langchain/tests/unit_tests/callbacks/test_file.py

View workflow job for this annotation

GitHub Actions / cd libs/langchain / make lint #3.13

Ruff (Q000)

tests/unit_tests/callbacks/test_file.py:43:30: Q000 Single quotes found but double quotes preferred

Check failure on line 43 in libs/langchain/tests/unit_tests/callbacks/test_file.py

View workflow job for this annotation

GitHub Actions / cd libs/langchain / make lint #3.9

Ruff (Q000)

tests/unit_tests/callbacks/test_file.py:43:30: Q000 Single quotes found but double quotes preferred
return ansi_escape.sub('', text)

Check failure on line 44 in libs/langchain/tests/unit_tests/callbacks/test_file.py

View workflow job for this annotation

GitHub Actions / cd libs/langchain / make lint #3.13

Ruff (Q000)

tests/unit_tests/callbacks/test_file.py:44:28: Q000 Single quotes found but double quotes preferred

Check failure on line 44 in libs/langchain/tests/unit_tests/callbacks/test_file.py

View workflow job for this annotation

GitHub Actions / cd libs/langchain / make lint #3.9

Ruff (Q000)

tests/unit_tests/callbacks/test_file.py:44:28: Q000 Single quotes found but double quotes preferred


def test_filecallback(tmp_path: pathlib.Path) -> None:
"""Test the file callback handler."""
log1 = tmp_path / "output.log"
Expand All @@ -40,7 +52,7 @@
chain_test.invoke({"foo": "bar"})
handler.close()
# Assert the output is as expected
assert "Entering new FakeChain chain" in log1.read_text()
assert "Entering new FakeChain chain" in strip_ansi(log1.read_text())

# Test using a callback manager
log2 = tmp_path / "output2.log"
Expand All @@ -49,12 +61,11 @@
chain_test = FakeChain(callbacks=[handler_cm])
chain_test.invoke({"foo": "bar"})

assert "Entering new FakeChain chain" in log2.read_text()
assert "Entering new FakeChain chain" in strip_ansi(log2.read_text())

# Test passing via invoke callbacks

log3 = tmp_path / "output3.log"

with FileCallbackHandler(str(log3)) as handler_cm:
chain_test.invoke({"foo": "bar"}, {"callbacks": [handler_cm]})
assert "Entering new FakeChain chain" in log3.read_text()
assert "Entering new FakeChain chain" in strip_ansi(log3.read_text())
Loading