40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
FRONTEND_DIR = Path(__file__).resolve().parents[2] / "apps" / "web" / "vue3"
|
|
PAGE_NAMES = ("index.html", "index_ml.html")
|
|
PAGE_SOURCES = {
|
|
name: (FRONTEND_DIR / name).read_text(encoding="utf-8")
|
|
for name in PAGE_NAMES
|
|
}
|
|
|
|
|
|
def _source(name: str) -> str:
|
|
return PAGE_SOURCES[name]
|
|
|
|
|
|
@pytest.mark.parametrize("name", PAGE_NAMES)
|
|
def test_rag_mode_uses_streaming_request(name: str) -> None:
|
|
source = _source(name)
|
|
start = source.index("// V2: RAG模式")
|
|
end = source.index("} catch (err)", start)
|
|
rag_block = source[start:end]
|
|
|
|
assert "requestRagWithProgress(" in rag_block
|
|
assert "fetchJsonWithTimeout('/api/agent/rag'" not in rag_block
|
|
|
|
|
|
@pytest.mark.parametrize("name", PAGE_NAMES)
|
|
def test_rag_stream_timeout_is_reset_after_each_received_chunk(name: str) -> None:
|
|
source = _source(name)
|
|
start = source.index("const readRagEventStream")
|
|
end = source.index("const requestRagWithProgress", start)
|
|
stream_block = source[start:end]
|
|
read_position = stream_block.index("const { value, done } = await reader.read();")
|
|
|
|
assert "const resetIdleTimer = () =>" in stream_block
|
|
assert stream_block.index("resetIdleTimer();", read_position) > read_position
|
|
assert "知识库检索连接空闲超时" in stream_block
|