mirror of
https://github.com/TencentCloud/Octop.git
synced 2026-10-20 06:19:17 +08:00
IronMurphy
e084e9c68a
* fix(knowledge): tell the user a restart is what the install needs When the local embedding packages land in pip's user site, CPython only puts that directory on sys.path if it existed at interpreter startup, so a long-running server can never import them. ensure_local_embedding_deps() already raises with 'Restart the server and try again', but _map_knowledge_error() matched the text on 'embedding'/'component' and replaced it with the generic prerequisites error, so the user retried a download that could not succeed. Map that case to its own code and message. * test(i18n): register the new code in the dashboard locales too tests/unit/i18n/test_errors.py asserts the dashboard apiErrors keys equal the backend ErrorCode set, so the new restart-required code has to be listed there as well.
762 lines
25 KiB
Python
762 lines
25 KiB
Python
"""Unit tests for the knowledge-base HTTP adapter."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from starlette.datastructures import Headers, UploadFile
|
|
from starlette.requests import Request
|
|
|
|
from octop.infra.errors import ErrorCode, OctopError
|
|
|
|
|
|
def _request() -> Request:
|
|
return Request(
|
|
{"type": "http", "method": "POST", "path": "/api/knowledge-bases", "headers": []}
|
|
)
|
|
|
|
|
|
def _services(**extra: object) -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
settings_repo=SimpleNamespace(get=lambda _key: None, set=lambda *_: None),
|
|
provider_repo=SimpleNamespace(list_all=lambda: []),
|
|
**extra,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class _Base:
|
|
id: str = "kb-1"
|
|
pk: int = 1
|
|
owner_user_id: int = 1
|
|
name: str = "Docs"
|
|
description: str = ""
|
|
default_open: bool = False
|
|
shared: bool = False
|
|
icon_name: str = ""
|
|
embedding_model: str = "model"
|
|
embedding_dim: int = 0
|
|
doc_count: int = 0
|
|
max_documents: int = 100
|
|
created_at: int = 1
|
|
updated_at: int = 1
|
|
|
|
|
|
@dataclass
|
|
class _Document:
|
|
id: str = "doc-1"
|
|
pk: int = 1
|
|
kb_id: str = "kb-1"
|
|
path: str = "readme.md"
|
|
filename: str = "readme.md"
|
|
is_dir: bool = False
|
|
content_type: str = "text/markdown"
|
|
byte_size: int = 2
|
|
content_hash: str = ""
|
|
status: str = "pending"
|
|
error_message: str = ""
|
|
chunk_count: int = 0
|
|
created_at: int = 1
|
|
updated_at: int = 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enabling_without_model_maps_prerequisite_error(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
server = SimpleNamespace(services=_services())
|
|
|
|
def fail(*_args: object, **_kwargs: object) -> None:
|
|
raise ValueError("enabling knowledge bases requires an embedding model")
|
|
|
|
monkeypatch.setattr(knowledge_bases, "set_feature_enabled", fail)
|
|
|
|
with pytest.raises(OctopError) as raised:
|
|
await knowledge_bases.put_feature(
|
|
knowledge_bases.FeatureBody(enabled=True),
|
|
request=_request(),
|
|
server=server,
|
|
_admin=object(),
|
|
)
|
|
|
|
assert raised.value.code == ErrorCode.KNOWLEDGE_PREREQUISITES_FAILED
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enabling_reindexes_documents_when_model_changes(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
settings = {
|
|
"knowledge_bases_enabled": "true",
|
|
"knowledge_embedding_model": "old-model",
|
|
"knowledge_embedding_backend": "onnx",
|
|
"knowledge_embedding_provider_id": "",
|
|
}
|
|
calls: list[tuple[object, str]] = []
|
|
server = SimpleNamespace(
|
|
services=SimpleNamespace(
|
|
settings_repo=SimpleNamespace(get=settings.get, set=settings.__setitem__),
|
|
provider_repo=SimpleNamespace(list_all=lambda: []),
|
|
)
|
|
)
|
|
|
|
def set_feature(
|
|
_get: object,
|
|
settings_set: object,
|
|
*,
|
|
enabled: bool,
|
|
model: str | None,
|
|
backend: str | None = None,
|
|
provider_id: str | None = None,
|
|
provider_repo: object = None,
|
|
) -> None:
|
|
assert enabled is True
|
|
assert model == "new-model"
|
|
settings_set("knowledge_embedding_model", model)
|
|
|
|
monkeypatch.setattr(knowledge_bases, "set_feature_enabled", set_feature)
|
|
monkeypatch.setattr(
|
|
knowledge_bases,
|
|
"reindex_all_documents",
|
|
lambda services, model: calls.append((services, model)),
|
|
)
|
|
monkeypatch.setattr(
|
|
knowledge_bases,
|
|
"get_capability",
|
|
lambda get, _provider_repo=None: {
|
|
"feature_enabled": get("knowledge_bases_enabled") == "true",
|
|
"selected_model": get("knowledge_embedding_model"),
|
|
"backend": get("knowledge_embedding_backend") or "onnx",
|
|
"provider_id": get("knowledge_embedding_provider_id") or "",
|
|
},
|
|
)
|
|
|
|
await knowledge_bases.put_feature(
|
|
knowledge_bases.FeatureBody(enabled=True, model="new-model"),
|
|
request=_request(),
|
|
server=server,
|
|
_admin=object(),
|
|
)
|
|
|
|
assert calls == [(server.services, "new-model")]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_base_maps_disabled_feature_error(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
server = SimpleNamespace(services=_services())
|
|
user = SimpleNamespace(id=1, is_admin=False)
|
|
|
|
def fail(*_args: object, **_kwargs: object) -> None:
|
|
raise RuntimeError("knowledge feature is disabled")
|
|
|
|
monkeypatch.setattr(knowledge_bases, "assert_knowledge_usable", fail)
|
|
|
|
with pytest.raises(OctopError) as raised:
|
|
await knowledge_bases.create_base(
|
|
knowledge_bases.CreateBaseBody(name="Docs"),
|
|
request=_request(),
|
|
server=server,
|
|
user=user,
|
|
)
|
|
|
|
assert raised.value.code == ErrorCode.KNOWLEDGE_FEATURE_DISABLED
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_base_uses_selected_model_when_usable(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
created: dict[str, object] = {}
|
|
service = SimpleNamespace(
|
|
create_base=lambda **kwargs: created.update(kwargs) or _Base(),
|
|
)
|
|
server = SimpleNamespace(services=_services())
|
|
user = SimpleNamespace(id=1, is_admin=False)
|
|
monkeypatch.setattr(knowledge_bases, "assert_knowledge_usable", lambda *_a, **_k: None)
|
|
monkeypatch.setattr(knowledge_bases, "_knowledge_service", lambda _server: service)
|
|
|
|
response = await knowledge_bases.create_base(
|
|
knowledge_bases.CreateBaseBody(name="Docs"),
|
|
request=_request(),
|
|
server=server,
|
|
user=user,
|
|
)
|
|
|
|
assert response["id"] == "kb-1"
|
|
assert created == {
|
|
"owner_user_id": 1,
|
|
"name": "Docs",
|
|
"description": "",
|
|
"default_open": False,
|
|
"shared": False,
|
|
"icon_name": "",
|
|
"max_documents": 100,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_enqueues_document_indexing(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
calls: list[tuple[object, str, str]] = []
|
|
service = SimpleNamespace(
|
|
upload_document=lambda *_args, **_kwargs: _Document(),
|
|
)
|
|
server = SimpleNamespace(services=_services())
|
|
user = SimpleNamespace(id=1, is_admin=False)
|
|
upload = UploadFile(
|
|
file=__import__("io").BytesIO(b"# hi"),
|
|
filename="readme.md",
|
|
headers=Headers({"content-type": "text/markdown"}),
|
|
)
|
|
monkeypatch.setattr(knowledge_bases, "_knowledge_service", lambda _server: service)
|
|
monkeypatch.setattr(knowledge_bases, "assert_knowledge_usable", lambda *_a, **_k: None)
|
|
monkeypatch.setattr(
|
|
knowledge_bases,
|
|
"enqueue_index_document",
|
|
lambda services, kb_id, doc_id: calls.append((services, kb_id, doc_id)),
|
|
)
|
|
|
|
response = await knowledge_bases.upload_document(
|
|
"kb-1", request=_request(), upload=upload, server=server, user=user
|
|
)
|
|
|
|
assert response["id"] == "doc-1"
|
|
assert calls == [(server.services, "kb-1", "doc-1")]
|
|
|
|
|
|
def test_base_payload_includes_owner_display_fields() -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
owner = SimpleNamespace(username="Test", display_name=None)
|
|
server = SimpleNamespace(
|
|
services=SimpleNamespace(user_repo=SimpleNamespace(get=lambda _id: owner))
|
|
)
|
|
base = _Base(owner_user_id=7)
|
|
|
|
payload = knowledge_bases._base_payload(server, base)
|
|
|
|
assert payload["owner_user_id"] == 7
|
|
assert payload["owner_username"] == "Test"
|
|
assert payload["owner_display_name"] == "Test"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_preview_document_returns_extracted_text(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
service = SimpleNamespace(
|
|
preview_document=lambda *_args, **_kwargs: {
|
|
"id": "doc-1",
|
|
"filename": "notes.md",
|
|
"text": "# Hello",
|
|
}
|
|
)
|
|
server = SimpleNamespace(services=_services())
|
|
user = SimpleNamespace(id=1, is_admin=False)
|
|
monkeypatch.setattr(knowledge_bases, "_knowledge_service", lambda _server: service)
|
|
|
|
response = await knowledge_bases.preview_document(
|
|
"kb-1", "doc-1", request=_request(), server=server, user=user
|
|
)
|
|
|
|
assert response == {"id": "doc-1", "filename": "notes.md", "text": "# Hello"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_document_file_returns_original_bytes(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
path = tmp_path / "doc.pdf"
|
|
path.write_bytes(b"%PDF-1.4 test")
|
|
service = SimpleNamespace(
|
|
resolve_document_file=lambda *_args, **_kwargs: (
|
|
path,
|
|
"报告.pdf",
|
|
"application/pdf",
|
|
)
|
|
)
|
|
server = SimpleNamespace(services=_services())
|
|
user = SimpleNamespace(id=1, is_admin=False)
|
|
monkeypatch.setattr(knowledge_bases, "_knowledge_service", lambda _server: service)
|
|
|
|
response = await knowledge_bases.download_document_file(
|
|
"kb-1",
|
|
"doc-1",
|
|
request=_request(),
|
|
disposition="inline",
|
|
server=server,
|
|
user=user,
|
|
)
|
|
|
|
assert response.path == path
|
|
assert response.media_type == "application/pdf"
|
|
assert "inline" in response.headers["Content-Disposition"]
|
|
assert "filename*" in response.headers["Content-Disposition"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_document_file_missing_maps_not_found(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
def missing(*_args: object, **_kwargs: object) -> None:
|
|
raise FileNotFoundError("knowledge document original file not found")
|
|
|
|
service = SimpleNamespace(resolve_document_file=missing)
|
|
server = SimpleNamespace(services=_services())
|
|
user = SimpleNamespace(id=1, is_admin=False)
|
|
monkeypatch.setattr(knowledge_bases, "_knowledge_service", lambda _server: service)
|
|
|
|
with pytest.raises(OctopError) as raised:
|
|
await knowledge_bases.download_document_file(
|
|
"kb-1",
|
|
"doc-1",
|
|
request=_request(),
|
|
disposition="attachment",
|
|
server=server,
|
|
user=user,
|
|
)
|
|
|
|
assert raised.value.code == ErrorCode.KNOWLEDGE_NOT_FOUND
|
|
|
|
|
|
def test_row_payload_includes_has_original(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
from octop.infra.knowledge import files as knowledge_files
|
|
|
|
monkeypatch.setattr(
|
|
knowledge_files,
|
|
"documents_dir",
|
|
lambda _kb_id: tmp_path,
|
|
)
|
|
doc = _Document(filename="notes.md")
|
|
missing = knowledge_bases._row_payload(doc)
|
|
assert missing["has_original"] is False
|
|
|
|
(tmp_path / f"{doc.id}.md").write_text("hi", encoding="utf-8")
|
|
present = knowledge_bases._row_payload(doc)
|
|
assert present["has_original"] is True
|
|
assert present["document_id"] == doc.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_embedding_options_excludes_onnx_local_provider() -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
providers = [
|
|
SimpleNamespace(
|
|
id=1,
|
|
name="ONNX (Local)",
|
|
api_key="onnx",
|
|
get_models=lambda: [
|
|
{
|
|
"id": "BAAI/bge-small-zh-v1.5",
|
|
"name": "bge",
|
|
"embedding": True,
|
|
}
|
|
],
|
|
),
|
|
SimpleNamespace(
|
|
id=2,
|
|
name="OpenAI",
|
|
api_key="sk-test",
|
|
get_models=lambda: [
|
|
{"id": "text-embedding-3-small", "name": "embed", "embedding": True},
|
|
{"id": "gpt-4o", "name": "GPT-4o"},
|
|
],
|
|
),
|
|
]
|
|
server = SimpleNamespace(
|
|
services=SimpleNamespace(provider_repo=SimpleNamespace(list_all=lambda: providers))
|
|
)
|
|
|
|
options = await knowledge_bases.embedding_options(server=server, _admin=object())
|
|
|
|
assert [row["provider_name"] for row in options["remote"]] == ["OpenAI"]
|
|
assert options["remote"][0]["models"] == [{"id": "text-embedding-3-small", "name": "embed"}]
|
|
assert isinstance(options["onnx"], list)
|
|
from octop.infra.agents.providers.onnx_catalog import ONNX_PRESET_MODEL_IDS
|
|
|
|
onnx_ids = [row["id"] for row in options["onnx"]]
|
|
assert onnx_ids == list(ONNX_PRESET_MODEL_IDS)
|
|
assert all(row.get("recommended") for row in options["onnx"])
|
|
assert all("downloaded" in row for row in options["onnx"])
|
|
assert any(row.get("size_gb") for row in options["onnx"])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ocr_options_only_lists_image_capable_models() -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
provider = SimpleNamespace(
|
|
id=2,
|
|
name="OpenAI",
|
|
enabled=True,
|
|
api_key="sk-test",
|
|
base_url="https://example.test/v1",
|
|
get_models=lambda: [
|
|
{"id": "text-only", "name": "Text", "input": ["text"]},
|
|
{"id": "vision-1", "name": "Vision", "input": ["text", "image"]},
|
|
{"id": "embed-1", "name": "Embed", "embedding": True, "input": ["image"]},
|
|
],
|
|
)
|
|
server = SimpleNamespace(
|
|
services=SimpleNamespace(provider_repo=SimpleNamespace(list_all=lambda: [provider]))
|
|
)
|
|
|
|
options = await knowledge_bases.ocr_options(server=server, _admin=object())
|
|
|
|
assert options["local"] == {"id": "rapidocr", "name": "RapidOCR (ONNX)"}
|
|
assert options["remote"] == [
|
|
{
|
|
"provider_id": "2",
|
|
"provider_name": "OpenAI",
|
|
"models": [{"id": "vision-1", "name": "Vision"}],
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_embedding_options_omits_non_recommended_onnx() -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
from octop.infra.agents.providers.onnx_catalog import ONNX_PRESET_MODEL_IDS
|
|
|
|
extra = "BAAI/bge-small-en-v1.5"
|
|
server = SimpleNamespace(
|
|
services=SimpleNamespace(
|
|
provider_repo=SimpleNamespace(list_all=lambda: []),
|
|
settings_repo=SimpleNamespace(
|
|
get=lambda key: extra if key == "knowledge_embedding_model" else None
|
|
),
|
|
)
|
|
)
|
|
|
|
options = await knowledge_bases.embedding_options(server=server, _admin=object())
|
|
|
|
assert extra not in [row["id"] for row in options["onnx"]]
|
|
assert [row["id"] for row in options["onnx"]] == list(ONNX_PRESET_MODEL_IDS)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_embedding_options_all_onnx_includes_extras() -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
from octop.infra.agents.providers.onnx_catalog import ONNX_PRESET_MODEL_IDS
|
|
|
|
server = SimpleNamespace(
|
|
services=SimpleNamespace(provider_repo=SimpleNamespace(list_all=lambda: []))
|
|
)
|
|
|
|
options = await knowledge_bases.embedding_options(all_onnx=True, server=server, _admin=object())
|
|
|
|
ids = [row["id"] for row in options["onnx"]]
|
|
assert ids[: len(ONNX_PRESET_MODEL_IDS)] == list(ONNX_PRESET_MODEL_IDS)
|
|
assert len(ids) > len(ONNX_PRESET_MODEL_IDS)
|
|
assert any(not row.get("recommended") for row in options["onnx"])
|
|
assert all("downloaded" in row for row in options["onnx"])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reindex_document_enqueues_job(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
calls: list[tuple[object, str, str]] = []
|
|
service = SimpleNamespace(
|
|
reindex_document=lambda *_args, **_kwargs: _Document(status="pending"),
|
|
)
|
|
server = SimpleNamespace(services=_services())
|
|
user = SimpleNamespace(id=1, is_admin=False)
|
|
monkeypatch.setattr(knowledge_bases, "_knowledge_service", lambda _server: service)
|
|
monkeypatch.setattr(knowledge_bases, "_require_usable", lambda *_a, **_k: None)
|
|
monkeypatch.setattr(
|
|
knowledge_bases,
|
|
"enqueue_index_document",
|
|
lambda services, kb_id, doc_id: calls.append((services, kb_id, doc_id)),
|
|
)
|
|
|
|
response = await knowledge_bases.reindex_document(
|
|
"kb-1", "doc-1", request=_request(), server=server, user=user
|
|
)
|
|
|
|
assert response["status"] == "pending"
|
|
assert calls == [(server.services, "kb-1", "doc-1")]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_onnx_download_starts_catalog_model(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
class _State:
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"status": "downloading",
|
|
"progress": 0.1,
|
|
"model_name": "BAAI/bge-small-zh-v1.5",
|
|
"error": None,
|
|
}
|
|
|
|
async def start(model: str) -> _State:
|
|
assert model == "BAAI/bge-small-zh-v1.5"
|
|
return _State()
|
|
|
|
monkeypatch.setattr(knowledge_bases.DOWNLOAD_MANAGER, "start_download", start)
|
|
|
|
response = await knowledge_bases.start_onnx_download(
|
|
knowledge_bases.OnnxDownloadBody(model="BAAI/bge-small-zh-v1.5"),
|
|
request=_request(),
|
|
_=object(),
|
|
)
|
|
|
|
assert response["status"] == "downloading"
|
|
assert response["model_name"] == "BAAI/bge-small-zh-v1.5"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_onnx_activate_enables_downloaded_service(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
saved: dict[str, object] = {}
|
|
|
|
async def ready(**_kwargs: object) -> str:
|
|
return "ready"
|
|
|
|
monkeypatch.setattr(knowledge_bases, "ensure_local_embedding_deps_async", ready)
|
|
monkeypatch.setattr(knowledge_bases, "assert_catalog_model", lambda model: model)
|
|
monkeypatch.setattr(knowledge_bases, "is_model_downloaded", lambda _model: True)
|
|
monkeypatch.setattr(
|
|
knowledge_bases,
|
|
"save_config",
|
|
lambda _setter, config: saved.update(config.to_dict()) or config,
|
|
)
|
|
monkeypatch.setattr(
|
|
knowledge_bases,
|
|
"status_payload",
|
|
lambda *_args, **_kwargs: {"enabled": True, "model": "BAAI/bge-small-zh-v1.5"},
|
|
)
|
|
server = SimpleNamespace(services=_services())
|
|
|
|
response = await knowledge_bases.activate_onnx_service(
|
|
knowledge_bases.OnnxDownloadBody(model="BAAI/bge-small-zh-v1.5"),
|
|
request=_request(),
|
|
server=server,
|
|
_=object(),
|
|
)
|
|
|
|
assert saved == {"enabled": True, "model": "BAAI/bge-small-zh-v1.5"}
|
|
assert response["enabled"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rename_document_returns_row_payload(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
server = SimpleNamespace(services=_services())
|
|
renamed = _Document(id="doc-1", is_dir=True, path="docs/new", filename="new")
|
|
|
|
class _FakeService:
|
|
def rename_document(self, *_args: object, **_kwargs: object) -> _Document:
|
|
return renamed
|
|
|
|
monkeypatch.setattr(knowledge_bases, "_knowledge_service", lambda _server: _FakeService())
|
|
|
|
result = await knowledge_bases.rename_document(
|
|
kb_id="kb-1",
|
|
doc_id="doc-1",
|
|
body=knowledge_bases.RenameDocumentBody(new_name="new"),
|
|
request=_request(),
|
|
server=server,
|
|
user=SimpleNamespace(id=1, is_admin=False),
|
|
)
|
|
|
|
assert result["document_id"] == "doc-1"
|
|
assert result["path"] == "docs/new"
|
|
assert result["filename"] == "new"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rename_document_maps_not_found(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
def fail(*_args: object, **_kwargs: object) -> None:
|
|
raise LookupError("knowledge document not found")
|
|
|
|
monkeypatch.setattr(knowledge_bases, "_knowledge_service", fail)
|
|
|
|
with pytest.raises(OctopError) as raised:
|
|
await knowledge_bases.rename_document(
|
|
kb_id="kb-1",
|
|
doc_id="doc-1",
|
|
body=knowledge_bases.RenameDocumentBody(new_name="new"),
|
|
request=_request(),
|
|
server=SimpleNamespace(services=_services()),
|
|
user=object(),
|
|
)
|
|
|
|
assert raised.value.code == ErrorCode.KNOWLEDGE_NOT_FOUND
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rename_document_maps_name_taken(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
def fail(*_args: object, **_kwargs: object) -> None:
|
|
raise ValueError("a knowledge document with this name already exists")
|
|
|
|
monkeypatch.setattr(knowledge_bases, "_knowledge_service", fail)
|
|
|
|
with pytest.raises(OctopError) as raised:
|
|
await knowledge_bases.rename_document(
|
|
kb_id="kb-1",
|
|
doc_id="doc-1",
|
|
body=knowledge_bases.RenameDocumentBody(new_name="dup"),
|
|
request=_request(),
|
|
server=SimpleNamespace(services=_services()),
|
|
user=object(),
|
|
)
|
|
|
|
assert raised.value.code == ErrorCode.KNOWLEDGE_NAME_TAKEN
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rename_document_maps_invalid_name(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
def fail(*_args: object, **_kwargs: object) -> None:
|
|
raise ValueError("invalid knowledge document name")
|
|
|
|
monkeypatch.setattr(knowledge_bases, "_knowledge_service", fail)
|
|
|
|
with pytest.raises(OctopError) as raised:
|
|
await knowledge_bases.rename_document(
|
|
kb_id="kb-1",
|
|
doc_id="doc-1",
|
|
body=knowledge_bases.RenameDocumentBody(new_name="a/b"),
|
|
request=_request(),
|
|
server=SimpleNamespace(services=_services()),
|
|
user=object(),
|
|
)
|
|
|
|
assert raised.value.code == ErrorCode.KNOWLEDGE_NAME_INVALID
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_base_accepts_max_documents(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
received: dict[str, object] = {}
|
|
|
|
class _StubSvc:
|
|
def update_base(self, *_: object, **kw: object) -> object:
|
|
received.update(kw)
|
|
return _Base(max_documents=42)
|
|
|
|
monkeypatch.setattr(knowledge_bases, "_knowledge_service", lambda _s: _StubSvc())
|
|
|
|
response = await knowledge_bases.update_base(
|
|
kb_id="kb-1",
|
|
body=knowledge_bases.UpdateBaseBody(max_documents=42),
|
|
request=_request(),
|
|
server=SimpleNamespace(services=_services()),
|
|
user=SimpleNamespace(id=1, is_admin=False),
|
|
)
|
|
assert received["max_documents"] == 42
|
|
assert response["max_documents"] == 42
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_base_rejects_max_documents_out_of_range() -> None:
|
|
from pydantic import ValidationError
|
|
|
|
from octop.api.routers import knowledge_bases
|
|
|
|
with pytest.raises(ValidationError):
|
|
knowledge_bases.UpdateBaseBody(max_documents=-1)
|
|
with pytest.raises(ValidationError):
|
|
knowledge_bases.UpdateBaseBody(max_documents=10_001)
|
|
assert knowledge_bases.UpdateBaseBody(max_documents=0).max_documents == 0
|
|
assert knowledge_bases.UpdateBaseBody(max_documents=10_000).max_documents == 10_000
|
|
assert knowledge_bases.UpdateBaseBody().max_documents is None
|
|
|
|
|
|
def test_map_knowledge_error_unclassified_returns_internal_error() -> None:
|
|
from octop.api.routers.knowledge_bases import _map_knowledge_error
|
|
|
|
err = _map_knowledge_error(RuntimeError("column max_documents does not exist"), locale="zh")
|
|
assert err.code == ErrorCode.INTERNAL_ERROR
|
|
assert err.status == 500
|
|
|
|
err_generic = _map_knowledge_error(Exception("unexpected db failure"), locale="zh")
|
|
assert err_generic.code == ErrorCode.INTERNAL_ERROR
|
|
assert err_generic.status == 500
|
|
|
|
|
|
def test_map_knowledge_error_prerequisites_distinguished() -> None:
|
|
from octop.api.routers.knowledge_bases import _map_knowledge_error
|
|
|
|
err = _map_knowledge_error(
|
|
RuntimeError("knowledge embedding prerequisites are not satisfied"), locale="zh"
|
|
)
|
|
assert err.code == ErrorCode.KNOWLEDGE_PREREQUISITES_FAILED
|
|
assert err.status == 409
|
|
|
|
err_model = _map_knowledge_error(
|
|
ValueError("enabling knowledge bases requires an embedding model"), locale="zh"
|
|
)
|
|
assert err_model.code == ErrorCode.KNOWLEDGE_PREREQUISITES_FAILED
|
|
assert err_model.status == 409
|
|
|
|
|
|
class TestRestartRequiredMapping:
|
|
"""#1340: an install that only takes effect after a restart must say so."""
|
|
|
|
RESTART_HINT = (
|
|
"Local embedding components were installed but could not be loaded. "
|
|
"Restart the server and try again."
|
|
)
|
|
|
|
def test_install_that_needs_a_restart_says_so(self) -> None:
|
|
from octop.api.routers.knowledge_bases import _map_knowledge_error
|
|
from octop.infra.errors import ErrorCode
|
|
|
|
err = _map_knowledge_error(RuntimeError(self.RESTART_HINT), locale="zh")
|
|
|
|
assert err.code is ErrorCode.KNOWLEDGE_PREREQUISITES_RESTART_REQUIRED
|
|
assert "重启" in err.localized_message("zh")
|
|
assert "restart" in err.localized_message("en").lower()
|
|
|
|
def test_other_prerequisite_failures_keep_the_generic_error(self) -> None:
|
|
from octop.api.routers.knowledge_bases import _map_knowledge_error
|
|
from octop.infra.errors import ErrorCode
|
|
|
|
err = _map_knowledge_error(RuntimeError("embedding provider is not ready"), locale="zh")
|
|
|
|
assert err.code is ErrorCode.KNOWLEDGE_PREREQUISITES_FAILED
|