Knowledge, notes, memory and skills, and a harness to make them used
Four places a model can reach for, differing in who writes a record and how it gets in front of the model. **Knowledge** is uploaded by a person and searched by the model. It goes through `services/files.py:prepare` — the same pipeline as a chat attachment — so the same PDF produces the same text whichever way it arrived, and `Document` carries the same content columns as `Attachment` for the same reason. **Notes** are written by the model and edited by you. Too long to inject, so they are searched. **Memory** is short facts, and every one of them goes into every request. That single decision is where the rest of its design comes from: records are capped short, the block has a budget, there is no search tool because the model is already looking at them, and they are not shareable — a record about a person is not content to hand round. **Skills** are saved procedures. Only the name and description are injected; the body is fetched when the model decides one applies, which is what makes a hundred skills affordable. A model may write and revise its own — the safety story is not a gate but a record: every revision is kept, attributed and revertible. A model that has just read a hostile page can save a skill that outlives the conversation, and the honest mitigation is that it is visible and undoable rather than that it was prevented. **The harness** is why any of it gets used. A model handed a tools array ignores it and answers from recall, because nothing in the request suggests otherwise. `services/harness.py` assembles a preamble from what this chat actually has: when to reach for each tool, the memories, the skill index. This is an exception to "system prompts are precedence, not concatenation", and a deliberate one. That rule governs the three *authored* layers and is untouched — exactly one still wins. The harness is a different axis: it describes the machinery rather than the behaviour, nobody authored it, and there is nothing for it to disagree with. It is prepended to whichever authored prompt won, in one system message, since several endpoints reject a second. Supporting changes: - **Sharing**, in one helper. `visible_to()` is the only definition of who can see a library item and every listing and tool goes through it. Sharing grants *reading*; two people editing one note with no history and no merge is worse than copying it. **Administrators do not bypass this** — they bypass permissions elsewhere because an admin can grant themselves those anyway, but reading somebody's private notes is a different act. - **FTS5**, created by `db/migrations.py:ensure_fts` with the triggers an external-content index needs. Idempotent, like the column sync beside it. Terms are ANDed and then ORed: the caller is usually a model writing a whole question, and requiring every word loses the match on one absent term. - **The attach button is a menu** — file, image, a web page, or a document from the library. Attaching a document copies it, because history must not change when a document is edited later. - **A URL fetcher with an SSRF guard.** This server can reach the router, the other services on the box and LLeMbas itself, and the address can come from a model. Private ranges are refused *after resolution* and redirects are followed by hand so every hop is checked. An admin can open it deliberately. - **Model capabilities split** into protocol support and a toggle per built-in tool. Rows predating the split have no `tool_*` keys, and absent counts as on when `tools` is on — otherwise an upgrade silently takes web search away from every model already configured for it. Also fixes the test fixture, which built the schema with `create_all` and so ran against a database without the FTS tables production has; it now runs `sync_schema`, the same path startup takes. 430 tests, ruff clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jaroslav Beneš
@@ -46,7 +46,14 @@ def fresh_database(tmp_path: Path) -> Iterator[None]:
|
||||
|
||||
import lembas.db.models # noqa: F401 (registers the tables)
|
||||
|
||||
# sync_schema rather than create_all: it is what startup runs, and it also
|
||||
# builds the full-text indexes, which are not SQLAlchemy models and so are
|
||||
# invisible to create_all. Tests were otherwise running against a schema
|
||||
# production does not have.
|
||||
from lembas.db.migrations import sync_schema
|
||||
|
||||
Base.metadata.create_all(bind=get_engine())
|
||||
sync_schema(get_engine())
|
||||
yield
|
||||
reset_engine()
|
||||
|
||||
|
||||
@@ -0,0 +1,194 @@
|
||||
"""Fetching a web page, and refusing to fetch the wrong ones.
|
||||
|
||||
The refusals are the important half. This runs on a server that can reach the
|
||||
router, the other services on the box, and LLeMbas itself — and the URL can come
|
||||
from a model, which can be talked into things by a page it just read.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from lembas.services.fetch import FetchError, check_url, fetch, html_to_text
|
||||
|
||||
|
||||
# --- What is refused ---------------------------------------------------------
|
||||
@pytest.mark.parametrize(
|
||||
"url",
|
||||
[
|
||||
"http://127.0.0.1:8080/admin", # LLeMbas itself
|
||||
"http://localhost/", # the same, by name
|
||||
"http://10.0.0.1/", # the network the server is on
|
||||
"http://192.168.1.1/", # a router
|
||||
"http://172.16.5.4/",
|
||||
"http://169.254.169.254/latest/meta-data/", # cloud metadata, i.e. credentials
|
||||
"http://[::1]/",
|
||||
"http://0.0.0.0/",
|
||||
],
|
||||
)
|
||||
def test_private_addresses_are_refused(url):
|
||||
with pytest.raises(FetchError, match="private or local"):
|
||||
check_url(url)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"url", ["file:///etc/passwd", "ftp://host/x", "gopher://host/", "javascript:alert(1)"]
|
||||
)
|
||||
def test_only_http_and_https(url):
|
||||
with pytest.raises(FetchError, match="http"):
|
||||
check_url(url)
|
||||
|
||||
|
||||
def test_a_url_with_no_host_is_refused():
|
||||
with pytest.raises(FetchError):
|
||||
check_url("http:///nothing")
|
||||
|
||||
|
||||
def test_the_check_is_on_the_resolved_address(monkeypatch):
|
||||
"""A hostname pointing at 127.0.0.1 is the obvious way past a check that
|
||||
only reads the text of the URL."""
|
||||
import socket
|
||||
|
||||
monkeypatch.setattr(
|
||||
socket, "getaddrinfo", lambda *a, **k: [(2, 1, 6, "", ("127.0.0.1", 80))]
|
||||
)
|
||||
with pytest.raises(FetchError, match="private or local"):
|
||||
check_url("http://sneaky.example.com/")
|
||||
|
||||
|
||||
def test_one_private_address_among_several_is_still_refused(monkeypatch):
|
||||
"""A name resolving to one public and one private address must not be
|
||||
usable to reach the private one."""
|
||||
import socket
|
||||
|
||||
monkeypatch.setattr(
|
||||
socket,
|
||||
"getaddrinfo",
|
||||
lambda *a, **k: [(2, 1, 6, "", ("93.184.216.34", 80)), (2, 1, 6, "", ("10.0.0.1", 80))],
|
||||
)
|
||||
with pytest.raises(FetchError, match="private or local"):
|
||||
check_url("http://mixed.example.com/")
|
||||
|
||||
|
||||
def test_an_administrator_can_open_it_deliberately():
|
||||
assert check_url("http://127.0.0.1:8080/", allow_private=True)
|
||||
|
||||
|
||||
def test_a_public_address_passes(monkeypatch):
|
||||
import socket
|
||||
|
||||
monkeypatch.setattr(
|
||||
socket, "getaddrinfo", lambda *a, **k: [(2, 1, 6, "", ("93.184.216.34", 80))]
|
||||
)
|
||||
assert check_url("https://example.com/x") == "https://example.com/x"
|
||||
|
||||
|
||||
# --- Redirects ---------------------------------------------------------------
|
||||
async def test_a_redirect_to_a_private_address_is_refused(mock_http, monkeypatch):
|
||||
"""httpx's own following would validate the first address and then happily
|
||||
land on localhost, which is why redirects are followed by hand."""
|
||||
import socket
|
||||
|
||||
monkeypatch.setattr(
|
||||
socket,
|
||||
"getaddrinfo",
|
||||
lambda host, *a, **k: [(2, 1, 6, "", ("93.184.216.34", 80))]
|
||||
if host == "example.com"
|
||||
else [(2, 1, 6, "", ("127.0.0.1", 80))],
|
||||
)
|
||||
mock_http(
|
||||
lambda _r: httpx.Response(302, headers={"location": "http://127.0.0.1:8080/admin"})
|
||||
)
|
||||
with pytest.raises(FetchError, match="private or local"):
|
||||
await fetch("https://example.com/")
|
||||
|
||||
|
||||
async def test_endless_redirects_end(mock_http, monkeypatch):
|
||||
import socket
|
||||
|
||||
monkeypatch.setattr(
|
||||
socket, "getaddrinfo", lambda *a, **k: [(2, 1, 6, "", ("93.184.216.34", 80))]
|
||||
)
|
||||
mock_http(lambda r: httpx.Response(302, headers={"location": str(r.url)}))
|
||||
with pytest.raises(FetchError, match="redirected too many"):
|
||||
await fetch("https://example.com/")
|
||||
|
||||
|
||||
# --- Reducing a page ---------------------------------------------------------
|
||||
def test_script_and_style_are_dropped():
|
||||
_, text = html_to_text(
|
||||
"<html><body><script>alert(1)</script><style>p{}</style><p>Real text.</p></body></html>"
|
||||
)
|
||||
assert text == "Real text."
|
||||
|
||||
|
||||
def test_the_title_is_taken_and_not_repeated_in_the_body():
|
||||
title, text = html_to_text(
|
||||
"<html><head><title> A Page </title></head><body><p>Body.</p></body></html>"
|
||||
)
|
||||
assert title == "A Page"
|
||||
assert text == "Body."
|
||||
|
||||
|
||||
def test_block_tags_become_line_breaks():
|
||||
"""Without this the whole page arrives as one paragraph."""
|
||||
_, text = html_to_text("<p>One</p><p>Two</p><li>Three</li>")
|
||||
assert text.splitlines() == ["One", "Two", "Three"]
|
||||
|
||||
|
||||
def test_entities_are_unescaped():
|
||||
_, text = html_to_text("<p>Salt & pepper, 5 > 3</p>")
|
||||
assert text == "Salt & pepper, 5 > 3"
|
||||
|
||||
|
||||
# --- Fetching ----------------------------------------------------------------
|
||||
async def test_a_page_is_reduced_to_text(mock_http, monkeypatch):
|
||||
import socket
|
||||
|
||||
monkeypatch.setattr(
|
||||
socket, "getaddrinfo", lambda *a, **k: [(2, 1, 6, "", ("93.184.216.34", 80))]
|
||||
)
|
||||
mock_http(
|
||||
lambda _r: httpx.Response(
|
||||
200,
|
||||
headers={"content-type": "text/html"},
|
||||
text=(
|
||||
"<html><head><title>Mallorn</title></head>"
|
||||
"<body><p>A golden tree.</p></body></html>"
|
||||
),
|
||||
)
|
||||
)
|
||||
page = await fetch("https://example.com/mallorn")
|
||||
assert page.title == "Mallorn"
|
||||
assert page.text == "A golden tree."
|
||||
|
||||
|
||||
async def test_a_binary_response_is_refused_with_advice(mock_http, monkeypatch):
|
||||
import socket
|
||||
|
||||
monkeypatch.setattr(
|
||||
socket, "getaddrinfo", lambda *a, **k: [(2, 1, 6, "", ("93.184.216.34", 80))]
|
||||
)
|
||||
mock_http(
|
||||
lambda _r: httpx.Response(200, headers={"content-type": "image/png"}, content=b"\x89PNG")
|
||||
)
|
||||
with pytest.raises(FetchError, match="Attach it as a file"):
|
||||
await fetch("https://example.com/x.png")
|
||||
|
||||
|
||||
async def test_a_page_with_no_readable_text_says_so(mock_http, monkeypatch):
|
||||
import socket
|
||||
|
||||
monkeypatch.setattr(
|
||||
socket, "getaddrinfo", lambda *a, **k: [(2, 1, 6, "", ("93.184.216.34", 80))]
|
||||
)
|
||||
mock_http(
|
||||
lambda _r: httpx.Response(
|
||||
200,
|
||||
headers={"content-type": "text/html"},
|
||||
text="<html><body><div id=app></div></body></html>",
|
||||
)
|
||||
)
|
||||
with pytest.raises(FetchError, match="JavaScript"):
|
||||
await fetch("https://example.com/")
|
||||
@@ -0,0 +1,173 @@
|
||||
"""The operational preamble, and how it sits beside the authored prompt."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from lembas.db.models import Chat, Connection, Model, User
|
||||
from lembas.security.passwords import hash_password
|
||||
from lembas.services import chat as chat_service
|
||||
from lembas.services import harness, settings_store
|
||||
from lembas.services import tools as tools_service
|
||||
from lembas.services.library import memories as memories_service
|
||||
from lembas.services.library import skills as skills_service
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def owner(db):
|
||||
user = User(name="Frodo", email="f@shire.test", password_hash=hash_password("x"))
|
||||
db.add(user)
|
||||
db.commit()
|
||||
return user
|
||||
|
||||
|
||||
def _tools(*names):
|
||||
return [tools_service.REGISTRY[name].schema for name in names]
|
||||
|
||||
|
||||
# --- Composition -------------------------------------------------------------
|
||||
def test_no_tools_means_no_harness(db, owner):
|
||||
"""An empty harness is worse than none: tokens that say only that there is
|
||||
nothing to say."""
|
||||
assert harness.compose(db, owner, []) == ""
|
||||
assert harness.compose(db, owner, None) == ""
|
||||
|
||||
|
||||
def test_only_the_guidance_for_offered_tools_appears(db, owner):
|
||||
text = harness.compose(db, owner, _tools("web_search"))
|
||||
assert "Look things up" in text
|
||||
assert "You keep notes" not in text
|
||||
assert "Skills are procedures" not in text
|
||||
|
||||
|
||||
def test_the_memory_block_is_included_when_memory_is_offered(db, owner):
|
||||
memories_service.add(db, owner=owner, content="Prefers metric units.")
|
||||
text = harness.compose(db, owner, _tools("memory_add"))
|
||||
assert "What you know about this person" in text
|
||||
assert "Prefers metric units." in text
|
||||
|
||||
|
||||
def test_memories_are_absent_without_the_memory_tool(db, owner):
|
||||
"""A model not given the memory tool has no business being told them."""
|
||||
memories_service.add(db, owner=owner, content="Prefers metric units.")
|
||||
text = harness.compose(db, owner, _tools("web_search"))
|
||||
assert "Prefers metric units." not in text
|
||||
|
||||
|
||||
def test_the_skill_index_is_names_and_descriptions_only(db, owner):
|
||||
skills_service.create(
|
||||
db, owner=owner, name="weekly-report", description="When asked.", body="SECRET"
|
||||
)
|
||||
text = harness.compose(db, owner, _tools("skill_get"))
|
||||
assert "weekly-report: When asked." in text
|
||||
assert "SECRET" not in text
|
||||
|
||||
|
||||
def test_an_empty_store_contributes_no_heading(db, owner):
|
||||
"""And the guidance must not point at a heading that is not there: telling a
|
||||
model to consult an absent section is a good way to make it invent one."""
|
||||
text = harness.compose(db, owner, _tools("memory_add", "skill_get"))
|
||||
assert "### What you know about this person" not in text
|
||||
assert "### Skills available" not in text
|
||||
assert "was remembered earlier" not in text
|
||||
assert "You can remember durable facts" in text
|
||||
|
||||
|
||||
def test_the_harness_is_capped(db, owner, monkeypatch):
|
||||
monkeypatch.setattr(harness, "MAX_HARNESS_CHARS", 200)
|
||||
for index in range(50):
|
||||
skills_service.create(
|
||||
db, owner=owner, name=f"skill-{index}", description="x" * 200, body="y"
|
||||
)
|
||||
assert len(harness.compose(db, owner, _tools("skill_get"))) <= 202
|
||||
|
||||
|
||||
# --- Joining -----------------------------------------------------------------
|
||||
def test_the_authored_prompt_comes_last():
|
||||
"""It is closest to the conversation, and it is what the user actually
|
||||
wrote."""
|
||||
joined = harness.join("HARNESS", "AUTHORED")
|
||||
assert joined.index("HARNESS") < joined.index("AUTHORED")
|
||||
|
||||
|
||||
def test_either_half_alone_is_returned_unchanged():
|
||||
assert harness.join("", "AUTHORED") == "AUTHORED"
|
||||
assert harness.join("HARNESS", "") == "HARNESS"
|
||||
assert harness.join("", "") == ""
|
||||
|
||||
|
||||
# --- Through build_request ---------------------------------------------------
|
||||
def _chat(db, owner, *, capabilities, model_prompt="", chat_prompt=""):
|
||||
connection = Connection(name="c", base_url="http://h", api_key_encrypted="")
|
||||
db.add(connection)
|
||||
db.commit()
|
||||
db.add(
|
||||
Model(
|
||||
connection_id=connection.id,
|
||||
model_id="m",
|
||||
capabilities_json=capabilities,
|
||||
system_prompt=model_prompt,
|
||||
)
|
||||
)
|
||||
db.commit()
|
||||
chat = Chat(
|
||||
user_id=owner.id, model_id="m", connection_id=connection.id, system_prompt=chat_prompt
|
||||
)
|
||||
db.add(chat)
|
||||
db.commit()
|
||||
return chat
|
||||
|
||||
|
||||
def _system(body):
|
||||
first = body["messages"][0] if body["messages"] else {}
|
||||
return first.get("content", "") if first.get("role") == "system" else ""
|
||||
|
||||
|
||||
def test_a_request_without_tools_has_no_harness_and_no_tools_key(db, owner):
|
||||
chat = _chat(db, owner, capabilities={})
|
||||
body = chat_service.build_request(db, chat, tools=[], user=owner)
|
||||
assert "tools" not in body
|
||||
assert "How to work" not in _system(body)
|
||||
|
||||
|
||||
def test_the_harness_precedes_the_authored_prompt(db, owner):
|
||||
settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)
|
||||
chat = _chat(db, owner, capabilities={"tools": True}, chat_prompt="Speak as Gandalf.")
|
||||
offered = tools_service.enabled_tools(db, chat, owner)
|
||||
|
||||
system = _system(chat_service.build_request(db, chat, tools=offered, user=owner))
|
||||
assert system.index("How to work") < system.index("Speak as Gandalf.")
|
||||
|
||||
|
||||
def test_precedence_between_the_authored_layers_is_untouched(db, owner):
|
||||
"""The harness is a different axis. Exactly one authored layer still wins,
|
||||
and it is still the most specific one."""
|
||||
settings_store.update(db, {"system_prompt": "Instance."})
|
||||
settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)
|
||||
|
||||
chat = _chat(
|
||||
db, owner, capabilities={"tools": True}, model_prompt="Model.", chat_prompt="Chat."
|
||||
)
|
||||
offered = tools_service.enabled_tools(db, chat, owner)
|
||||
system = _system(chat_service.build_request(db, chat, tools=offered, user=owner))
|
||||
|
||||
assert "Chat." in system
|
||||
assert "Model." not in system
|
||||
assert "Instance." not in system
|
||||
# And the resolver on its own is unchanged.
|
||||
assert chat_service.effective_system_prompt(db, chat) == "Chat."
|
||||
|
||||
|
||||
def test_a_model_prompt_wins_when_the_chat_has_none(db, owner):
|
||||
settings_store.update(db, {"system_prompt": "Instance."})
|
||||
chat = _chat(db, owner, capabilities={}, model_prompt="Model.")
|
||||
system = _system(chat_service.build_request(db, chat, user=owner))
|
||||
assert system == "Model."
|
||||
|
||||
|
||||
def test_the_tools_array_rides_along(db, owner):
|
||||
settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)
|
||||
chat = _chat(db, owner, capabilities={"tools": True})
|
||||
offered = tools_service.enabled_tools(db, chat, owner)
|
||||
body = chat_service.build_request(db, chat, tools=offered, user=owner)
|
||||
assert body["tools"] == offered
|
||||
@@ -0,0 +1,291 @@
|
||||
"""The four stores: ingestion, search, memory limits, skill history."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import text
|
||||
|
||||
from lembas.db.migrations import ensure_fts
|
||||
from lembas.db.models import AUTHOR_MODEL, Attachment, Document, User
|
||||
from lembas.db.session import get_engine
|
||||
from lembas.security.passwords import hash_password
|
||||
from lembas.services import files as files_service
|
||||
from lembas.services.library import documents as documents_service
|
||||
from lembas.services.library import memories as memories_service
|
||||
from lembas.services.library import notes as notes_service
|
||||
from lembas.services.library import skills as skills_service
|
||||
from lembas.services.library.fts import fts_query
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def owner(db):
|
||||
user = User(name="Frodo", email="f@shire.test", password_hash=hash_password("x"))
|
||||
db.add(user)
|
||||
db.commit()
|
||||
return user
|
||||
|
||||
|
||||
# --- The index ---------------------------------------------------------------
|
||||
def test_ensure_fts_is_idempotent():
|
||||
"""It runs at every startup, like the column sync beside it."""
|
||||
assert ensure_fts(get_engine()) == []
|
||||
assert ensure_fts(get_engine()) == []
|
||||
|
||||
|
||||
def test_the_query_builder_survives_punctuation():
|
||||
"""FTS5 MATCH has operators and a quoting rule, so a bare quote or asterisk
|
||||
is a syntax error rather than a search that finds nothing."""
|
||||
assert fts_query('what "is" a mallorn?') == '"what" AND "is" AND "a" AND "mallorn"'
|
||||
assert fts_query("a AND b OR NEAR *") == '"a" AND "AND" AND "b" AND "OR" AND "NEAR"'
|
||||
assert fts_query(" ") == ""
|
||||
assert fts_query("!!!") == ""
|
||||
|
||||
|
||||
def test_editing_moves_a_note_in_the_index(db, owner):
|
||||
"""The triggers are what keep an external-content index correct."""
|
||||
note = notes_service.create(db, owner=owner, title="Tree", body="A golden tree.")
|
||||
assert notes_service.search(db, owner, "golden")
|
||||
|
||||
notes_service.update(db, note, body="A silver tree.")
|
||||
assert notes_service.search(db, owner, "golden") == []
|
||||
assert notes_service.search(db, owner, "silver")
|
||||
|
||||
|
||||
def test_a_deleted_note_leaves_the_index(db, owner):
|
||||
note = notes_service.create(db, owner=owner, title="Tree", body="A golden tree.")
|
||||
notes_service.delete(db, note)
|
||||
assert notes_service.search(db, owner, "golden") == []
|
||||
|
||||
|
||||
def test_the_description_is_searched_as_well_as_the_contents(db, owner):
|
||||
"""A line of description is how somebody makes a document findable when its
|
||||
own words do not include the term."""
|
||||
document = documents_service.store_upload(
|
||||
db, owner=owner, payload=b"Opaque contents.", filename="a.txt", title="A"
|
||||
)
|
||||
document.description = "Everything about invoicing."
|
||||
db.commit()
|
||||
assert documents_service.search(db, owner, "invoicing")
|
||||
|
||||
|
||||
def test_a_natural_language_question_still_finds_the_document(db, owner):
|
||||
"""The caller is usually a model, which asks "who built the west gate of
|
||||
Moria and what is its password" rather than "moria gate". Requiring every
|
||||
term would lose the match on one absent word."""
|
||||
documents_service.store_upload(
|
||||
db,
|
||||
owner=owner,
|
||||
payload=b"The west gate of Moria was built by Narvi. The password is mellon.",
|
||||
filename="gate.txt",
|
||||
title="Moria gate",
|
||||
)
|
||||
found = documents_service.search(
|
||||
db, owner, "who built the west gate of Moria and what is its password"
|
||||
)
|
||||
assert [d.title for d in found] == ["Moria gate"]
|
||||
|
||||
|
||||
def test_requiring_every_term_still_wins_when_it_can(db, owner):
|
||||
"""AND first, so a document matching all the words beats one matching some."""
|
||||
documents_service.store_upload(
|
||||
db, owner=owner, payload=b"Golden mallorn trees of Lothlorien.", filename="a.txt",
|
||||
title="Both",
|
||||
)
|
||||
documents_service.store_upload(
|
||||
db, owner=owner, payload=b"Golden light on the water.", filename="b.txt", title="One",
|
||||
)
|
||||
found = documents_service.search(db, owner, "golden mallorn")
|
||||
assert [d.title for d in found] == ["Both"]
|
||||
|
||||
|
||||
def test_a_broken_index_does_not_break_the_page(db, owner):
|
||||
"""Search degrading to "finds nothing" is bad; a 500 on the library page is
|
||||
worse, and a failed statement otherwise poisons the whole session."""
|
||||
notes_service.create(db, owner=owner, title="Tree", body="Golden.")
|
||||
db.execute(text("DROP TABLE notes_fts"))
|
||||
db.commit()
|
||||
|
||||
assert notes_service.search(db, owner, "golden") == []
|
||||
# The session must still be usable afterwards.
|
||||
assert notes_service.recent(db, owner)
|
||||
|
||||
|
||||
# --- Ingestion ---------------------------------------------------------------
|
||||
def test_a_document_and_an_attachment_extract_identically(db, owner):
|
||||
"""Both go through files.prepare, which is what guarantees the same file
|
||||
produces the same text whichever way it arrived."""
|
||||
payload = b"# Waybread\n\nOne bite is enough."
|
||||
|
||||
document = documents_service.store_upload(
|
||||
db, owner=owner, payload=payload, filename="lembas.md"
|
||||
)
|
||||
attachment = files_service.store(
|
||||
db, user_id=owner.id, chat_id=None, payload=payload, filename="lembas.md"
|
||||
)
|
||||
assert document.extracted_text == attachment.extracted_text
|
||||
assert document.kind == attachment.kind
|
||||
|
||||
|
||||
def test_a_document_keeps_its_own_file(db, owner):
|
||||
document = documents_service.store_upload(
|
||||
db, owner=owner, payload=b"hello", filename="a.txt"
|
||||
)
|
||||
path = documents_service.stored_path(document.stored_name)
|
||||
assert path is not None and path.read_bytes() == b"hello"
|
||||
|
||||
|
||||
def test_the_library_path_check_refuses_an_escape(db):
|
||||
"""Same resolve-and-check as chat attachments, against a different root."""
|
||||
assert documents_service.stored_path("../../etc/passwd") is None
|
||||
assert documents_service.stored_path("") is None
|
||||
assert documents_service.stored_path(".hidden") is None
|
||||
|
||||
|
||||
def test_attaching_a_document_copies_it(db, owner):
|
||||
"""History must not change under a conversation because a document was
|
||||
edited or deleted later."""
|
||||
document = documents_service.store_upload(
|
||||
db, owner=owner, payload=b"The ring is round.", filename="ring.txt"
|
||||
)
|
||||
attachment = files_service.copy_document(
|
||||
db, user_id=owner.id, chat_id=None, document=document
|
||||
)
|
||||
documents_service.delete(db, document)
|
||||
|
||||
db.refresh(attachment)
|
||||
assert attachment.extracted_text == "The ring is round."
|
||||
assert files_service.stored_path(attachment.stored_name) is not None
|
||||
assert db.get(Document, document.id) is None
|
||||
assert db.get(Attachment, attachment.id) is not None
|
||||
|
||||
|
||||
# --- Memory ------------------------------------------------------------------
|
||||
def test_a_memory_is_trimmed_rather_than_refused(db, owner):
|
||||
"""A tool that writes an essay is told so and can put the long version in a
|
||||
note; failing the write would just lose it."""
|
||||
memory = memories_service.add(db, owner=owner, content="x" * 5000)
|
||||
assert len(memory.content) == memories_service.MAX_MEMORY_CHARS
|
||||
|
||||
|
||||
def test_an_empty_memory_is_refused(db, owner):
|
||||
with pytest.raises(ValueError):
|
||||
memories_service.add(db, owner=owner, content=" ")
|
||||
|
||||
|
||||
def test_memory_whitespace_is_collapsed(db, owner):
|
||||
memory = memories_service.add(db, owner=owner, content=" two\n\nlines ")
|
||||
assert memory.content == "two lines"
|
||||
|
||||
|
||||
def test_the_injected_block_is_bounded(db, owner):
|
||||
"""Every memory costs tokens on every request, so the block has a ceiling.
|
||||
Nothing disappears -- the full list is still in settings."""
|
||||
for index in range(200):
|
||||
memories_service.add(db, owner=owner, content=f"Fact number {index}. " + "x" * 200)
|
||||
|
||||
block = memories_service.block(db, owner)
|
||||
assert len(block) < memories_service.MAX_TOTAL_CHARS + 200
|
||||
assert "more, see your settings" in block
|
||||
|
||||
|
||||
def test_the_oldest_memories_survive_truncation(db, owner):
|
||||
"""A fact that has lasted is likelier to be a standing preference than
|
||||
something said once this morning."""
|
||||
memories_service.add(db, owner=owner, content="The oldest fact.")
|
||||
for index in range(100):
|
||||
memories_service.add(db, owner=owner, content=f"Later fact {index}. " + "y" * 200)
|
||||
|
||||
assert "The oldest fact." in memories_service.block(db, owner)
|
||||
|
||||
|
||||
def test_there_is_a_hard_ceiling_on_records(db, owner, monkeypatch):
|
||||
monkeypatch.setattr(memories_service, "MAX_RECORDS", 3)
|
||||
for index in range(3):
|
||||
memories_service.add(db, owner=owner, content=f"Fact {index}")
|
||||
with pytest.raises(ValueError, match="note instead"):
|
||||
memories_service.add(db, owner=owner, content="One too many")
|
||||
|
||||
|
||||
def test_memory_block_of_nobody_is_empty(db):
|
||||
assert memories_service.block(db, None) == ""
|
||||
|
||||
|
||||
# --- Skills ------------------------------------------------------------------
|
||||
def test_a_skill_name_is_slugified(db, owner):
|
||||
skill = skills_service.create(
|
||||
db, owner=owner, name="Weekly Report!", description="When asked.", body="x"
|
||||
)
|
||||
assert skill.name == "weekly-report"
|
||||
|
||||
|
||||
def test_a_skill_needs_a_description(db, owner):
|
||||
"""It is the only thing the model sees until it opens the skill."""
|
||||
with pytest.raises(skills_service.SkillError, match="when to use it"):
|
||||
skills_service.create(db, owner=owner, name="thing", description=" ", body="x")
|
||||
|
||||
|
||||
def test_a_duplicate_name_is_refused(db, owner):
|
||||
skills_service.create(db, owner=owner, name="report", description="When.", body="x")
|
||||
with pytest.raises(skills_service.SkillError, match="already exists"):
|
||||
skills_service.create(db, owner=owner, name="report", description="When.", body="y")
|
||||
|
||||
|
||||
def test_every_edit_keeps_what_was_there(db, owner):
|
||||
"""The whole safety story for a model rewriting its own instructions."""
|
||||
skill = skills_service.create(
|
||||
db, owner=owner, name="report", description="When.", body="First."
|
||||
)
|
||||
skills_service.update(db, skill, body="Second.", author=AUTHOR_MODEL)
|
||||
db.refresh(skill)
|
||||
|
||||
assert skill.body == "Second."
|
||||
assert skill.author == AUTHOR_MODEL
|
||||
assert [r.body for r in skill.revisions] == ["First."]
|
||||
|
||||
|
||||
def test_an_edit_that_changes_nothing_writes_no_revision(db, owner):
|
||||
skill = skills_service.create(
|
||||
db, owner=owner, name="report", description="When.", body="First."
|
||||
)
|
||||
skills_service.update(db, skill, body="First.", description="When.")
|
||||
db.refresh(skill)
|
||||
assert skill.revisions == []
|
||||
|
||||
|
||||
def test_reverting_restores_and_is_itself_undoable(db, owner):
|
||||
skill = skills_service.create(
|
||||
db, owner=owner, name="report", description="When.", body="First."
|
||||
)
|
||||
skills_service.update(db, skill, body="Second.", author=AUTHOR_MODEL)
|
||||
db.refresh(skill)
|
||||
|
||||
skills_service.revert(db, skill, skill.revisions[-1], author="user")
|
||||
db.refresh(skill)
|
||||
assert skill.body == "First."
|
||||
# Going back is undoable too: the state before the revert was kept.
|
||||
assert "Second." in [r.body for r in skill.revisions]
|
||||
|
||||
|
||||
def test_the_index_lists_only_enabled_skills(db, owner):
|
||||
skills_service.create(db, owner=owner, name="on", description="Use me.", body="x")
|
||||
off = skills_service.create(db, owner=owner, name="off", description="Not me.", body="x")
|
||||
skills_service.update(db, off, enabled=False)
|
||||
|
||||
index = skills_service.index_block(db, owner)
|
||||
assert "on: Use me." in index
|
||||
assert "off" not in index
|
||||
|
||||
|
||||
def test_the_index_is_name_and_description_only(db, owner):
|
||||
"""Bodies are fetched with skill_get; injecting them is what makes a hundred
|
||||
skills unaffordable."""
|
||||
skills_service.create(
|
||||
db, owner=owner, name="report", description="When asked.", body="SECRET STEPS"
|
||||
)
|
||||
assert "SECRET STEPS" not in skills_service.index_block(db, owner)
|
||||
|
||||
|
||||
def test_a_skill_is_found_by_the_name_a_model_would_use(db, owner):
|
||||
skills_service.create(db, owner=owner, name="weekly-report", description="W.", body="x")
|
||||
assert skills_service.by_name(db, "Weekly Report", owner) is not None
|
||||
assert skills_service.by_name(db, "nope", owner) is None
|
||||
@@ -0,0 +1,231 @@
|
||||
"""Who can see a document, a note or a skill.
|
||||
|
||||
The most consequential tests in the library: everything else is a feature not
|
||||
working, this is somebody reading somebody else's material.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
|
||||
from lembas.db.models import (
|
||||
PRINCIPAL_GROUP,
|
||||
PRINCIPAL_USER,
|
||||
Document,
|
||||
Group,
|
||||
Note,
|
||||
Share,
|
||||
User,
|
||||
)
|
||||
from lembas.security.passwords import hash_password
|
||||
from lembas.services import sharing
|
||||
from lembas.services.library import documents as documents_service
|
||||
from lembas.services.library import notes as notes_service
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def people(db):
|
||||
"""Three accounts: an owner, a stranger, and an administrator."""
|
||||
made = {}
|
||||
for name, role in (("frodo", "user"), ("gollum", "user"), ("gandalf", "admin")):
|
||||
user = User(
|
||||
name=name, email=f"{name}@shire.test", password_hash=hash_password("x"), role=role
|
||||
)
|
||||
db.add(user)
|
||||
made[name] = user
|
||||
db.commit()
|
||||
return made
|
||||
|
||||
|
||||
def _note(db, owner, title="Secret"):
|
||||
return notes_service.create(db, owner=owner, title=title, body="The ring is in the drawer.")
|
||||
|
||||
|
||||
# --- The rule ----------------------------------------------------------------
|
||||
def test_the_owner_sees_their_own(db, people):
|
||||
note = _note(db, people["frodo"])
|
||||
assert sharing.can_read(db, note, people["frodo"])
|
||||
assert note in db.scalars(notes_service.visible(db, people["frodo"]))
|
||||
|
||||
|
||||
def test_a_stranger_sees_nothing(db, people):
|
||||
note = _note(db, people["frodo"])
|
||||
assert not sharing.can_read(db, note, people["gollum"])
|
||||
assert note not in db.scalars(notes_service.visible(db, people["gollum"]))
|
||||
|
||||
|
||||
def test_an_administrator_gets_no_free_pass(db, people):
|
||||
"""Admins bypass permissions elsewhere, deliberately -- an admin can grant
|
||||
themselves those in two clicks. This is different: nobody made this
|
||||
available to anyone, and administering a box is not being invited."""
|
||||
note = _note(db, people["frodo"])
|
||||
assert not sharing.can_read(db, note, people["gandalf"])
|
||||
assert note not in db.scalars(notes_service.visible(db, people["gandalf"]))
|
||||
|
||||
|
||||
def test_sharing_with_a_person_lets_them_read(db, people):
|
||||
note = _note(db, people["frodo"])
|
||||
sharing.set_grants(db, note, user_ids=[people["gollum"].id], group_ids=[])
|
||||
assert sharing.can_read(db, note, people["gollum"])
|
||||
assert note in db.scalars(notes_service.visible(db, people["gollum"]))
|
||||
|
||||
|
||||
def test_sharing_with_a_group_lets_its_members_read(db, people):
|
||||
group = Group(name="Fellowship")
|
||||
group.users.append(people["gollum"])
|
||||
db.add(group)
|
||||
db.commit()
|
||||
|
||||
note = _note(db, people["frodo"])
|
||||
sharing.set_grants(db, note, user_ids=[], group_ids=[group.id])
|
||||
assert sharing.can_read(db, note, people["gollum"])
|
||||
|
||||
|
||||
def test_leaving_a_group_takes_the_access_with_it(db, people):
|
||||
group = Group(name="Fellowship")
|
||||
group.users.append(people["gollum"])
|
||||
db.add(group)
|
||||
db.commit()
|
||||
note = _note(db, people["frodo"])
|
||||
sharing.set_grants(db, note, user_ids=[], group_ids=[group.id])
|
||||
|
||||
group.users.remove(people["gollum"])
|
||||
db.commit()
|
||||
db.refresh(people["gollum"])
|
||||
assert not sharing.can_read(db, note, people["gollum"])
|
||||
|
||||
|
||||
def test_signed_out_sees_nothing(db, people):
|
||||
_note(db, people["frodo"])
|
||||
assert list(db.scalars(notes_service.visible(db, None))) == []
|
||||
|
||||
|
||||
# --- Sharing grants reading only ---------------------------------------------
|
||||
def test_a_share_does_not_grant_writing(db, people):
|
||||
"""Two people editing one note with no history and no merge is worse than
|
||||
the inconvenience of copying it."""
|
||||
note = _note(db, people["frodo"])
|
||||
sharing.set_grants(db, note, user_ids=[people["gollum"].id], group_ids=[])
|
||||
assert sharing.can_read(db, note, people["gollum"])
|
||||
assert not sharing.can_write(note, people["gollum"])
|
||||
assert sharing.can_write(note, people["frodo"])
|
||||
|
||||
|
||||
# --- Managing grants ---------------------------------------------------------
|
||||
def test_set_grants_replaces_rather_than_adds(db, people):
|
||||
note = _note(db, people["frodo"])
|
||||
sharing.set_grants(db, note, user_ids=[people["gollum"].id], group_ids=[])
|
||||
sharing.set_grants(db, note, user_ids=[people["gandalf"].id], group_ids=[])
|
||||
|
||||
assert not sharing.can_read(db, note, people["gollum"])
|
||||
assert sharing.can_read(db, note, people["gandalf"])
|
||||
|
||||
|
||||
def test_sharing_with_yourself_is_ignored(db, people):
|
||||
note = _note(db, people["frodo"])
|
||||
sharing.set_grants(db, note, user_ids=[people["frodo"].id], group_ids=[])
|
||||
assert sharing.grants_for(db, note) == []
|
||||
|
||||
|
||||
def test_deleting_a_note_drops_its_shares(db, people):
|
||||
"""Shares carry no foreign key to their resource -- one column pointing at
|
||||
three tables cannot have one -- so nothing cascades."""
|
||||
note = _note(db, people["frodo"])
|
||||
sharing.set_grants(db, note, user_ids=[people["gollum"].id], group_ids=[])
|
||||
notes_service.delete(db, note)
|
||||
assert db.scalar(select(Share).where(Share.resource_id == note.id)) is None
|
||||
|
||||
|
||||
def test_forgetting_a_principal_drops_their_shares(db, people):
|
||||
"""A stale row would grant access to whoever next received that id."""
|
||||
note = _note(db, people["frodo"])
|
||||
sharing.set_grants(db, note, user_ids=[people["gollum"].id], group_ids=[])
|
||||
assert sharing.forget_principal(db, PRINCIPAL_USER, people["gollum"].id) == 1
|
||||
assert sharing.grants_for(db, note) == []
|
||||
|
||||
|
||||
def test_two_kinds_of_resource_do_not_collide(db, people):
|
||||
"""One shares table across three resource types, so the type must be part
|
||||
of the match -- otherwise a note and a document sharing an id would share
|
||||
each other's access."""
|
||||
note = _note(db, people["frodo"])
|
||||
document = documents_service.store_upload(
|
||||
db, owner=people["frodo"], payload=b"hello", filename="a.txt"
|
||||
)
|
||||
sharing.set_grants(db, note, user_ids=[people["gollum"].id], group_ids=[])
|
||||
|
||||
assert sharing.can_read(db, note, people["gollum"])
|
||||
assert not sharing.can_read(db, document, people["gollum"])
|
||||
|
||||
|
||||
def test_resource_type_refuses_something_unshareable(db, people):
|
||||
"""Memory is deliberately not shareable: a record about a person is not
|
||||
content to hand round."""
|
||||
from lembas.db.models import Memory
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
sharing.resource_type(Memory)
|
||||
|
||||
|
||||
# --- Through the search path -------------------------------------------------
|
||||
def test_search_does_not_leak_across_owners(db, people):
|
||||
"""The index is searched first and the visibility filter applied to what it
|
||||
returned. Getting that order wrong leaks a hit even without the contents."""
|
||||
notes_service.create(
|
||||
db, owner=people["frodo"], title="Mallorn", body="A golden tree of Lothlorien."
|
||||
)
|
||||
assert notes_service.search(db, people["frodo"], "golden")
|
||||
assert notes_service.search(db, people["gollum"], "golden") == []
|
||||
assert notes_service.search(db, people["gandalf"], "golden") == []
|
||||
|
||||
|
||||
def test_a_shared_document_is_findable_by_the_person_it_was_shared_with(db, people):
|
||||
document = documents_service.store_upload(
|
||||
db,
|
||||
owner=people["frodo"],
|
||||
payload=b"The mallorn is a golden tree.",
|
||||
filename="tree.txt",
|
||||
)
|
||||
assert documents_service.search(db, people["gollum"], "mallorn") == []
|
||||
|
||||
sharing.set_grants(db, document, user_ids=[people["gollum"].id], group_ids=[])
|
||||
found = documents_service.search(db, people["gollum"], "mallorn")
|
||||
assert [d.id for d in found] == [document.id]
|
||||
|
||||
|
||||
def test_the_shares_table_records_what_was_asked_for(db, people):
|
||||
group = Group(name="Fellowship")
|
||||
db.add(group)
|
||||
db.commit()
|
||||
note = _note(db, people["frodo"])
|
||||
sharing.set_grants(
|
||||
db, note, user_ids=[people["gollum"].id], group_ids=[group.id]
|
||||
)
|
||||
kinds = {(s.principal_type, s.principal_id) for s in sharing.grants_for(db, note)}
|
||||
assert kinds == {
|
||||
(PRINCIPAL_USER, people["gollum"].id),
|
||||
(PRINCIPAL_GROUP, group.id),
|
||||
}
|
||||
|
||||
|
||||
def test_visibility_is_a_query_filter_not_a_python_loop(db, people):
|
||||
"""visible_to returns a condition so callers can page and order on the
|
||||
database side; a Python filter would break pagination silently."""
|
||||
for index in range(3):
|
||||
_note(db, people["frodo"], title=f"Note {index}")
|
||||
_note(db, people["gollum"], title="Theirs")
|
||||
|
||||
rows = db.scalars(
|
||||
notes_service.visible(db, people["frodo"]).order_by(Note.title).limit(2)
|
||||
)
|
||||
assert [n.title for n in rows] == ["Note 0", "Note 1"]
|
||||
|
||||
|
||||
def test_documents_and_notes_use_the_same_rule(db, people):
|
||||
document = documents_service.store_upload(
|
||||
db, owner=people["frodo"], payload=b"x", filename="a.txt"
|
||||
)
|
||||
assert document in db.scalars(documents_service.visible(db, people["frodo"]))
|
||||
assert document not in db.scalars(documents_service.visible(db, people["gollum"]))
|
||||
assert list(db.scalars(select(Document).where(sharing.visible_to(Document, None)))) == []
|
||||
+55
-18
@@ -104,12 +104,19 @@ def _user(db, user_id):
|
||||
return db.get(User, user_id)
|
||||
|
||||
|
||||
def test_nothing_is_offered_when_search_is_off(db, user_id):
|
||||
def _names(offered):
|
||||
return {tool["function"]["name"] for tool in offered}
|
||||
|
||||
|
||||
def test_web_search_is_absent_when_search_is_off(db, user_id):
|
||||
"""The library tools do not depend on a search provider, so they stay."""
|
||||
chat = _chat_with(db, user_id, capabilities={"tools": True})
|
||||
assert tools_service.enabled_tools(db, chat, _user(db, user_id)) == []
|
||||
offered = _names(tools_service.enabled_tools(db, chat, _user(db, user_id)))
|
||||
assert "web_search" not in offered
|
||||
assert "notes_search" in offered
|
||||
|
||||
|
||||
def test_nothing_is_offered_to_a_model_without_the_tools_capability(db, user_id):
|
||||
def test_nothing_at_all_without_the_tools_capability(db, user_id):
|
||||
"""Sending a tools array to an endpoint that does not implement tool calling
|
||||
fails the entire request, exactly as image parts do without vision."""
|
||||
settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)
|
||||
@@ -120,13 +127,29 @@ def test_nothing_is_offered_to_a_model_without_the_tools_capability(db, user_id)
|
||||
def test_web_search_is_offered_when_everything_lines_up(db, user_id):
|
||||
settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)
|
||||
chat = _chat_with(db, user_id, capabilities={"tools": True})
|
||||
|
||||
offered = tools_service.enabled_tools(db, chat, _user(db, user_id))
|
||||
assert len(offered) == 1
|
||||
assert offered[0]["function"]["name"] == "web_search"
|
||||
assert "web_search" in _names(tools_service.enabled_tools(db, chat, _user(db, user_id)))
|
||||
|
||||
|
||||
def test_nothing_is_offered_when_the_provider_cannot_run(db, user_id, monkeypatch):
|
||||
def test_a_model_predating_the_split_keeps_its_tools(db, user_id):
|
||||
"""Rows configured before the per-tool flags existed have no tool_* keys.
|
||||
Reading absent as off would silently take web search away from every model
|
||||
already set up for it."""
|
||||
settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)
|
||||
chat = _chat_with(db, user_id, capabilities={"tools": True})
|
||||
assert "web_search" in _names(tools_service.enabled_tools(db, chat, _user(db, user_id)))
|
||||
|
||||
|
||||
def test_a_family_turned_off_for_the_model_is_withheld(db, user_id):
|
||||
settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)
|
||||
chat = _chat_with(
|
||||
db, user_id, capabilities={"tools": True, "tool_notes": False}
|
||||
)
|
||||
offered = _names(tools_service.enabled_tools(db, chat, _user(db, user_id)))
|
||||
assert "notes_search" not in offered
|
||||
assert "web_search" in offered, "turning one family off must not affect another"
|
||||
|
||||
|
||||
def test_web_search_is_withheld_when_the_provider_cannot_run(db, user_id, monkeypatch):
|
||||
"""Offering a tool that will fail on every call is worse than not offering
|
||||
it at all."""
|
||||
monkeypatch.setattr(
|
||||
@@ -134,7 +157,13 @@ def test_nothing_is_offered_when_the_provider_cannot_run(db, user_id, monkeypatc
|
||||
)
|
||||
settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)
|
||||
chat = _chat_with(db, user_id, capabilities={"tools": True})
|
||||
assert tools_service.enabled_tools(db, chat, _user(db, user_id)) == []
|
||||
assert "web_search" not in _names(
|
||||
tools_service.enabled_tools(db, chat, _user(db, user_id))
|
||||
)
|
||||
|
||||
|
||||
def _context(**kwargs):
|
||||
return tools_service.ToolContext(owner_id="someone", **kwargs)
|
||||
|
||||
|
||||
# --- Running one -------------------------------------------------------------
|
||||
@@ -144,7 +173,7 @@ async def test_running_web_search_formats_results_for_the_model(monkeypatch):
|
||||
|
||||
monkeypatch.setattr("lembas.services.search.run", fake_run)
|
||||
|
||||
outcome = await tools_service.run_tool({}, "web_search", '{"query": "mallorn"}')
|
||||
outcome = await tools_service.run_tool(_context(), "web_search", '{"query": "mallorn"}')
|
||||
assert "A title" in outcome.content
|
||||
assert "https://a.test" in outcome.content
|
||||
assert outcome.event["status"] == "ok"
|
||||
@@ -161,7 +190,7 @@ async def test_malformed_argument_json_is_treated_as_the_query(monkeypatch):
|
||||
return []
|
||||
|
||||
monkeypatch.setattr("lembas.services.search.run", fake_run)
|
||||
await tools_service.run_tool({}, "web_search", "mallorn tree")
|
||||
await tools_service.run_tool(_context(), "web_search", "mallorn tree")
|
||||
assert seen["query"] == "mallorn tree"
|
||||
|
||||
|
||||
@@ -174,19 +203,19 @@ async def test_a_failed_search_hands_the_model_an_explanation(monkeypatch):
|
||||
|
||||
monkeypatch.setattr("lembas.services.search.run", fake_run)
|
||||
|
||||
outcome = await tools_service.run_tool({}, "web_search", '{"query": "x"}')
|
||||
outcome = await tools_service.run_tool(_context(), "web_search", '{"query": "x"}')
|
||||
assert "rate limiting" in outcome.content
|
||||
assert outcome.event["status"] == "error"
|
||||
|
||||
|
||||
async def test_an_unknown_tool_is_reported_rather_than_raised():
|
||||
outcome = await tools_service.run_tool({}, "launch_missiles", "{}")
|
||||
outcome = await tools_service.run_tool(_context(), "launch_missiles", "{}")
|
||||
assert "no tool called" in outcome.content
|
||||
assert outcome.event["status"] == "error"
|
||||
|
||||
|
||||
async def test_a_call_with_no_query_is_reported():
|
||||
outcome = await tools_service.run_tool({}, "web_search", '{"query": " "}')
|
||||
outcome = await tools_service.run_tool(_context(), "web_search", '{"query": " "}')
|
||||
assert outcome.event["status"] == "error"
|
||||
|
||||
|
||||
@@ -220,8 +249,16 @@ def test_the_tool_turn_carries_the_call_id():
|
||||
}
|
||||
|
||||
|
||||
def test_the_schema_is_valid_json():
|
||||
"""It is sent verbatim to the endpoint; a schema that will not serialise
|
||||
def test_every_schema_is_valid_json():
|
||||
"""They are sent verbatim to the endpoint; a schema that will not serialise
|
||||
fails every request rather than one."""
|
||||
json.dumps(tools_service.WEB_SEARCH_SCHEMA)
|
||||
assert tools_service.WEB_SEARCH_SCHEMA["function"]["parameters"]["required"] == ["query"]
|
||||
for name, tool in tools_service.REGISTRY.items():
|
||||
json.dumps(tool.schema)
|
||||
assert tool.schema["function"]["name"] == name
|
||||
assert tool.family in tools_service.FAMILIES
|
||||
|
||||
|
||||
def test_every_tool_describes_when_to_use_it():
|
||||
"""The description is all the model has to decide with."""
|
||||
for tool in tools_service.REGISTRY.values():
|
||||
assert len(tool.description) > 40, tool.name
|
||||
|
||||
Reference in New Issue
Block a user