792 lines
28 KiB
Python
792 lines
28 KiB
Python
import asyncio
|
|
import io
|
|
import json
|
|
import zipfile
|
|
from urllib.parse import urlsplit
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
from memrelay.memory_service import build_context, rebuild_memory_curation_states
|
|
from memrelay.models import CuratedDocument, MemoryReference, UsageProfile
|
|
from tests.conftest import csrf_headers
|
|
from tests.fakes import FakeBasicMemory
|
|
|
|
|
|
def create_project(client: TestClient) -> str:
|
|
response = client.post(
|
|
"/api/v1/projects",
|
|
json={"name": "Memory Project"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
return response.json()["id"]
|
|
|
|
|
|
def save_memory(client: TestClient, payload: dict) -> dict:
|
|
response = client.post("/api/v1/memories", json=payload, headers=csrf_headers(client))
|
|
assert response.status_code == 201, response.text
|
|
return response.json()
|
|
|
|
|
|
def test_api_timestamps_carry_utc_offset(initialized_client: TestClient) -> None:
|
|
# SQLite 丢弃时区信息，ORM 读回 naive 值直接 isoformat 会输出无偏移的时间串，
|
|
# 前端 new Date() 与 MCP 客户端会把它误当本地时间（显示偏差 8 小时）。
|
|
# UTCDateTime 类型必须保证 DB 往返后的序列化仍带 +00:00。
|
|
client = initialized_client
|
|
client.app.state.basic_memory = FakeBasicMemory()
|
|
save_memory(
|
|
client,
|
|
{
|
|
"request_id": "request-tz",
|
|
"scope": "global",
|
|
"memory_type": "fact",
|
|
"title": "Timezone check",
|
|
"content": "timestamps must stay UTC-aware",
|
|
},
|
|
)
|
|
listed = client.get("/api/v1/memories?scope=global").json()
|
|
record = next(item for item in listed if item["title"] == "Timezone check")
|
|
for field in ("created_at", "updated_at"):
|
|
value = record[field]
|
|
assert value.endswith("+00:00") or value.endswith("Z"), (field, value)
|
|
|
|
|
|
def test_memory_revision_idempotency_search_and_archive(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
fake = FakeBasicMemory()
|
|
client.app.state.basic_memory = fake
|
|
project_id = create_project(client)
|
|
payload = {
|
|
"request_id": "request-1",
|
|
"scope": "project",
|
|
"project_id": project_id,
|
|
"memory_type": "decision",
|
|
"title": "Database choice",
|
|
"content": "Use SQLite for system metadata.",
|
|
"tags": ["database"],
|
|
}
|
|
created = save_memory(client, payload)
|
|
assert created["revision"] == 1
|
|
assert fake.write_calls == 1
|
|
assert fake.notes[created["path"]]["metadata"]["stable_id"] == created["id"]
|
|
created_at = fake.notes[created["path"]]["metadata"]["created_at"]
|
|
|
|
repeated = save_memory(client, {**payload, "content": "ignored by idempotency"})
|
|
assert repeated == created
|
|
assert fake.write_calls == 1
|
|
|
|
conflict = client.post(
|
|
"/api/v1/memories",
|
|
json={
|
|
**payload,
|
|
"id": created["id"],
|
|
"request_id": "request-2",
|
|
"content": "Changed",
|
|
"expected_revision": 2,
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert conflict.status_code == 409
|
|
assert conflict.json()["error"]["code"] == "REVISION_CONFLICT"
|
|
|
|
updated = save_memory(
|
|
client,
|
|
{
|
|
**payload,
|
|
"id": created["id"],
|
|
"request_id": "request-3",
|
|
"content": "Use SQLite with WAL for metadata.",
|
|
"expected_revision": 1,
|
|
},
|
|
)
|
|
assert updated["revision"] == 2
|
|
assert fake.notes[updated["path"]]["metadata"]["created_at"] == created_at
|
|
assert (
|
|
client.get(f"/api/v1/memories/{created['id']}").json()["content"].startswith("Use SQLite")
|
|
)
|
|
|
|
search = client.get("/api/v1/memories/search", params={"query": "SQLite"}).json()
|
|
assert search["results"][0]["id"] == created["id"]
|
|
assert search["results"][0]["project_id"] == project_id
|
|
assert search["results"][0]["project_name"] == "Memory Project"
|
|
|
|
archived = client.post(
|
|
f"/api/v1/memories/{created['id']}/archive",
|
|
params={"expected_revision": 2},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert archived.status_code == 200
|
|
assert archived.json()["status"] == "archived"
|
|
deleted = client.delete(f"/api/v1/memories/{created['id']}", headers=csrf_headers(client))
|
|
assert deleted.status_code == 204
|
|
assert client.get(f"/api/v1/memories/{created['id']}").status_code == 404
|
|
|
|
|
|
def test_context_order_limit_and_semantic_fallback(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
fake = FakeBasicMemory()
|
|
client.app.state.basic_memory = fake
|
|
project_id = create_project(client)
|
|
global_rule = save_memory(
|
|
client,
|
|
{
|
|
"request_id": "global-rule",
|
|
"scope": "global",
|
|
"memory_type": "rule",
|
|
"title": "Language",
|
|
"content": "Always reply in Chinese.",
|
|
},
|
|
)
|
|
project_fact = save_memory(
|
|
client,
|
|
{
|
|
"request_id": "project-fact",
|
|
"scope": "project",
|
|
"project_id": project_id,
|
|
"memory_type": "fact",
|
|
"title": "Stack",
|
|
"content": "The backend uses FastAPI.",
|
|
},
|
|
)
|
|
fake.fail_semantic = True
|
|
context = client.post(
|
|
"/api/v1/memories/context",
|
|
json={"project_id": project_id, "query": "FastAPI", "max_chars": 4000},
|
|
)
|
|
assert context.status_code == 200
|
|
body = context.json()
|
|
assert body["semantic_degraded"] is True
|
|
assert body["included_memory_ids"][:2] == [global_rule["id"], project_fact["id"]]
|
|
assert body["content"].index("Always reply") < body["content"].index("FastAPI")
|
|
|
|
|
|
def test_project_search_includes_global_memory_unless_scope_is_explicit(
|
|
initialized_client: TestClient,
|
|
) -> None:
|
|
client = initialized_client
|
|
fake = FakeBasicMemory()
|
|
client.app.state.basic_memory = fake
|
|
project_id = create_project(client)
|
|
global_rule = save_memory(
|
|
client,
|
|
{
|
|
"request_id": "project-search-global-rule",
|
|
"scope": "global",
|
|
"memory_type": "rule",
|
|
"title": "Shared retrieval rule",
|
|
"content": "Shared project search marker for every workspace.",
|
|
},
|
|
)
|
|
with client.app.state.session_factory() as db:
|
|
document = CuratedDocument(
|
|
scope="global",
|
|
usage_profile_id=global_rule["usage_profile_id"],
|
|
document_type="profile",
|
|
title="Shared retrieval profile",
|
|
path="global/curated/shared-retrieval-profile",
|
|
source_hash="c" * 64,
|
|
)
|
|
db.add(document)
|
|
db.commit()
|
|
fake.notes[document.path] = {
|
|
"title": document.title,
|
|
"content": "Shared project search marker in the consolidated profile.",
|
|
"permalink": document.path,
|
|
}
|
|
|
|
inherited = client.get(
|
|
"/api/v1/memories/search",
|
|
params={"query": "shared project search marker", "project_id": project_id},
|
|
)
|
|
assert inherited.status_code == 200
|
|
inherited_results = inherited.json()["results"]
|
|
assert {item["id"] for item in inherited_results} == {global_rule["id"], document.id}
|
|
assert {item["scope"] for item in inherited_results} == {"global"}
|
|
|
|
strict_project = client.get(
|
|
"/api/v1/memories/search",
|
|
params={
|
|
"query": "shared project search marker",
|
|
"project_id": project_id,
|
|
"scope": "project",
|
|
},
|
|
)
|
|
assert strict_project.status_code == 200
|
|
assert strict_project.json()["results"] == []
|
|
|
|
strict_global = client.get(
|
|
"/api/v1/memories/search",
|
|
params={
|
|
"query": "shared project search marker",
|
|
"project_id": project_id,
|
|
"scope": "global",
|
|
},
|
|
)
|
|
assert strict_global.status_code == 200
|
|
assert {item["id"] for item in strict_global.json()["results"]} == {
|
|
global_rule["id"],
|
|
document.id,
|
|
}
|
|
|
|
|
|
def test_context_includes_every_durable_memory_type_without_curation(
|
|
initialized_client: TestClient,
|
|
) -> None:
|
|
client = initialized_client
|
|
fake = FakeBasicMemory()
|
|
client.app.state.basic_memory = fake
|
|
project_id = create_project(client)
|
|
memory_types = [
|
|
"rule",
|
|
"preference",
|
|
"fact",
|
|
"decision",
|
|
"experience",
|
|
"prd",
|
|
"task_list",
|
|
]
|
|
expected_ids: list[str] = []
|
|
for scope in ("global", "project"):
|
|
for memory_type in memory_types:
|
|
marker = f"{scope}-{memory_type}"
|
|
payload = {
|
|
"request_id": marker,
|
|
"scope": scope,
|
|
"memory_type": memory_type,
|
|
"title": marker,
|
|
"content": f"Durable context for {marker}.",
|
|
}
|
|
if scope == "project":
|
|
payload["project_id"] = project_id
|
|
expected_ids.append(save_memory(client, payload)["id"])
|
|
|
|
context = client.post(
|
|
"/api/v1/memories/context",
|
|
json={"project_id": project_id, "max_chars": 100_000},
|
|
)
|
|
|
|
assert context.status_code == 200
|
|
body = context.json()
|
|
assert body["included_memory_ids"] == expected_ids
|
|
for scope in ("global", "project"):
|
|
positions = [
|
|
body["content"].index(f"Durable context for {scope}-{item}") for item in memory_types
|
|
]
|
|
assert positions == sorted(positions)
|
|
|
|
|
|
def test_context_keeps_recent_progress_when_curated_documents_are_large(
|
|
initialized_client: TestClient,
|
|
) -> None:
|
|
client = initialized_client
|
|
fake = FakeBasicMemory()
|
|
client.app.state.basic_memory = fake
|
|
project_id = create_project(client)
|
|
checkpoint = save_memory(
|
|
client,
|
|
{
|
|
"request_id": "large-context-checkpoint",
|
|
"scope": "project",
|
|
"project_id": project_id,
|
|
"memory_type": "checkpoint",
|
|
"title": "Latest progress",
|
|
"content": "The latest checkpoint must remain visible.",
|
|
},
|
|
)
|
|
with client.app.state.session_factory() as db:
|
|
documents = [
|
|
CuratedDocument(
|
|
scope="project",
|
|
project_id=project_id,
|
|
document_type=document_type,
|
|
title=f"Large {document_type}",
|
|
path=f"projects/{project_id}/curated/{document_type}",
|
|
source_hash=str(index) * 64,
|
|
)
|
|
for index, document_type in enumerate(("architecture", "decisions"), start=1)
|
|
]
|
|
db.add_all(documents)
|
|
db.commit()
|
|
for document in documents:
|
|
fake.notes[document.path] = {
|
|
"title": document.title,
|
|
"content": f"{document.title}\n" + ("Long stable knowledge. " * 200),
|
|
"permalink": document.path,
|
|
}
|
|
|
|
context = client.post(
|
|
"/api/v1/memories/context",
|
|
json={"project_id": project_id, "max_chars": 1000},
|
|
)
|
|
|
|
assert context.status_code == 200
|
|
body = context.json()
|
|
assert body["truncated"] is True
|
|
assert body["included_curated_document_ids"] == [item.id for item in documents]
|
|
assert checkpoint["id"] in body["included_memory_ids"]
|
|
assert "The latest checkpoint must remain visible" in body["content"]
|
|
|
|
|
|
def test_memory_lifecycle_hides_curated_sources_without_deleting_history(
|
|
initialized_client: TestClient,
|
|
) -> None:
|
|
client = initialized_client
|
|
fake = FakeBasicMemory()
|
|
client.app.state.basic_memory = fake
|
|
memory = save_memory(
|
|
client,
|
|
{
|
|
"request_id": "lifecycle-source",
|
|
"scope": "global",
|
|
"memory_type": "rule",
|
|
"title": "Lifecycle source",
|
|
"content": "Keep the original Markdown as historical evidence.",
|
|
},
|
|
)
|
|
with client.app.state.session_factory() as db:
|
|
document = CuratedDocument(
|
|
scope="global",
|
|
usage_profile_id=memory["usage_profile_id"],
|
|
document_type="profile",
|
|
title="Stable profile",
|
|
path="global/curated/profile",
|
|
source_hash="a" * 64,
|
|
metadata_json=json.dumps(
|
|
{
|
|
"cited_source_ids": [f"memory:{memory['id']}"],
|
|
"source_revisions": {f"memory:{memory['id']}": str(memory["revision"])},
|
|
"supersedes": [],
|
|
}
|
|
),
|
|
)
|
|
db.add(document)
|
|
db.commit()
|
|
rebuild_memory_curation_states(db)
|
|
fake.notes[document.path] = {
|
|
"title": document.title,
|
|
"content": "Stable profile preserves historical evidence.",
|
|
"permalink": document.path,
|
|
}
|
|
|
|
assert client.get("/api/v1/memories", params={"scope": "global"}).json() == []
|
|
covered = client.get(
|
|
"/api/v1/memories",
|
|
params={"scope": "global", "lifecycle": "covered"},
|
|
).json()
|
|
assert covered[0]["id"] == memory["id"]
|
|
assert covered[0]["curation_status"] == "covered"
|
|
assert covered[0]["curated_revision"] == memory["revision"]
|
|
assert covered[0]["covered_by"] == [document.id]
|
|
assert memory["path"] in fake.notes
|
|
|
|
current_search = client.get(
|
|
"/api/v1/memories/search",
|
|
params={"query": "historical evidence"},
|
|
).json()["results"]
|
|
assert len(current_search) == 1
|
|
assert current_search[0]["id"] == document.id
|
|
assert current_search[0]["result_kind"] == "curated_document"
|
|
assert current_search[0]["read_tool"] == "curated_document_get"
|
|
assert current_search[0]["document_type"] == "profile"
|
|
pending_search = client.get(
|
|
"/api/v1/memories/search",
|
|
params={"query": "historical evidence", "lifecycle": "pending"},
|
|
).json()["results"]
|
|
assert pending_search == []
|
|
history_search = client.get(
|
|
"/api/v1/memories/search",
|
|
params={"query": "historical evidence", "lifecycle": "all"},
|
|
).json()
|
|
assert history_search["results"][0]["curation_status"] == "covered"
|
|
|
|
reset = client.post(
|
|
f"/api/v1/memories/{memory['id']}/pending",
|
|
params={"expected_revision": memory["revision"]},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert reset.status_code == 200
|
|
assert reset.json()["curation_status"] == "pending"
|
|
assert (
|
|
client.get("/api/v1/memories", params={"scope": "global"}).json()[0]["id"] == memory["id"]
|
|
)
|
|
|
|
|
|
def test_legacy_memory_without_profile_is_covered_by_profile_scoped_document(
|
|
initialized_client: TestClient,
|
|
) -> None:
|
|
# 生产回归：早于记忆空间功能创建的全局记忆 usage_profile_id 为 NULL，对所有
|
|
# 空间可见并会被整理引用，但旧覆盖匹配不认 NULL，被引用后仍停留在待整理。
|
|
client = initialized_client
|
|
fake = FakeBasicMemory()
|
|
client.app.state.basic_memory = fake
|
|
memory = save_memory(
|
|
client,
|
|
{
|
|
"request_id": "legacy-profile-source",
|
|
"scope": "global",
|
|
"memory_type": "rule",
|
|
"title": "Legacy rule",
|
|
"content": "Created before usage profiles existed.",
|
|
},
|
|
)
|
|
with client.app.state.session_factory() as db:
|
|
reference = db.get(MemoryReference, memory["id"])
|
|
reference.usage_profile_id = None
|
|
document = CuratedDocument(
|
|
scope="global",
|
|
usage_profile_id=memory["usage_profile_id"],
|
|
document_type="profile",
|
|
title="Stable profile",
|
|
path="global/curated/legacy-profile",
|
|
source_hash="c" * 64,
|
|
metadata_json=json.dumps(
|
|
{
|
|
"cited_source_ids": [f"memory:{memory['id']}"],
|
|
"source_revisions": {f"memory:{memory['id']}": str(memory["revision"])},
|
|
"supersedes": [],
|
|
}
|
|
),
|
|
)
|
|
db.add(document)
|
|
db.commit()
|
|
counts = rebuild_memory_curation_states(db)
|
|
assert counts["covered"] == 1
|
|
|
|
covered = client.get(
|
|
"/api/v1/memories",
|
|
params={"scope": "global", "lifecycle": "covered"},
|
|
).json()
|
|
assert covered[0]["id"] == memory["id"]
|
|
assert covered[0]["covered_reason"] == "cited"
|
|
|
|
|
|
def test_editing_a_curated_memory_returns_it_to_pending(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
fake = FakeBasicMemory()
|
|
client.app.state.basic_memory = fake
|
|
memory = save_memory(
|
|
client,
|
|
{
|
|
"request_id": "curated-edit-source",
|
|
"scope": "global",
|
|
"memory_type": "fact",
|
|
"title": "Curated fact",
|
|
"content": "Original fact.",
|
|
},
|
|
)
|
|
with client.app.state.session_factory() as db:
|
|
document = CuratedDocument(
|
|
scope="global",
|
|
usage_profile_id=memory["usage_profile_id"],
|
|
document_type="profile",
|
|
title="Stable profile",
|
|
path="global/curated/edit-profile",
|
|
source_hash="b" * 64,
|
|
metadata_json=json.dumps(
|
|
{
|
|
"cited_source_ids": [f"memory:{memory['id']}"],
|
|
"source_revisions": {f"memory:{memory['id']}": "1"},
|
|
"supersedes": [],
|
|
}
|
|
),
|
|
)
|
|
db.add(document)
|
|
db.commit()
|
|
rebuild_memory_curation_states(db)
|
|
|
|
updated = save_memory(
|
|
client,
|
|
{
|
|
"id": memory["id"],
|
|
"request_id": "curated-edit-update",
|
|
"scope": "global",
|
|
"memory_type": "fact",
|
|
"title": "Curated fact",
|
|
"content": "Updated fact.",
|
|
"expected_revision": 1,
|
|
},
|
|
)
|
|
assert updated["revision"] == 2
|
|
assert updated["curation_status"] == "pending"
|
|
assert updated["curated_revision"] is None
|
|
assert updated["covered_by"] == []
|
|
|
|
|
|
def test_context_keeps_personal_preferences_in_the_current_usage_profile(
|
|
initialized_client: TestClient,
|
|
) -> None:
|
|
client = initialized_client
|
|
fake = FakeBasicMemory()
|
|
client.app.state.basic_memory = fake
|
|
project_id = create_project(client)
|
|
alice = client.post(
|
|
"/api/v1/curation/profiles",
|
|
json={"name": "alice", "description": "Alice preferences"},
|
|
headers=csrf_headers(client),
|
|
).json()
|
|
bob = client.post(
|
|
"/api/v1/curation/profiles",
|
|
json={"name": "bob", "description": "Bob preferences"},
|
|
headers=csrf_headers(client),
|
|
).json()
|
|
alice_memory = save_memory(
|
|
client,
|
|
{
|
|
"request_id": "alice-preference",
|
|
"scope": "global",
|
|
"usage_profile_id": alice["id"],
|
|
"memory_type": "preference",
|
|
"title": "Alice editor",
|
|
"content": "Alice prefers compact editor layouts.",
|
|
},
|
|
)
|
|
bob_memory = save_memory(
|
|
client,
|
|
{
|
|
"request_id": "bob-preference",
|
|
"scope": "global",
|
|
"usage_profile_id": bob["id"],
|
|
"memory_type": "preference",
|
|
"title": "Bob editor",
|
|
"content": "Bob prefers spacious editor layouts.",
|
|
},
|
|
)
|
|
project_fact = save_memory(
|
|
client,
|
|
{
|
|
"request_id": "shared-project-fact",
|
|
"scope": "project",
|
|
"project_id": project_id,
|
|
"usage_profile_id": alice["id"],
|
|
"memory_type": "fact",
|
|
"title": "Shared stack",
|
|
"content": "The project uses FastAPI.",
|
|
},
|
|
)
|
|
|
|
with client.app.state.session_factory() as db:
|
|
alice_context = asyncio.run(build_context(db, fake, project_id, None, 4000, alice["id"]))
|
|
with client.app.state.session_factory() as db:
|
|
bob_context = asyncio.run(build_context(db, fake, project_id, None, 4000, bob["id"]))
|
|
|
|
assert alice_memory["id"] in alice_context.included_memory_ids
|
|
assert bob_memory["id"] not in alice_context.included_memory_ids
|
|
assert bob_memory["id"] in bob_context.included_memory_ids
|
|
assert alice_memory["id"] not in bob_context.included_memory_ids
|
|
assert project_fact["id"] in alice_context.included_memory_ids
|
|
assert project_fact["id"] in bob_context.included_memory_ids
|
|
assert fake.notes[alice_memory["path"]]["metadata"]["usage_profile_id"] == alice["id"]
|
|
|
|
|
|
def test_context_includes_only_the_current_workspace_scenario_documents(
|
|
initialized_client: TestClient,
|
|
) -> None:
|
|
client = initialized_client
|
|
fake = FakeBasicMemory()
|
|
client.app.state.basic_memory = fake
|
|
project = client.post(
|
|
"/api/v1/projects",
|
|
json={"name": "Development context", "workspace_type": "development"},
|
|
headers=csrf_headers(client),
|
|
).json()
|
|
alice = client.post(
|
|
"/api/v1/curation/profiles",
|
|
json={"name": "context-alice", "description": "Alice context"},
|
|
headers=csrf_headers(client),
|
|
).json()
|
|
bob = client.post(
|
|
"/api/v1/curation/profiles",
|
|
json={"name": "context-bob", "description": "Bob context"},
|
|
headers=csrf_headers(client),
|
|
).json()
|
|
with client.app.state.session_factory() as db:
|
|
shared = db.scalar(select(UsageProfile).where(UsageProfile.is_shared.is_(True)))
|
|
assert shared is not None
|
|
documents = [
|
|
CuratedDocument(
|
|
scope="global",
|
|
usage_profile_id=shared.id,
|
|
document_type="preferences",
|
|
title="Shared preferences",
|
|
path="global/curated/shared/preferences",
|
|
source_hash="1" * 64,
|
|
),
|
|
CuratedDocument(
|
|
scope="global",
|
|
usage_profile_id=alice["id"],
|
|
document_type="context_development",
|
|
title="Alice development",
|
|
path="global/curated/alice/context-development",
|
|
source_hash="2" * 64,
|
|
),
|
|
CuratedDocument(
|
|
scope="global",
|
|
usage_profile_id=alice["id"],
|
|
document_type="context_office",
|
|
title="Alice office",
|
|
path="global/curated/alice/context-office",
|
|
source_hash="3" * 64,
|
|
),
|
|
CuratedDocument(
|
|
scope="global",
|
|
usage_profile_id=bob["id"],
|
|
document_type="context_development",
|
|
title="Bob development",
|
|
path="global/curated/bob/context-development",
|
|
source_hash="4" * 64,
|
|
),
|
|
CuratedDocument(
|
|
scope="project",
|
|
project_id=project["id"],
|
|
document_type="overview",
|
|
title="Workspace overview",
|
|
path=f"projects/{project['id']}/curated/overview",
|
|
source_hash="5" * 64,
|
|
),
|
|
]
|
|
db.add_all(documents)
|
|
db.commit()
|
|
document_ids = {item.title: item.id for item in documents}
|
|
for title, path in {
|
|
"Shared preferences": "global/curated/shared/preferences",
|
|
"Alice development": "global/curated/alice/context-development",
|
|
"Alice office": "global/curated/alice/context-office",
|
|
"Bob development": "global/curated/bob/context-development",
|
|
"Workspace overview": f"projects/{project['id']}/curated/overview",
|
|
}.items():
|
|
fake.notes[path] = {
|
|
"title": title,
|
|
"content": f"# {title}\n\n{title} body",
|
|
"permalink": path,
|
|
}
|
|
|
|
with client.app.state.session_factory() as db:
|
|
context = asyncio.run(build_context(db, fake, project["id"], None, 10_000, alice["id"]))
|
|
|
|
assert document_ids["Shared preferences"] in context.included_curated_document_ids
|
|
assert document_ids["Alice development"] in context.included_curated_document_ids
|
|
assert document_ids["Workspace overview"] in context.included_curated_document_ids
|
|
assert document_ids["Alice office"] not in context.included_curated_document_ids
|
|
assert document_ids["Bob development"] not in context.included_curated_document_ids
|
|
|
|
|
|
def test_project_scope_validation(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
client.app.state.basic_memory = FakeBasicMemory()
|
|
response = client.post(
|
|
"/api/v1/memories",
|
|
json={
|
|
"request_id": "invalid-scope",
|
|
"scope": "project",
|
|
"memory_type": "fact",
|
|
"title": "Invalid",
|
|
"content": "Missing project.",
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert response.status_code == 422
|
|
assert response.json()["error"]["code"] == "MEMORY_SCOPE_INVALID"
|
|
|
|
|
|
def test_memory_title_can_be_renamed(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
fake = FakeBasicMemory()
|
|
client.app.state.basic_memory = fake
|
|
created = save_memory(
|
|
client,
|
|
{
|
|
"request_id": "rename-create",
|
|
"scope": "global",
|
|
"memory_type": "fact",
|
|
"title": "Old title",
|
|
"content": "Original content",
|
|
},
|
|
)
|
|
old_path = created["path"]
|
|
renamed = save_memory(
|
|
client,
|
|
{
|
|
"id": created["id"],
|
|
"request_id": "rename-update",
|
|
"scope": "global",
|
|
"memory_type": "fact",
|
|
"title": "New title",
|
|
"content": "Original content",
|
|
"expected_revision": 1,
|
|
},
|
|
)
|
|
assert renamed["title"] == "New title"
|
|
assert renamed["revision"] == 2
|
|
assert renamed["path"] != old_path
|
|
assert old_path not in fake.notes
|
|
assert renamed["path"] in fake.notes
|
|
|
|
|
|
def test_project_export_contains_timeline_and_markdown(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
fake = FakeBasicMemory()
|
|
client.app.state.basic_memory = fake
|
|
project_id = create_project(client)
|
|
save_memory(
|
|
client,
|
|
{
|
|
"request_id": "export-global",
|
|
"scope": "global",
|
|
"memory_type": "rule",
|
|
"title": "Language",
|
|
"content": "Always reply in Chinese.",
|
|
},
|
|
)
|
|
checkpoint = save_memory(
|
|
client,
|
|
{
|
|
"request_id": "export-checkpoint",
|
|
"scope": "project",
|
|
"project_id": project_id,
|
|
"memory_type": "checkpoint",
|
|
"title": "Implemented export",
|
|
"content": "## 完成内容\n\nAdded direct ZIP export.",
|
|
},
|
|
)
|
|
with client.app.state.session_factory() as db:
|
|
curated = CuratedDocument(
|
|
scope="project",
|
|
project_id=project_id,
|
|
document_type="overview",
|
|
title="Current project overview",
|
|
path=f"projects/{project_id}/curated/overview",
|
|
revision=2,
|
|
source_hash="e" * 64,
|
|
)
|
|
db.add(curated)
|
|
db.commit()
|
|
fake.notes[curated.path] = {
|
|
"title": curated.title,
|
|
"content": "The current consolidated project overview.",
|
|
"permalink": curated.path,
|
|
}
|
|
|
|
prepared = client.get(f"/api/v1/projects/{project_id}/export")
|
|
assert prepared.status_code == 200
|
|
path = urlsplit(prepared.json()["url"])
|
|
downloaded = client.get(f"{path.path}?{path.query}")
|
|
assert downloaded.status_code == 200
|
|
assert downloaded.headers["cache-control"] == "no-store"
|
|
|
|
with zipfile.ZipFile(io.BytesIO(downloaded.content)) as archive:
|
|
names = archive.namelist()
|
|
assert "aidocs/project_context.md" in names
|
|
assert "aidocs/manifest.json" in names
|
|
assert "aidocs/curated/project/overview.md" in names
|
|
assert any(checkpoint["id"] in name for name in names)
|
|
timeline = archive.read("aidocs/project_context.md").decode()
|
|
assert "Implemented export" in timeline
|
|
assert "Added direct ZIP export" in timeline
|
|
assert "curated/project/overview.md" in timeline
|
|
curated_content = archive.read("aidocs/curated/project/overview.md").decode()
|
|
assert "The current consolidated project overview" in curated_content
|
|
assert "revision: 2" in curated_content
|
|
manifest = json.loads(archive.read("aidocs/manifest.json"))
|
|
assert manifest["project"]["id"] == project_id
|
|
assert manifest["project_curated_documents"][0]["id"] == curated.id
|