William Valentin
135 lines
5.0 KiB
Python
135 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
import gateway
|
|
|
|
|
|
def test_authority_envelope_is_advisory_and_forbids_side_effects() -> None:
|
|
env = gateway.build_envelope(
|
|
service="classifier",
|
|
operation="classify",
|
|
mode="shadow",
|
|
result={"labels": {"workflow_category": {"value": "devops"}}},
|
|
npu_busy_delta_us=123,
|
|
input_scope="explicit_text",
|
|
)
|
|
|
|
assert env["ok"] is True
|
|
assert env["mode"] == "shadow"
|
|
assert env["authority"] == {
|
|
"may_route": False,
|
|
"may_write_memory": False,
|
|
"may_send_external": False,
|
|
"may_process_private_dirs": False,
|
|
"may_execute_tools": False,
|
|
"may_restart_services": False,
|
|
}
|
|
assert env["npu_proof"] == {"required": True, "ok": True, "npu_busy_delta_us": 123}
|
|
|
|
|
|
def test_classify_calls_sidecar_and_logs_metadata_only(tmp_path: Path) -> None:
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_post(url: str, payload: dict, timeout_s: float) -> dict:
|
|
calls.append((url, payload))
|
|
return {
|
|
"labels": {"tool_needed": {"value": True}},
|
|
"npu_busy_delta_us": 55,
|
|
"sysfs_npu_busy_delta_us": 55,
|
|
}
|
|
|
|
logger = gateway.AdvisoryLogger(tmp_path / "events.sqlite")
|
|
env = gateway.classify_text(
|
|
"Inspect live service status",
|
|
trace_id="t1",
|
|
http_post_json=fake_post,
|
|
logger=logger,
|
|
)
|
|
|
|
assert calls[0][0].endswith(":18819/v1/classify")
|
|
assert calls[0][1]["options"]["dry_run"] is True
|
|
assert env["service"] == "classifier"
|
|
assert env["authority"]["may_route"] is False
|
|
assert env["npu_proof"]["ok"] is True
|
|
|
|
with sqlite3.connect(tmp_path / "events.sqlite") as con:
|
|
row = con.execute("select service, operation, input_ref, raw_payload from advisory_events").fetchone()
|
|
assert row == ("classifier", "classify", "text:sha256:" + gateway.sha256_text("Inspect live service status"), None)
|
|
|
|
|
|
def test_generate_allows_only_bounded_jobs() -> None:
|
|
with pytest.raises(ValueError, match="unsupported advisory generation job"):
|
|
gateway.generate_bounded("primary_chat", "hello", http_post_json=lambda *_: {})
|
|
|
|
|
|
def test_generate_wraps_draft_without_final_authority() -> None:
|
|
def fake_post(url: str, payload: dict, timeout_s: float) -> dict:
|
|
return {"text": "Short title", "npu_busy_delta_us": 99, "timing_ms": {"total": 10}}
|
|
|
|
env = gateway.generate_bounded("title", "Summarize this local health check", http_post_json=fake_post)
|
|
|
|
assert env["service"] == "genai"
|
|
assert env["operation"] == "generate:title"
|
|
assert env["result"]["draft_text"] == "Short title"
|
|
assert env["result"]["final_authority"] is False
|
|
assert env["authority"]["may_send_external"] is False
|
|
|
|
|
|
def test_doc_triage_requires_explicit_file_under_allowed_root(tmp_path: Path) -> None:
|
|
allowed = tmp_path / "allowed"
|
|
allowed.mkdir()
|
|
target = allowed / "synthetic.png"
|
|
target.write_bytes(b"not real image for unit test")
|
|
|
|
def fake_post(url: str, payload: dict, timeout_s: float) -> dict:
|
|
assert payload["path"] == str(target.resolve())
|
|
assert payload["options"]["allowed_roots"] == [str(allowed.resolve())]
|
|
return {"ok": True, "result": {"pages": [{"needs_attention": {"embedding": {"verified_npu": True, "npu_busy_delta_us": 42}}}]}}
|
|
|
|
env = gateway.triage_file(str(target), allowed_roots=[str(allowed)], configured_roots=[allowed], http_post_json=fake_post)
|
|
|
|
assert env["service"] == "doc_triage"
|
|
assert env["input_scope"] == "explicit_file"
|
|
assert env["npu_proof"]["ok"] is True
|
|
|
|
|
|
def test_doc_triage_rejects_private_root_broadening(tmp_path: Path) -> None:
|
|
allowed = tmp_path / "allowed"
|
|
allowed.mkdir()
|
|
with pytest.raises(ValueError, match="path must be inside an allowed root"):
|
|
gateway.triage_file(str(tmp_path / "outside.png"), allowed_roots=[str(allowed)], configured_roots=[allowed], http_post_json=lambda *_: {})
|
|
|
|
|
|
def test_doc_triage_rejects_requested_root_outside_configured_roots(tmp_path: Path) -> None:
|
|
configured = tmp_path / "configured"
|
|
requested = tmp_path / "private"
|
|
requested.mkdir()
|
|
target = requested / "file.png"
|
|
target.write_bytes(b"synthetic")
|
|
|
|
with pytest.raises(ValueError, match="requested allowed root is outside configured roots"):
|
|
gateway.triage_file(
|
|
str(target),
|
|
allowed_roots=[str(requested)],
|
|
configured_roots=[configured],
|
|
http_post_json=lambda *_: {},
|
|
)
|
|
|
|
|
|
def test_health_aggregates_dependencies_without_raw_private_data() -> None:
|
|
def fake_get(url: str, timeout_s: float) -> dict:
|
|
return {"ok": True, "service": url.rsplit(":", 1)[-1]}
|
|
|
|
health = gateway.health(http_get_json=fake_get)
|
|
|
|
assert health["ok"] is True
|
|
assert set(health["dependencies"]) == {"classifier", "genai", "doc_triage"}
|
|
assert "raw" not in json.dumps(health).lower()
|