220 lines
13 KiB
Python
220 lines
13 KiB
Python
"""Focused HUX-07 metadata-only backend tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
|
|
if str(FOUNDATION) not in sys.path:
|
|
sys.path.insert(0, str(FOUNDATION))
|
|
|
|
from hux import contracts, identity, multimodal, store
|
|
from hux.server import build_router
|
|
|
|
HEADERS = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat", "X-Hux-Relay-Key": "rk"}
|
|
OTHER = {**HEADERS, "X-Hux-Subject": "usr_fedcba9876543210"}
|
|
ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
|
|
SCHEMAS = contracts.load_all()
|
|
SCHEMAS["multimodal.schema.json"] = contracts.load_schema("multimodal.schema.json")
|
|
|
|
|
|
def call(router, method, path, body=None, headers=None):
|
|
raw = b"" if body is None else json.dumps(body).encode()
|
|
response = router.dispatch(method, path, {**HEADERS, **(headers or {})}, raw)
|
|
return response.status, response.body, response.headers
|
|
|
|
|
|
def tenant(router, headers=HEADERS):
|
|
return store.TenantStore(router.data_root, identity.resolve(headers, {"HUX_ROUTER_KEY": "rk"}))
|
|
|
|
|
|
def approval(router, conversation_id, capability="artifact_write", name="apr_upload0001"):
|
|
return tenant(router).put("approvals", {
|
|
"id": name, "status": "approved", "conversation_id": conversation_id, "capability": capability,
|
|
})
|
|
|
|
|
|
@pytest.fixture
|
|
def setup(tmp_path):
|
|
router = build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_ROUTER_KEY": "rk"})
|
|
_, project, _ = call(router, "POST", "/hux/v1/projects", {"name": "P"})
|
|
_, conversation, _ = call(router, "POST", "/hux/v1/conversations", {"title": "C", "project_id": project["id"]})
|
|
base = f"/hux/v1/projects/{project['id']}/conversations/{conversation['id']}"
|
|
approval(router, conversation["id"])
|
|
return router, project, conversation, base
|
|
|
|
|
|
def item_body(project, **changes):
|
|
body = {
|
|
"project_id": project["id"], "kind": "audio", "source": "upload", "filename": "note.webm",
|
|
"mime": "audio/webm", "bytes": 1234, "hash": "sha256:" + "a" * 64, "approval_id": "apr_upload0001",
|
|
}
|
|
return {**body, **changes}
|
|
|
|
|
|
def create_item(setup, key="media-key-0001", **changes):
|
|
router, project, _, base = setup
|
|
return call(router, "POST", base + "/multimodal/items", item_body(project, **changes), {"If-Match": "0", "Idempotency-Key": key})
|
|
|
|
|
|
def test_metadata_create_list_get_and_replay(setup):
|
|
router, _, _, base = setup
|
|
status, item, headers = create_item(setup)
|
|
assert (status, item["status"], headers["ETag"]) == (201, "metadata_only", "1")
|
|
assert contracts.validate("multimodal.schema.json", item, SCHEMAS, "/$defs/item") == []
|
|
status, listed, headers = call(router, "GET", base + "/multimodal/items")
|
|
assert status == 200 and listed["items"] == [item] and headers["Cache-Control"] == "no-store"
|
|
status, got, headers = call(router, "GET", base + f"/multimodal/items/{item['id']}")
|
|
assert status == 200 and got == item and headers["ETag"] == "1"
|
|
status, replayed, headers = create_item(setup)
|
|
assert status == 200 and replayed == item and headers["HUX-Replayed"] == "true"
|
|
status, error, _ = create_item(setup, filename="different.webm")
|
|
assert (status, error["code"]) == (409, "conflict")
|
|
assert not list(Path(router.data_root).rglob("*.webm")), "the endpoint stores metadata, never media"
|
|
|
|
|
|
@pytest.mark.parametrize("changes", [
|
|
{"filename": "payload.html", "mime": "text/html", "kind": "document"},
|
|
{"filename": "drawing.svg", "mime": "image/svg+xml", "kind": "image"},
|
|
{"filename": "photo.png", "mime": "audio/webm", "kind": "image"},
|
|
{"filename": "photo.exe", "mime": "image/png", "kind": "image"},
|
|
{"bytes": 0}, {"bytes": 30_000_000}, {"hash": "sha256:no"}, {"kind": "binary"},
|
|
{"content": "raw bytes are forbidden"}, {"url": "https://example.com/a"},
|
|
])
|
|
def test_media_validation_rejects_executable_content_and_unbounded_metadata(setup, changes):
|
|
status, error, _ = create_item(setup, key=f"invalid-media-{abs(hash(repr(changes)))}", **changes)
|
|
assert status == 400 and error["code"] == "invalid"
|
|
|
|
|
|
def test_approval_must_cover_scope_and_action(setup):
|
|
router, project, conversation, base = setup
|
|
common = item_body(project)
|
|
approval(router, conversation["id"], "external_side_effect", "apr_capture0001")
|
|
cases = [
|
|
({**common, "approval_id": "apr_missing0001"}, 403),
|
|
({**common, "approval_id": "apr_capture0001"}, 403),
|
|
({**common, "source": "camera", "approval_id": "apr_upload0001"}, 403),
|
|
]
|
|
for index, (body, expected) in enumerate(cases):
|
|
status, _, _ = call(router, "POST", base + "/multimodal/items", body, {"If-Match": "0", "Idempotency-Key": f"approval-bad-{index}"})
|
|
assert status == expected
|
|
camera = {**common, "source": "camera", "approval_id": "apr_capture0001"}
|
|
assert call(router, "POST", base + "/multimodal/items", camera, {"If-Match": "0", "Idempotency-Key": "approval-good-1"})[0] == 201
|
|
other_conv = call(router, "POST", "/hux/v1/conversations", {"title": "Other", "project_id": project["id"]})[1]
|
|
approval(router, other_conv["id"], name="apr_other00001")
|
|
assert call(router, "POST", base + "/multimodal/items", {**common, "approval_id": "apr_other00001"}, {"If-Match": "0", "Idempotency-Key": "approval-bad-other"})[0] == 403
|
|
|
|
|
|
def test_parent_and_artifact_lineage_are_scope_checked(setup):
|
|
router, project, _, base = setup
|
|
parent = create_item(setup, key="lineage-parent-1")[1]
|
|
status, child, _ = create_item(setup, key="lineage-child-01", lineage={"parent_item_id": parent["id"]})
|
|
assert status == 201 and child["lineage"] == {"parent_item_id": parent["id"]}
|
|
artifact_body = {
|
|
"type": "document", "title": "Notes", "project_id": project["id"], "conversation_id": base.split("/")[-1],
|
|
"content": "plain", "mime": "text/plain",
|
|
}
|
|
artifact = call(router, "POST", "/hux/v1/artifacts", artifact_body, {"Idempotency-Key": "artifact-lineage-1"})[1]
|
|
lineage = {"artifact_id": artifact["id"], "artifact_version": 1}
|
|
status, linked, _ = create_item(setup, key="lineage-artifact1", lineage=lineage)
|
|
assert status == 201 and linked["lineage"] == lineage
|
|
bad = [
|
|
{"parent_item_id": parent["id"], "artifact_id": artifact["id"]},
|
|
{"artifact_id": artifact["id"]},
|
|
{"artifact_id": artifact["id"], "artifact_version": 99},
|
|
{"unknown": "mmi_missing000"},
|
|
]
|
|
for index, value in enumerate(bad):
|
|
assert create_item(setup, key=f"bad-lineage-{index}", lineage=value)[0] in {400, 404}
|
|
html = call(router, "POST", "/hux/v1/artifacts", {**artifact_body, "type": "html", "title": "Unsafe", "content": "<b>x</b>", "mime": "text/html"}, {"Idempotency-Key": "artifact-lineage-2"})[1]
|
|
assert create_item(setup, key="bad-lineage-html", lineage={"artifact_id": html["id"], "artifact_version": 1})[0] == 400
|
|
|
|
|
|
def test_transcript_corrections_are_immutable_and_concurrent(setup):
|
|
router, project, _, base = setup
|
|
item = create_item(setup)[1]
|
|
path = base + f"/multimodal/items/{item['id']}/transcript-corrections"
|
|
body = {"project_id": project["id"], "replacement_text": "What I actually said."}
|
|
status, correction, headers = call(router, "POST", path, body, {"If-Match": "1", "Idempotency-Key": "correct-key-001"})
|
|
assert status == 201 and headers["ETag"] == "2"
|
|
assert contracts.validate("multimodal.schema.json", correction, SCHEMAS, "/$defs/correction") == []
|
|
got = call(router, "GET", base + f"/multimodal/items/{item['id']}")[1]
|
|
assert got["latest_correction_id"] == correction["id"] and got["revision"] == 2
|
|
status, replay, headers = call(router, "POST", path, body, {"If-Match": "1", "Idempotency-Key": "correct-key-001"})
|
|
assert status == 200 and replay == correction and headers["HUX-Replayed"] == "true"
|
|
assert call(router, "POST", path, {**body, "replacement_text": "changed"}, {"If-Match": "2", "Idempotency-Key": "correct-key-001"})[0] == 409
|
|
assert call(router, "POST", path, body, {"If-Match": "1", "Idempotency-Key": "correct-key-002"})[0] == 409
|
|
assert call(router, "POST", path, {**body, "replacement_text": ""}, {"If-Match": "2", "Idempotency-Key": "correct-key-003"})[0] == 400
|
|
image = create_item(setup, key="image-for-correct", kind="image", filename="a.png", mime="image/png")[1]
|
|
image_path = base + f"/multimodal/items/{image['id']}/transcript-corrections"
|
|
assert call(router, "POST", image_path, body, {"If-Match": "1", "Idempotency-Key": "correct-key-004"})[0] == 400
|
|
|
|
|
|
def test_capture_intent_is_inert_redacted_and_idempotent(setup):
|
|
router, project, _, base = setup
|
|
token = "ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
|
|
body = {"project_id": project["id"], "source": "screen", "purpose": f"Show issue {token}"}
|
|
headers = {"If-Match": "0", "Idempotency-Key": "capture-key-001"}
|
|
status, intent, response_headers = call(router, "POST", base + "/capture-intents", body, headers)
|
|
assert status == 201 and intent["status"] == "proposed" and intent["requires_approval"] is True
|
|
assert intent["execution_allowed"] is False and token not in intent["purpose"]
|
|
assert contracts.validate("multimodal.schema.json", intent, SCHEMAS, "/$defs/capture_intent") == []
|
|
status, replay, replay_headers = call(router, "POST", base + "/capture-intents", body, headers)
|
|
assert status == 200 and replay == intent and replay_headers["HUX-Replayed"] == "true"
|
|
for bad in ({**body, "source": "microphone"}, {**body, "purpose": ""}, {**body, "execute": True}):
|
|
assert call(router, "POST", base + "/capture-intents", bad, {"If-Match": "0", "Idempotency-Key": f"bad-cap-{abs(hash(repr(bad)))}"})[0] == 400
|
|
assert response_headers["Cache-Control"] == "no-store"
|
|
|
|
|
|
def test_scope_limits_and_malformed_requests_fail_closed(setup, monkeypatch):
|
|
router, project, _, base = setup
|
|
assert call(router, "GET", base + "/multimodal/items", headers=OTHER)[0] == 404
|
|
wrong = base.replace(project["id"], "prj_missing0000")
|
|
assert call(router, "GET", wrong + "/multimodal/items")[0] == 404
|
|
assert router.dispatch("POST", base + "/multimodal/items", HEADERS, b"[]").status == 400
|
|
body = item_body(project, filename="b.webm")
|
|
assert call(router, "POST", base + "/multimodal/items", body, {"Idempotency-Key": "missing-ifmatch"})[0] == 400
|
|
monkeypatch.setattr(multimodal, "MAX_ITEMS", 0)
|
|
assert create_item(setup, key="over-item-limit", filename="c.webm")[0] == 413
|
|
monkeypatch.setattr(multimodal, "MAX_INTENTS", 0)
|
|
intent = {"project_id": project["id"], "source": "camera", "purpose": "scan"}
|
|
assert call(router, "POST", base + "/capture-intents", intent, {"If-Match": "0", "Idempotency-Key": "over-intent-limit"})[0] == 413
|
|
assert call(router, "GET", base + "/multimodal/items/mmi_missing000")[0] == 404
|
|
|
|
|
|
def test_cross_scope_branches_missing_keys_and_correction_limit(setup, monkeypatch):
|
|
router, project, _, base = setup
|
|
assert call(router, "POST", base + "/multimodal/items", item_body(project), {"If-Match": "0"})[0] == 400
|
|
executable = item_body(project, filename="payload.txt", mime="text/html", kind="document")
|
|
assert call(router, "POST", base + "/multimodal/items", executable, {"If-Match": "0", "Idempotency-Key": "mime-executable1"})[0] == 400
|
|
_, other_conversation, _ = call(router, "POST", "/hux/v1/conversations", {"title": "Other", "project_id": project["id"]})
|
|
other_base = f"/hux/v1/projects/{project['id']}/conversations/{other_conversation['id']}"
|
|
approval(router, other_conversation["id"], name="apr_othermedia1")
|
|
other_body = item_body(project, approval_id="apr_othermedia1")
|
|
other_item = call(router, "POST", other_base + "/multimodal/items", other_body, {"If-Match": "0", "Idempotency-Key": "other-media-key1"})[1]
|
|
assert create_item(setup, key="cross-parent-key1", lineage={"parent_item_id": other_item["id"]})[0] == 404
|
|
artifact = call(router, "POST", "/hux/v1/artifacts", {
|
|
"type": "document", "title": "Other", "project_id": project["id"], "conversation_id": other_conversation["id"],
|
|
"content": "plain", "mime": "text/plain",
|
|
}, {"Idempotency-Key": "other-artifact-key"})[1]
|
|
assert create_item(setup, key="cross-artifact-1", lineage={"artifact_id": artifact["id"], "artifact_version": 1})[0] == 404
|
|
mine = create_item(setup, key="my-cross-media1")[1]
|
|
assert call(router, "GET", other_base + f"/multimodal/items/{mine['id']}")[0] == 404
|
|
correction = {"project_id": project["id"], "replacement_text": "correct"}
|
|
assert call(router, "POST", other_base + f"/multimodal/items/{mine['id']}/transcript-corrections", correction, {"If-Match": "1", "Idempotency-Key": "cross-correct-1"})[0] == 404
|
|
monkeypatch.setattr(multimodal, "MAX_CORRECTIONS", 0)
|
|
assert call(router, "POST", base + f"/multimodal/items/{mine['id']}/transcript-corrections", correction, {"If-Match": "1", "Idempotency-Key": "correct-limit-1"})[0] == 413
|
|
scoped = tenant(router)
|
|
scoped.delete("projects", project["id"])
|
|
assert call(router, "GET", base + "/multimodal/items")[0] == 404
|
|
|
|
|
|
def test_source_contract_and_test_files_stay_bounded():
|
|
for path in (FOUNDATION / "hux" / "multimodal.py", ROOT / "services" / "hermes" / "contracts" / "hux" / "multimodal.schema.json", Path(__file__)):
|
|
assert len(path.read_text().splitlines()) <= 500
|