199 lines
11 KiB
Python
199 lines
11 KiB
Python
"""Focused HUX-09 server-authoritative suggestion tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
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, privacy, store, suggestions
|
|
from hux.errors import Invalid
|
|
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"])
|
|
START = datetime(2026, 8, 24, 12, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
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):
|
|
return store.TenantStore(router.data_root, identity.resolve(HEADERS, {"HUX_ROUTER_KEY": "rk"}))
|
|
|
|
|
|
@pytest.fixture
|
|
def setup(tmp_path, monkeypatch):
|
|
moment = {"value": START}
|
|
monkeypatch.setattr(suggestions, "clock", lambda: moment["value"])
|
|
monkeypatch.setattr(suggestions, "now_iso", lambda: moment["value"].strftime("%Y-%m-%dT%H:%M:%SZ"))
|
|
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']}/suggestions"
|
|
return router, project, conversation, base, moment
|
|
|
|
|
|
def evaluate(setup, context="first_session", key="suggest-key-001", **extra):
|
|
router, _, _, base, _ = setup
|
|
return call(router, "POST", base + "/evaluate", {"context": context, **extra}, {"Idempotency-Key": key})
|
|
|
|
|
|
def test_every_context_returns_a_canonical_server_definition(setup):
|
|
for index, context in enumerate(suggestions.CONTEXTS):
|
|
status, result, headers = evaluate(setup, context, f"context-key-{index:02}")
|
|
assert status == 200 and result["stored"] is True and headers["Cache-Control"] == "no-store"
|
|
suggestion = result["suggestion"]
|
|
assert suggestion["trigger"] == {"surface": "chat", "context": context}
|
|
assert suggestion["suppression"] == {"dismissable": True, "max_shows": 3, "cooldown_seconds": 86400, "never_again_supported": True}
|
|
assert contracts.validate("suggestion.schema.json", suggestion, pointer="/$defs/suggestion") == []
|
|
assert contracts.validate("suggestion.schema.json", result["state"], pointer="/$defs/state") == []
|
|
|
|
|
|
def test_idempotency_cooldown_and_max_show_are_authoritative(setup):
|
|
_, _, _, _, moment = setup
|
|
status, first, _ = evaluate(setup)
|
|
assert status == 200 and first["state"]["shows"] == 1 and first["revision"] == 1
|
|
status, replay, headers = evaluate(setup)
|
|
assert status == 200 and replay == first and headers["HUX-Replayed"] == "true"
|
|
assert evaluate(setup, context="idle")[0] == 409
|
|
status, cooled, _ = evaluate(setup, key="suggest-key-002")
|
|
assert status == 200 and cooled == {"suggestion": None, "reason": "cooldown", "stored": False}
|
|
moment["value"] += timedelta(days=1, seconds=1)
|
|
second = evaluate(setup, key="suggest-key-003")[1]
|
|
assert second["state"]["shows"] == 2 and second["revision"] == 2
|
|
moment["value"] += timedelta(days=1, seconds=1)
|
|
third = evaluate(setup, key="suggest-key-004")[1]
|
|
assert third["state"]["shows"] == 3
|
|
moment["value"] += timedelta(days=1, seconds=1)
|
|
exhausted = evaluate(setup, key="suggest-key-005")[1]
|
|
assert exhausted == {"suggestion": None, "reason": "max_shows", "stored": False}
|
|
|
|
|
|
def test_decisions_require_click_if_match_and_idempotency(setup):
|
|
_, _, _, base, moment = setup
|
|
shown = evaluate(setup)[1]
|
|
suggestion_id = shown["suggestion"]["id"]
|
|
path = base + f"/{suggestion_id}/decisions"
|
|
headers = {"If-Match": "1", "Idempotency-Key": "decision-key-001"}
|
|
status, result, response_headers = call(setup[0], "POST", path, {"decision": "dismissed", "clicked": True}, headers)
|
|
assert status == 200 and result["decision"] == "dismissed" and result["revision"] == 2
|
|
assert response_headers["ETag"] == "2" and "dismissed_at" in result["state"]
|
|
status, replay, replay_headers = call(setup[0], "POST", path, {"decision": "dismissed", "clicked": True}, headers)
|
|
assert status == 200 and replay == result and replay_headers["HUX-Replayed"] == "true"
|
|
assert call(setup[0], "POST", path, {"decision": "acted", "clicked": True}, headers)[0] == 409
|
|
moment["value"] += timedelta(days=1, seconds=1)
|
|
shown_again = evaluate(setup, key="suggest-key-after-dismiss")[1]
|
|
assert shown_again["suggestion"]["id"] == suggestion_id and shown_again["state"]["shows"] == 2
|
|
invalid = [
|
|
({"decision": "acted", "clicked": False}, {"If-Match": "3", "Idempotency-Key": "decision-key-002"}),
|
|
({"decision": "silent", "clicked": True}, {"If-Match": "3", "Idempotency-Key": "decision-key-003"}),
|
|
({"decision": "acted", "clicked": True}, {"Idempotency-Key": "decision-key-004"}),
|
|
({"decision": "acted", "clicked": True}, {"If-Match": "3"}),
|
|
]
|
|
for body, bad_headers in invalid:
|
|
assert call(setup[0], "POST", path, body, bad_headers)[0] == 400
|
|
|
|
|
|
def test_never_again_is_explicit_and_permanent(setup):
|
|
_, _, _, base, moment = setup
|
|
shown = evaluate(setup)[1]
|
|
path = base + f"/{shown['suggestion']['id']}/decisions"
|
|
status, result, _ = call(setup[0], "POST", path, {"decision": "never_again", "clicked": True}, {"If-Match": "1", "Idempotency-Key": "never-key-0001"})
|
|
assert status == 200 and result["state"]["never_again"] is True
|
|
moment["value"] += timedelta(days=365)
|
|
blocked = evaluate(setup, key="suggest-key-never")[1]
|
|
assert blocked == {"suggestion": None, "reason": "never_again", "stored": False}
|
|
|
|
|
|
@pytest.mark.parametrize(("gate", "reason"), [
|
|
("no_store", "client_no_store"), ("sensitive", "client_no_store"), ("restricted", "client_no_store"),
|
|
])
|
|
def test_client_privacy_gates_do_not_write_state(setup, gate, reason):
|
|
body = {"no_store": True} if gate == "no_store" else {"sensitivity": gate}
|
|
status, result, _ = evaluate(setup, key=f"privacy-{gate}-1", **body)
|
|
assert status == 200 and result == {"suggestion": None, "reason": reason, "stored": False}
|
|
assert tenant(setup[0]).count(suggestions.FAMILY) == 0
|
|
assert tenant(setup[0]).read(suggestions.FAMILY, "idempotency") == []
|
|
|
|
|
|
def test_server_privacy_state_overrides_client_claims(setup):
|
|
router, _, conversation, base, _ = setup
|
|
privacy.mark_topic(tenant(router), conversation["id"], "health")
|
|
result = evaluate(setup, key="privacy-topic-01", sensitivity="public")[1]
|
|
assert result["reason"] == "sensitive_topic" and tenant(router).count(suggestions.FAMILY) == 0
|
|
privacy.set_flag(tenant(router), conversation["id"], "memory_disabled", True)
|
|
assert evaluate(setup, key="privacy-disabled", sensitivity="public")[1]["reason"] == "conversation_no_store"
|
|
privacy.set_flag(tenant(router), conversation["id"], "forgotten", True)
|
|
assert evaluate(setup, key="privacy-forgotten", sensitivity="public")[1]["reason"] == "conversation_no_store"
|
|
privacy.set_flag(tenant(router), conversation["id"], "forgotten", False)
|
|
privacy.set_flag(tenant(router), conversation["id"], "memory_disabled", False)
|
|
doc = tenant(router).get("privacy", privacy.TOPICS_DOC)
|
|
doc["items"][conversation["id"]]["topics"] = []
|
|
tenant(router).put("privacy", doc)
|
|
current = tenant(router).get("conversations", conversation["id"])
|
|
tenant(router).put("conversations", {**current, "mode": "private"}, current["revision"])
|
|
result = call(router, "POST", base + "/evaluate", {"context": "idle", "sensitivity": "public"}, {"Idempotency-Key": "privacy-private1"})[1]
|
|
assert result["reason"] == "private_mode"
|
|
|
|
|
|
def test_list_state_scope_and_limits(setup, monkeypatch):
|
|
router, project, _, base, _ = setup
|
|
shown = evaluate(setup)[1]
|
|
status, states, headers = call(router, "GET", base + "/states")
|
|
assert status == 200 and states["items"][0]["suggestion_id"] == shown["suggestion"]["id"]
|
|
assert states["items"][0]["revision"] == 1 and headers["Cache-Control"] == "no-store"
|
|
assert call(router, "GET", base + "/states", headers=OTHER)[0] == 404
|
|
wrong = base.replace(project["id"], "prj_missing0000")
|
|
assert call(router, "GET", wrong + "/states")[0] == 404
|
|
_, conversation2, _ = call(router, "POST", "/hux/v1/conversations", {"title": "C2", "project_id": project["id"]})
|
|
base2 = f"/hux/v1/projects/{project['id']}/conversations/{conversation2['id']}/suggestions"
|
|
monkeypatch.setattr(suggestions, "MAX_STATES", 0)
|
|
result = call(router, "POST", base2 + "/evaluate", {"context": "idle"}, {"Idempotency-Key": "state-limit-key"})[1]
|
|
assert result == {"suggestion": None, "reason": "state_limit", "stored": False}
|
|
|
|
|
|
def test_bad_context_body_and_unknown_decision_fail_closed(setup):
|
|
router, _, _, base, _ = setup
|
|
for body in ([], {"context": "nope"}, {"context": "idle", "priority": 100}):
|
|
raw = json.dumps(body).encode()
|
|
assert router.dispatch("POST", base + "/evaluate", {**HEADERS, "Idempotency-Key": "bad-context-key"}, raw).status == 400
|
|
assert call(router, "POST", base + "/evaluate", {"context": "idle"})[0] == 400
|
|
missing = base + "/sug_missing0000/decisions"
|
|
assert call(router, "POST", missing, {"decision": "acted", "clicked": True}, {"If-Match": "1", "Idempotency-Key": "missing-decision"})[0] == 404
|
|
|
|
|
|
def test_defensive_validation_stale_act_and_private_inspection(setup):
|
|
router, _, conversation, base, _ = setup
|
|
with pytest.raises(Invalid, match="suggestion failed"):
|
|
suggestions._suggestion("idle", "invalid-surface")
|
|
with pytest.raises(Invalid, match="state failed"):
|
|
suggestions._public({"schema": "hux.suggestion_state.v1", "owner": "raw-user", "suggestion_id": "sug_idle_workflow", "shows": 1, "never_again": False})
|
|
shown = evaluate(setup)[1]
|
|
path = base + f"/{shown['suggestion']['id']}/decisions"
|
|
assert call(router, "POST", path, {"decision": "acted", "clicked": True}, {"If-Match": "9", "Idempotency-Key": "stale-decision1"})[0] == 409
|
|
status, acted, _ = call(router, "POST", path, {"decision": "acted", "clicked": True}, {"If-Match": "1", "Idempotency-Key": "acted-decision1"})
|
|
assert status == 200 and "acted_at" in acted["state"] and "dismissed_at" not in acted["state"]
|
|
current = tenant(router).get("conversations", conversation["id"])
|
|
tenant(router).put("conversations", {**current, "mode": "private"}, current["revision"])
|
|
assert call(router, "POST", path, {"decision": "dismissed", "clicked": True}, {"If-Match": "2", "Idempotency-Key": "private-decision"})[0] == 404
|
|
status, states, _ = call(router, "GET", base + "/states")
|
|
assert status == 200 and states == {"items": [], "next": None}
|
|
|
|
|
|
def test_source_and_test_files_stay_bounded():
|
|
assert len((FOUNDATION / "hux" / "suggestions.py").read_text().splitlines()) <= 500
|
|
assert len(Path(__file__).read_text().splitlines()) <= 500
|