Sixteen mutating routes now refuse a caller who cannot present the token the server started with; queries stay open, because a question is not a write. The environment variable is ELY_EYE_AUTH_TOKEN, following the repo's env_prefix convention — the shorter name I reached for first returned 403 where the smoke expected 401, and the server printed the reason. Words from a stranger are quarantined on the way in: an HTTP upload is third_party by definition, and a sentence carrying an instruction-shaped phrase is quarantined whoever wrote it. A quarantined atom stays findable and stays out of the compiled context until a second, live, unquarantined source agrees with its rarest words. Corroboration is computed at read time because it is a property of the corpus now, not of the moment a memory was written; it is term-level, and the docstring says so. forget now erases the source row and the compiled contexts beside the atoms, so a forgotten memory leaves nothing that can still speak. The semantic cache key carries a store revision, which fixes a real staleness: a context compiled before new evidence arrived kept its old quarantine decision forever. bench.audit --product still pointed at evals.py, which the p1-6 split moved into backend/src/ely_eye/proofs/. It failed fast, but the anti-self-justification rule had quietly stopped auditing the product; it now walks the package, and a test pins that the package exists and holds no re-roll. Gates: pytest 306 passed; ruff clean; mypy cold cache 24 errors in 9 files, the recorded baseline, net zero; pyright 0 errors on the 23 backend files and the 2 bench files; go build/vet/test and gofmt clean; npm run build 221.39 kB with eslint exit 0; bench/filesize_check clean; bench.audit --product clean over 31 harness files; scripts/mcp_write_smoke.sh PASS with the gate live across two servers and the MCP child.
70 lines
2.7 KiB
Python
70 lines
2.7 KiB
Python
"""The HTTP face stamps provenance itself: a client cannot say who owns a memory."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
import ely_eye.api as api_module
|
|
from ely_eye.config import Settings
|
|
|
|
KEYS = "我们的钥匙放在玄关抽屉里"
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, write_token: str) -> TestClient:
|
|
settings = Settings(home=tmp_path, default_holder="wife", auth_token=write_token)
|
|
monkeypatch.setattr(api_module, "get_settings", lambda: settings)
|
|
return TestClient(api_module.create_app(), headers={"Authorization": f"Bearer {write_token}"})
|
|
|
|
|
|
def test_post_memory_stamps_the_holder_the_server_knows(client: TestClient) -> None:
|
|
response = client.post("/api/memory", json={"text": KEYS, "scenario": "home"})
|
|
assert response.status_code == 200
|
|
atom = response.json()
|
|
assert atom["holder_id"] == "wife"
|
|
assert atom["origin"] == "self"
|
|
assert atom["scenario"] == "home"
|
|
assert atom["ingested_at"], "ingestion time is stamped, never supplied"
|
|
|
|
|
|
def test_a_client_cannot_claim_another_holder(client: TestClient) -> None:
|
|
response = client.post(
|
|
"/api/memory",
|
|
json={"text": KEYS, "holder_id": "attacker", "origin": "third_party", "ingested_at": "2020-01-01T00:00:00+00:00"},
|
|
)
|
|
assert response.status_code == 200
|
|
atom = response.json()
|
|
assert atom["holder_id"] == "wife", "the request body has no holder field to forge"
|
|
assert atom["origin"] == "self"
|
|
|
|
|
|
def test_an_empty_memory_is_rejected_with_its_reason(client: TestClient) -> None:
|
|
response = client.post("/api/memory", json={"text": " "})
|
|
assert response.status_code == 422
|
|
assert "needs text" in response.json()["detail"]
|
|
|
|
|
|
def test_supersede_then_forget_reports_what_it_removed(client: TestClient) -> None:
|
|
old = client.post("/api/memory", json={"text": KEYS, "scenario": "home"}).json()
|
|
new = client.post(
|
|
"/api/memory/supersede",
|
|
json={"atom_id": old["atom_id"], "text": "钥匙现在在书房第二个抽屉", "scenario": "home"},
|
|
).json()
|
|
assert new["supersedes"] == [old["atom_id"]]
|
|
|
|
forgotten = client.post("/api/memory/forget", json={"atom_id": old["atom_id"], "reason": "moved"})
|
|
assert forgotten.status_code == 200
|
|
report = forgotten.json()
|
|
assert report["atoms_deleted"] == 1
|
|
assert report["atom_ids"] == [old["atom_id"]]
|
|
assert report["reason"] == "moved"
|
|
|
|
|
|
def test_a_forget_that_names_nothing_is_rejected(client: TestClient) -> None:
|
|
response = client.post("/api/memory/forget", json={"reason": "vague"})
|
|
assert response.status_code == 422
|
|
assert "exactly one" in response.json()["detail"]
|