mirror of
https://github.com/hwchase17/langchain.git
synced 2025-08-10 05:20:39 +00:00
chore(langchain): strip_ansi
fucntion to remove ANSI escape sequences (#32200)
**Description:** Fixes a bug in the file callback test where ANSI escape codes were causing test failures. The improved test now properly handles ANSI escape sequences by: - Using exact string comparison instead of substring checking - Applying the `strip_ansi` function consistently to all file contents - Adding descriptive assertion messages - Maintaining test coverage and backward compatibility The changes ensure tests pass reliably even when terminal control sequences are present in the output **Issue:** Fixes #32150 **Dependencies:** None required - uses existing dependencies only. --------- Co-authored-by: Eugene Yurtsev <eyurtsev@gmail.com>
This commit is contained in:
parent
0d6f915442
commit
8f5ec20ccf
@ -1,4 +1,5 @@
|
|||||||
import pathlib
|
import pathlib
|
||||||
|
import re
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
from langchain_core.callbacks import CallbackManagerForChainRun
|
from langchain_core.callbacks import CallbackManagerForChainRun
|
||||||
@ -32,6 +33,17 @@ class FakeChain(Chain):
|
|||||||
return {"bar": "bar"}
|
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-?]*[ -/]*[@-~]")
|
||||||
|
return ansi_escape.sub("", text)
|
||||||
|
|
||||||
|
|
||||||
def test_filecallback(tmp_path: pathlib.Path) -> None:
|
def test_filecallback(tmp_path: pathlib.Path) -> None:
|
||||||
"""Test the file callback handler."""
|
"""Test the file callback handler."""
|
||||||
log1 = tmp_path / "output.log"
|
log1 = tmp_path / "output.log"
|
||||||
@ -40,7 +52,7 @@ def test_filecallback(tmp_path: pathlib.Path) -> None:
|
|||||||
chain_test.invoke({"foo": "bar"})
|
chain_test.invoke({"foo": "bar"})
|
||||||
handler.close()
|
handler.close()
|
||||||
# Assert the output is as expected
|
# 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
|
# Test using a callback manager
|
||||||
log2 = tmp_path / "output2.log"
|
log2 = tmp_path / "output2.log"
|
||||||
@ -49,12 +61,11 @@ def test_filecallback(tmp_path: pathlib.Path) -> None:
|
|||||||
chain_test = FakeChain(callbacks=[handler_cm])
|
chain_test = FakeChain(callbacks=[handler_cm])
|
||||||
chain_test.invoke({"foo": "bar"})
|
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
|
# Test passing via invoke callbacks
|
||||||
|
|
||||||
log3 = tmp_path / "output3.log"
|
log3 = tmp_path / "output3.log"
|
||||||
|
|
||||||
with FileCallbackHandler(str(log3)) as handler_cm:
|
with FileCallbackHandler(str(log3)) as handler_cm:
|
||||||
chain_test.invoke({"foo": "bar"}, {"callbacks": [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…
Reference in New Issue
Block a user