Codex
82e2c91f42
Publish documentation / publish (push) Successful in 1m27s
Add PostgreSQL-backed memory, editable evidence with source review and activation, and human-approved archive repairs across the harness, API, and UI. Include migrations, deployment support, regression coverage, and validation documentation. Refresh permissions from validated session roles so existing administrator logins can access newly deployed archive management features.
91 lines
4.1 KiB
Python
91 lines
4.1 KiB
Python
"""CLI adaptation: trusted context, verified recall and explicit availability failures."""
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from typer.testing import CliRunner
|
|
|
|
from tht.cli import app
|
|
from tht.memory.models import MemoryUnavailable
|
|
from tht.ports.vector import VectorReadUnavailable, VectorStoreError
|
|
|
|
|
|
@pytest.fixture
|
|
def runtime(monkeypatch):
|
|
service = SimpleNamespace(close=lambda: None, recall=lambda *a, **kw: [])
|
|
monkeypatch.setattr("tht.cli.memory_cmd._load_config_or_exit", lambda path: object())
|
|
# The config is supplied by the runner; mapping to the service is tested here.
|
|
monkeypatch.setattr("tht.cli.memory_cmd._load_config_or_exit",
|
|
lambda path: SimpleNamespace(embeddings=object(),
|
|
database=SimpleNamespace(database="sales", db_schema="public")))
|
|
monkeypatch.setattr("tht.cli.memory_cmd.memory_service", lambda cfg: service)
|
|
monkeypatch.setattr("tht.cli.vector_cmd.open_searcher", lambda cfg: object())
|
|
monkeypatch.setattr("tht.cli.vector_cmd.make_embedder", lambda cfg: object())
|
|
return service
|
|
|
|
|
|
@pytest.mark.parametrize("failure", [VectorStoreError, VectorReadUnavailable])
|
|
def test_solved_search_warns_when_vector_is_unavailable(runtime, monkeypatch, failure):
|
|
def boom(cfg):
|
|
raise failure("private adapter details")
|
|
monkeypatch.setattr("tht.cli.vector_cmd.open_searcher", boom)
|
|
result = CliRunner().invoke(app, ["memory", "solved-search", "orders", "--json"])
|
|
assert result.exit_code == 0
|
|
assert json.loads(result.stdout) == []
|
|
assert "exemplar non disponibili" in result.stderr
|
|
assert "private adapter details" not in result.output
|
|
|
|
|
|
def test_solved_search_passes_only_verified_service_payload(runtime):
|
|
def recall(question, **kwargs):
|
|
assert question == "orders"
|
|
assert kwargs["solved"] and kwargs["top"] == 3
|
|
assert kwargs["scope"].database == "sales"
|
|
assert kwargs["scope"].schema_name == "public"
|
|
return [{"id": "mem-id", "question": "Orders?", "sql": "select 1", "tables": []}]
|
|
runtime.recall = recall
|
|
result = CliRunner().invoke(app, ["memory", "solved-search", "orders", "--json"])
|
|
assert result.exit_code == 0
|
|
assert json.loads(result.stdout)[0]["sql"] == "select 1"
|
|
|
|
|
|
def test_archive_failure_is_not_an_empty_success(runtime):
|
|
def recall(*args, **kwargs):
|
|
raise MemoryUnavailable("Memory archive is unavailable")
|
|
runtime.recall = recall
|
|
result = CliRunner().invoke(app, ["memory", "solved-search", "orders", "--json"])
|
|
assert result.exit_code == 1
|
|
assert json.loads(result.stdout)["status"] == 503
|
|
|
|
|
|
def test_missing_principal_cannot_bypass_admin(monkeypatch, tmp_path):
|
|
monkeypatch.delenv("THT_PRINCIPAL_ISSUER", raising=False)
|
|
monkeypatch.delenv("THT_PRINCIPAL_SUBJECT", raising=False)
|
|
request = tmp_path / "request.json"
|
|
request.write_text(json.dumps({"action": "list", "runtime": {}}))
|
|
result = CliRunner().invoke(app, ["memory", "admin", "--workspace", "sales", "-c", str(request)])
|
|
assert result.exit_code == 1
|
|
assert json.loads(result.stdout)["status"] == 403
|
|
|
|
|
|
@pytest.mark.parametrize("command", ["search", "solved-search"])
|
|
def test_recall_cannot_override_runtime_database_context(runtime, command):
|
|
result = CliRunner().invoke(app, ["memory", command, "orders", "--json",
|
|
"--filters", '{"database":"outside"}'])
|
|
assert result.exit_code == 1
|
|
assert json.loads(result.stdout)["status"] == 400
|
|
|
|
|
|
@pytest.mark.parametrize("command", ["search", "solved-search"])
|
|
def test_recall_passes_explicit_business_and_table_filters(runtime, command):
|
|
def recall(question, **kwargs):
|
|
scope = kwargs["scope"]
|
|
assert (scope.database, scope.schema_name, scope.table, scope.scope) == (
|
|
"sales", "public", "orders", "Sales")
|
|
assert scope.concepts == ["grain"]
|
|
return []
|
|
runtime.recall = recall
|
|
result = CliRunner().invoke(app, ["memory", command, "orders", "--json", "--filters",
|
|
'{"table":"orders","scope":"Sales","concepts":["grain"]}'])
|
|
assert result.exit_code == 0, result.output
|