Spaces:
Runtime error
Runtime error
File size: 6,435 Bytes
b74c6ae | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | import uuid
from contextlib import contextmanager
from types import SimpleNamespace
import pytest
from app.models import WorkspaceMemory
from app.services import memory
from tests.conftest import last_code
def _signup(client, email):
client.post("/api/auth/email/register/start", json={"email": email, "password": "stardust-42"})
r = client.post("/api/auth/email/register/verify", json={"email": email, "code": last_code()})
return {"Authorization": f"Bearer {r.json()['access_token']}"}
@pytest.fixture()
def auth(client):
return _signup(client, "ada@example.com")
def put(client, auth, key, value):
return client.put(f"/api/memory/{key}", json={"value": value}, headers=auth)
def workspace_id(db_session):
return db_session.query(WorkspaceMemory).one().workspace_id
def test_create_update_and_delete(client, auth):
assert client.get("/api/memory", headers=auth).json() == {"items": [], "limit": memory.MAX_FACTS}
assert put(client, auth, "Audience", "indie founders").json()["source"] == "user"
put(client, auth, "audience", " solo founders ") # same key once cleaned: replaces, never duplicates
items = client.get("/api/memory", headers=auth).json()["items"]
assert [(i["key"], i["value"]) for i in items] == [("audience", "solo founders")]
assert client.delete("/api/memory/audience", headers=auth).status_code == 200
assert client.delete("/api/memory/audience", headers=auth).status_code == 404
assert client.get("/api/memory", headers=auth).json()["items"] == []
def test_a_workspace_has_one_row_holding_every_note(client, auth, db_session):
for i in range(3):
put(client, auth, f"note-{i}", "v")
rows = db_session.query(WorkspaceMemory).all()
assert len(rows) == 1 and sorted(rows[0].facts) == ["note-0", "note-1", "note-2"]
def test_other_workspaces_cannot_see_or_touch_notes(client, auth):
put(client, auth, "tone", "short answers")
other = _signup(client, "bob@example.com")
assert client.get("/api/memory", headers=other).json()["items"] == []
assert client.delete("/api/memory/tone", headers=other).status_code == 404
put(client, other, "tone", "long answers")
assert client.get("/api/memory", headers=auth).json()["items"][0]["value"] == "short answers"
def test_limits_and_empty_values(client, auth):
assert put(client, auth, "big", "x" * (memory.MAX_VALUE + 1)).status_code == 422
assert put(client, auth, "blank", " ").status_code == 422
assert put(client, auth, "---", "value").status_code == 422 # the name cleans to nothing
for i in range(memory.MAX_FACTS):
assert put(client, auth, f"fact-{i}", "v").status_code == 200
assert put(client, auth, "one-too-many", "v").status_code == 422
assert put(client, auth, "fact-0", "changed").status_code == 200 # changing a note at the cap is fine
def test_the_job_adds_updates_and_deletes_its_own_notes(client, auth, db_session):
put(client, auth, "audience", "typed by the writer")
ws = workspace_id(db_session)
applied = memory.apply_operations(db_session, ws, [
{"op": "add", "key": "Tone", "value": "short answers"},
{"op": "add", "key": "avoid", "value": "jargon"},
{"op": "update", "key": "tone", "value": "warm and short"},
{"op": "delete", "key": "avoid"},
])
assert [(a["op"], a["key"]) for a in applied] == [("add", "tone"), ("add", "avoid"), ("update", "tone"),
("delete", "avoid")]
items = {i["key"]: i for i in memory.list_facts(db_session, ws)}
assert items["tone"]["value"] == "warm and short" and items["tone"]["source"] == "auto"
assert "avoid" not in items
def test_the_job_never_changes_notes_the_writer_typed(client, auth, db_session):
put(client, auth, "audience", "indie founders")
ws = workspace_id(db_session)
applied = memory.apply_operations(db_session, ws, [
{"op": "update", "key": "audience", "value": "designers"},
{"op": "delete", "key": "audience"},
])
assert applied == []
assert memory.list_facts(db_session, ws)[0]["value"] == "indie founders"
def test_the_job_skips_bad_operations_repeats_and_the_cap(client, auth, db_session):
put(client, auth, "seed", "v")
ws = workspace_id(db_session)
assert memory.apply_operations(db_session, ws, [
{"op": "add", "key": "", "value": "no key"},
{"op": "add", "key": "empty", "value": " "},
{"op": "add", "key": "huge", "value": "x" * (memory.MAX_VALUE + 1)},
{"op": "delete", "key": "missing"},
{"op": "rename", "key": "seed", "value": "x"},
]) == []
assert memory.apply_operations(db_session, ws, [{"op": "add", "key": "tone", "value": "short"}])
assert memory.apply_operations(db_session, ws, [{"op": "update", "key": "tone", "value": "short"}]) == []
for i in range(memory.MAX_FACTS):
memory.apply_operations(db_session, ws, [{"op": "add", "key": f"auto-{i}", "value": "v"}])
assert len(memory.list_facts(db_session, ws)) == memory.MAX_FACTS
def test_profile_text_is_capped_and_marks_empty():
assert memory.profile_text([]) == "(none)"
items = [{"key": f"k{i}", "value": "v" * 400} for i in range(10)]
assert len(memory.profile_text(items)) == memory.PROFILE_CHARS
def test_research_gives_the_profile_to_the_agent(monkeypatch, tmp_path):
from app.corpus import Corpus
from app.pipeline import research as r
seen = {}
def fake_react(signature, tools, max_iters):
def run(**inputs):
seen.update(inputs)
return SimpleNamespace(answer="x", citations=[], unsupported=True)
return run
@contextmanager
def no_usage(lm):
yield SimpleNamespace(prompt_tokens=0, completion_tokens=0)
monkeypatch.setattr(r.dspy, "ReAct", fake_react)
monkeypatch.setattr(r, "main_lm", lambda: None)
monkeypatch.setattr(r, "track_usage", no_usage)
monkeypatch.setattr(r, "triage", lambda *a: r.Triage("archive", "q", "quick", ""))
corpus = Corpus(uuid.uuid4(), cache_dir=str(tmp_path))
monkeypatch.setattr(corpus, "sync", lambda: {})
r.research(corpus, {"a.md": "On Pricing"}, "What did I charge?", profile="- audience: founders")
assert seen["writer_profile"] == "- audience: founders"
r.research(corpus, {"a.md": "On Pricing"}, "What did I charge?")
assert seen["writer_profile"] == "(none)"
|