javadandCursor
b93ecf1691
Keep source, config, and deploy files. Leave runtime data, uploads, and logs out of git. Co-authored-by: Cursor <cursoragent@cursor.com>
99 lines
3.7 KiB
Python
99 lines
3.7 KiB
Python
import pytest
|
|
|
|
from kamidso.contracts.user import CurrentUser
|
|
from kamidso.resources.files import FileService
|
|
from kamidso.resources.knowledge import KnowledgeService, _parse_context, aggregate_space_status
|
|
from kamidso.repositories.store import RepositoryStore
|
|
from kamidso.settings.flags import load_kamidso_settings
|
|
from kamidso.storage.layout import DataLayout
|
|
from kamidso.storage.local import LocalStorage
|
|
|
|
|
|
def _user() -> CurrentUser:
|
|
return CurrentUser(
|
|
id="usr_test",
|
|
email="test@kamidso.test",
|
|
name="Test",
|
|
role="developer",
|
|
workspace_id="ws_test",
|
|
)
|
|
|
|
|
|
async def _service(tmp_path):
|
|
settings = load_kamidso_settings()
|
|
layout = DataLayout.from_base_path(tmp_path / "data")
|
|
layout.ensure()
|
|
store = RepositoryStore()
|
|
storage = LocalStorage(str(layout.root))
|
|
files = FileService(store, storage, settings)
|
|
knowledge = KnowledgeService(store, storage, layout)
|
|
return knowledge, files, store, layout
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_knowledge_create_copies_files_into_data_tree(tmp_path):
|
|
knowledge, files, store, layout = await _service(tmp_path)
|
|
user = _user()
|
|
uploaded = await files.upload(user, "notes.txt", b"hello knowledge", mime_type="text/plain")
|
|
source = await knowledge.create(
|
|
user, name="reports", document_resource_ids=[uploaded["_id"]]
|
|
)
|
|
assert source["_id"].startswith("kb_")
|
|
copied = layout.knowledge / source["_id"] / "notes.txt"
|
|
assert copied.is_file()
|
|
assert copied.read_bytes() == b"hello knowledge"
|
|
listed = await knowledge.list(user)
|
|
assert listed[0]["_id"] == source["_id"]
|
|
docs = list(store.knowledge_documents.values())
|
|
assert docs
|
|
assert docs[0]["doc_name"] == "notes.txt"
|
|
public = KnowledgeService.to_public(source)
|
|
assert public["id"] == source["_id"]
|
|
assert public["name"] == "reports"
|
|
assert public["document_count"] == 1
|
|
assert public["status"] in {"processing", "ready", "partial", "failed"}
|
|
assert "description" in public
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_knowledge_text_source_retrieve_and_delete(tmp_path):
|
|
knowledge, _files, store, layout = await _service(tmp_path)
|
|
user = _user()
|
|
source = await knowledge.create(
|
|
user,
|
|
name="texts",
|
|
description="catalog",
|
|
sources=[{"type": "text", "text": "tehran prices increased", "title": "price report"}],
|
|
)
|
|
docs = await knowledge.list_documents(user, source["_id"])
|
|
assert len(docs) == 1
|
|
await store.create_knowledge_chunk(
|
|
{
|
|
"document_id": docs[0]["_id"],
|
|
"content": "tehran prices increased",
|
|
"doc_name": "price report",
|
|
}
|
|
)
|
|
hits = await knowledge.retrieve(user, source["_id"], query="tehran")
|
|
assert hits
|
|
assert "tehran" in hits[0]["content"]
|
|
chunks = await knowledge.list_chunks(user, source["_id"], docs[0]["_id"])
|
|
assert chunks["total"] == 1
|
|
await knowledge.delete(user, source["_id"])
|
|
assert await store.get_knowledge_source(user.workspace_id, source["_id"]) is None
|
|
assert not list(store.knowledge_documents.values())
|
|
assert not (layout.knowledge / source["_id"]).exists()
|
|
|
|
|
|
def test_parse_context_accepts_dict_and_json():
|
|
assert _parse_context({"embedding": {"chunk_size": 512}})["embedding"]["chunk_size"] == 512
|
|
assert _parse_context('{"embedding": {"chunk_size": 10}}')["embedding"]["chunk_size"] == 10
|
|
|
|
|
|
def test_aggregate_space_status():
|
|
assert aggregate_space_status([]) == "ready"
|
|
assert aggregate_space_status([{"status": "FINISHED"}]) == "ready"
|
|
assert aggregate_space_status([{"status": "TODO"}]) == "processing"
|
|
assert aggregate_space_status([{"status": "FAILED"}]) == "failed"
|
|
assert aggregate_space_status([{"status": "FINISHED"}, {"status": "FAILED"}]) == "partial"
|