test: verify real kbqa workflow
This commit is contained in:
@@ -0,0 +1,13 @@
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from kbqa.api.errors import AppError
|
||||
from kbqa.indexing.loaders import load_document
|
||||
|
||||
|
||||
def test_text_loader_rejects_non_utf8(tmp_path: Path) -> None:
|
||||
path = tmp_path / "bad.txt"
|
||||
path.write_bytes(b"\xff\xfe")
|
||||
with pytest.raises(AppError, match="UTF-8"):
|
||||
load_document(path, "txt")
|
||||
@@ -0,0 +1,12 @@
|
||||
from kbqa.indexing.loaders import LoadedPage
|
||||
from kbqa.indexing.splitter import split_pages
|
||||
|
||||
|
||||
def test_split_pages_is_deterministic() -> None:
|
||||
document_id = "12345678-1234-5678-1234-567812345678"
|
||||
pages = [LoadedPage(content="知识库" * 300, page_number=1)]
|
||||
first = split_pages(document_id, pages, 100, 10)
|
||||
second = split_pages(document_id, pages, 100, 10)
|
||||
assert [chunk.id for chunk in first] == [chunk.id for chunk in second]
|
||||
assert first[0].page_number == 1
|
||||
assert [chunk.order_index for chunk in first] == list(range(len(first)))
|
||||
@@ -0,0 +1,8 @@
|
||||
from kbqa.chat.sse import encode_event, encode_ping
|
||||
|
||||
|
||||
def test_sse_encodes_single_line_utf8_json() -> None:
|
||||
assert encode_event("token", {"content": "知识"}) == (
|
||||
'event: token\ndata: {"content":"知识"}\n\n'.encode()
|
||||
)
|
||||
assert encode_ping() == b": ping\n\n"
|
||||
Reference in New Issue
Block a user