6483c7b164
Some checks are pending
CI / lint-and-test (push) Waiting to run
* Add protected canonical claims read API * Normalize claim loader exits in worker thread * Allow file-gated claims reader defaults
540 lines
19 KiB
Python
540 lines
19 KiB
Python
"""Tests for canonical KB claim routes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
pytest.importorskip("aiohttp")
|
|
|
|
from aiohttp import web
|
|
from aiohttp.test_utils import make_mocked_request
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO_ROOT / "diagnostics"))
|
|
|
|
import app as argus_app # noqa: E402
|
|
import kb_claim_routes as routes # noqa: E402
|
|
|
|
CLAIM_ID = "d0000000-0000-0000-0000-000000000005"
|
|
CONNECTED_ID = "e0000000-0000-0000-0000-000000000006"
|
|
SECOND_CLAIM_ID = "c0000000-0000-0000-0000-000000000004"
|
|
THIRD_CLAIM_ID = "b0000000-0000-0000-0000-000000000003"
|
|
API_KEY = "test-argus-api-key"
|
|
|
|
|
|
def _claim_list_app() -> web.Application:
|
|
app = web.Application()
|
|
app[routes.KB_CLAIM_LIST_API_KEY] = API_KEY
|
|
return app
|
|
|
|
|
|
def _claim_list_request(path: str, *, app: web.Application):
|
|
return make_mocked_request("GET", path, app=app, headers={"X-Api-Key": API_KEY})
|
|
|
|
|
|
def _claim_data():
|
|
return {
|
|
"claim": {
|
|
"id": CLAIM_ID,
|
|
"type": "strategy_kernel",
|
|
"text": "Claims should be falsifiable enough to support review.",
|
|
"status": "open",
|
|
"confidence": "0.72",
|
|
"tags": ["claims", "review"],
|
|
"created_at": "2026-07-09T00:00:00Z",
|
|
"updated_at": "2026-07-09T00:00:00Z",
|
|
},
|
|
"evidence": [
|
|
{
|
|
"role": "grounds",
|
|
"weight": "0.8",
|
|
"source_type": "telegram",
|
|
"url": "https://example.test/source",
|
|
"storage_path": None,
|
|
"excerpt": "m3taversal asked Leo to split a broad claim into atomic claims.",
|
|
}
|
|
],
|
|
"edges": [
|
|
{
|
|
"direction": "outgoing",
|
|
"edge_type": "supports",
|
|
"connected_id": CONNECTED_ID,
|
|
"connected_text": "Reviewable claims need supporting evidence and edges.",
|
|
"connected_status": "open",
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def _claim_summary(claim_id: str, *, timestamp: str, text: str = "Canonical claim"):
|
|
return {
|
|
"id": claim_id,
|
|
"type": "strategy_kernel",
|
|
"text": text,
|
|
"text_truncated": False,
|
|
"status": "open",
|
|
"confidence": "0.72",
|
|
"tags": ["claims", "review"],
|
|
"created_at": timestamp,
|
|
"updated_at": timestamp,
|
|
"evidence_count": 3,
|
|
"edge_count": 5,
|
|
"_cursor_timestamp": timestamp,
|
|
"internal_secret": "must-not-leak",
|
|
}
|
|
|
|
|
|
def test_claim_link_html_links_valid_claim_ids() -> None:
|
|
html = routes.claim_link_html(CLAIM_ID)
|
|
|
|
assert f'href="/kb/claims/{CLAIM_ID}"' in html
|
|
assert f"<code>{CLAIM_ID}</code>" in html
|
|
|
|
|
|
def test_render_kb_claim_page_shows_body_evidence_and_linked_edges() -> None:
|
|
html = routes.render_kb_claim_page(_claim_data())
|
|
|
|
assert "Claims should be falsifiable enough to support review." in html
|
|
assert "m3taversal asked Leo" in html
|
|
assert f'href="/kb/claims/{CONNECTED_ID}"' in html
|
|
assert "<code>supports</code>" in html
|
|
assert "<code>open</code>" in html
|
|
|
|
|
|
def test_api_kb_claim_uses_loader_and_returns_claim_graph() -> None:
|
|
app = web.Application()
|
|
app[routes.KB_CLAIM_LOADER_KEY] = lambda claim_id: _claim_data() if claim_id == CLAIM_ID else None
|
|
req = make_mocked_request(
|
|
"GET",
|
|
f"/api/kb/claims/{CLAIM_ID}",
|
|
app=app,
|
|
match_info={"claim_id": CLAIM_ID},
|
|
)
|
|
|
|
response = asyncio.run(routes.handle_api_kb_claim(req))
|
|
data = json.loads(response.body.decode())
|
|
|
|
assert response.status == 200
|
|
assert data["claim"]["id"] == CLAIM_ID
|
|
assert data["edges"][0]["connected_id"] == CONNECTED_ID
|
|
assert response.headers["Cache-Control"] == "private, no-store, max-age=0"
|
|
|
|
|
|
def test_claim_cursor_round_trip() -> None:
|
|
timestamp = "2026-07-13T21:12:00+00:00"
|
|
cursor = routes.encode_claim_cursor(timestamp, CLAIM_ID)
|
|
|
|
assert routes.decode_claim_cursor(cursor) == (timestamp, CLAIM_ID)
|
|
|
|
|
|
def test_canonical_claim_list_database_role_fails_closed(monkeypatch, tmp_path) -> None:
|
|
monkeypatch.delenv("KB_CLAIM_BROWSER_ROLE", raising=False)
|
|
monkeypatch.delenv("KB_CLAIM_BROWSER_SECRETS_FILE", raising=False)
|
|
default_secrets_file = tmp_path / "missing-kb-observatory-read-password"
|
|
monkeypatch.setattr(routes, "CLAIM_LIST_DEFAULT_SECRETS_FILE", str(default_secrets_file))
|
|
|
|
with pytest.raises(RuntimeError, match="read role is not configured"):
|
|
routes._claim_list_db_args()
|
|
|
|
default_secrets_file.write_text("PGPASSWORD=not-printed\n", encoding="utf-8")
|
|
default_secrets_file.chmod(0o600)
|
|
default_args = routes._claim_list_db_args()
|
|
|
|
assert default_args.role == routes.CLAIM_LIST_REQUIRED_ROLE
|
|
assert default_args.secrets_file == str(default_secrets_file)
|
|
|
|
monkeypatch.setenv("KB_CLAIM_BROWSER_ROLE", "kb_apply")
|
|
monkeypatch.setenv("KB_CLAIM_BROWSER_SECRETS_FILE", "/run/secrets/read-password")
|
|
with pytest.raises(RuntimeError, match=routes.CLAIM_LIST_REQUIRED_ROLE):
|
|
routes._claim_list_db_args()
|
|
|
|
monkeypatch.setenv("KB_CLAIM_BROWSER_ROLE", routes.CLAIM_LIST_REQUIRED_ROLE)
|
|
args = routes._claim_list_db_args()
|
|
|
|
assert args.role == routes.CLAIM_LIST_REQUIRED_ROLE
|
|
assert args.secrets_file == "/run/secrets/read-password"
|
|
|
|
|
|
def test_canonical_claim_list_database_session_must_be_read_only(monkeypatch, tmp_path) -> None:
|
|
secrets_file = tmp_path / "kb-observatory-read.env"
|
|
secrets_file.write_text("PGPASSWORD=not-printed\n", encoding="utf-8")
|
|
secrets_file.chmod(0o640)
|
|
monkeypatch.setenv("KB_CLAIM_BROWSER_ROLE", routes.CLAIM_LIST_REQUIRED_ROLE)
|
|
monkeypatch.setenv("KB_CLAIM_BROWSER_SECRETS_FILE", str(secrets_file))
|
|
observed_sql = {}
|
|
|
|
def run_psql(_args, sql, _password):
|
|
observed_sql["value"] = sql
|
|
return json.dumps({"session_read_only": "off", "total": 0, "rows": []})
|
|
|
|
monkeypatch.setattr(routes, "_run_claim_list_psql", run_psql)
|
|
filters = {
|
|
"q": "",
|
|
"status": "all",
|
|
"type": "",
|
|
"tag": "",
|
|
"limit": 25,
|
|
"cursor": "",
|
|
"cursor_values": None,
|
|
}
|
|
|
|
with pytest.raises(RuntimeError, match="session is not read-only"):
|
|
routes._load_claim_list_from_db(filters)
|
|
|
|
assert "current_setting('transaction_read_only')" in observed_sql["value"]
|
|
assert "set statement_timeout = '5000ms'" in observed_sql["value"]
|
|
assert observed_sql["value"].index("page_rows as") < observed_sql["value"].index(
|
|
"from public.claim_evidence"
|
|
)
|
|
|
|
|
|
def test_canonical_claim_list_secrets_file_accepts_only_pgpassword(tmp_path) -> None:
|
|
secrets_file = tmp_path / "kb-observatory-read.env"
|
|
secrets_file.write_text("# dedicated read role\nPGPASSWORD='read-only-secret'\n", encoding="utf-8")
|
|
secrets_file.chmod(0o640)
|
|
|
|
assert routes._load_claim_list_password(str(secrets_file)) == "read-only-secret"
|
|
|
|
secrets_file.write_text("KB_APPLY_PASSWORD=wrong-boundary\n", encoding="utf-8")
|
|
with pytest.raises(RuntimeError, match="invalid format"):
|
|
routes._load_claim_list_password(str(secrets_file))
|
|
|
|
secrets_file.write_text("PGPASSWORD=secret\n", encoding="utf-8")
|
|
secrets_file.chmod(0o644)
|
|
with pytest.raises(RuntimeError, match="world-accessible"):
|
|
routes._load_claim_list_password(str(secrets_file))
|
|
|
|
|
|
def test_canonical_claim_list_process_timeout_is_enforced_and_sanitized(monkeypatch) -> None:
|
|
args = routes.argparse.Namespace(
|
|
container="postgres",
|
|
db="teleo",
|
|
host="127.0.0.1",
|
|
role=routes.CLAIM_LIST_REQUIRED_ROLE,
|
|
)
|
|
observed = {}
|
|
|
|
def timeout_run(command, **kwargs):
|
|
observed["command"] = command
|
|
observed.update(kwargs)
|
|
raise routes.subprocess.TimeoutExpired(command, kwargs["timeout"], stderr="secret")
|
|
|
|
monkeypatch.setattr(routes.subprocess, "run", timeout_run)
|
|
|
|
with pytest.raises(TimeoutError, match="database query timed out") as exc_info:
|
|
routes._run_claim_list_psql(args, "select 1", "must-not-leak")
|
|
|
|
assert "secret" not in str(exc_info.value)
|
|
assert observed["timeout"] == routes.CLAIM_LIST_PROCESS_TIMEOUT_SECONDS
|
|
assert f"PGCONNECT_TIMEOUT={routes.CLAIM_LIST_CONNECT_TIMEOUT_SECONDS}" in observed[
|
|
"command"
|
|
]
|
|
|
|
|
|
def test_canonical_claim_list_api_key_loads_only_from_dedicated_file(
|
|
monkeypatch, tmp_path
|
|
) -> None:
|
|
api_key_file = tmp_path / "kb-observatory-api-key"
|
|
api_key_file.write_text("route-scoped-observatory-key\n", encoding="utf-8")
|
|
api_key_file.chmod(0o640)
|
|
monkeypatch.setenv("KB_CLAIM_BROWSER_API_KEY_FILE", str(api_key_file))
|
|
|
|
assert routes.load_claim_list_api_key() == "route-scoped-observatory-key"
|
|
|
|
api_key_file.write_text("two tokens\n", encoding="utf-8")
|
|
assert routes.load_claim_list_api_key() is None
|
|
|
|
api_key_file.write_text("é" * 24, encoding="utf-8")
|
|
assert routes.load_claim_list_api_key() is None
|
|
|
|
api_key_file.write_text("route-scoped-observatory-key\n", encoding="utf-8")
|
|
api_key_file.chmod(0o644)
|
|
assert routes.load_claim_list_api_key() is None
|
|
|
|
|
|
def test_canonical_claim_list_is_versioned_paginated_and_sanitized() -> None:
|
|
app = _claim_list_app()
|
|
captured_filters = {}
|
|
|
|
def loader(filters):
|
|
captured_filters.update(filters)
|
|
return {
|
|
"total": 1837,
|
|
"rows": [
|
|
_claim_summary(CLAIM_ID, timestamp="2026-07-13T21:12:00+00:00"),
|
|
_claim_summary(SECOND_CLAIM_ID, timestamp="2026-07-13T20:12:00+00:00"),
|
|
],
|
|
}
|
|
|
|
app[routes.KB_CLAIM_LIST_LOADER_KEY] = loader
|
|
req = _claim_list_request(
|
|
"/api/kb/claims?limit=1&q=falsifiable&type=strategy_kernel&tag=review",
|
|
app=app,
|
|
)
|
|
|
|
response = asyncio.run(routes.handle_api_kb_claims(req))
|
|
data = json.loads(response.body.decode())
|
|
|
|
assert response.status == 200
|
|
assert response.headers["Cache-Control"] == "private, no-store, max-age=0"
|
|
assert response.headers["Vary"] == "X-Api-Key"
|
|
assert data["schema"] == routes.CLAIM_LIST_SCHEMA
|
|
assert data["read_only"] is True
|
|
assert data["source"]["store"] == "canonical_postgres"
|
|
assert data["page"] == {
|
|
"limit": 1,
|
|
"returned": 1,
|
|
"total": 1837,
|
|
"has_more": True,
|
|
"next_cursor": data["page"]["next_cursor"],
|
|
}
|
|
assert routes.decode_claim_cursor(data["page"]["next_cursor"])[1] == CLAIM_ID
|
|
assert data["claims"][0]["id"] == CLAIM_ID
|
|
assert "internal_secret" not in data["claims"][0]
|
|
assert captured_filters["q"] == "falsifiable"
|
|
assert captured_filters["type"] == "strategy_kernel"
|
|
assert captured_filters["tag"] == "review"
|
|
|
|
|
|
def test_canonical_claim_list_caps_response_bytes_without_skipping_rows(monkeypatch) -> None:
|
|
app = _claim_list_app()
|
|
rows = [
|
|
_claim_summary(CLAIM_ID, timestamp="2026-07-13T21:12:00+00:00", text="🔥" * 1200),
|
|
_claim_summary(
|
|
SECOND_CLAIM_ID,
|
|
timestamp="2026-07-13T20:12:00+00:00",
|
|
text="🔥" * 1200,
|
|
),
|
|
_claim_summary(
|
|
THIRD_CLAIM_ID,
|
|
timestamp="2026-07-13T19:12:00+00:00",
|
|
text="🔥" * 1200,
|
|
),
|
|
]
|
|
for row in rows:
|
|
row["tags"] = ["🔥" * 128] * 32
|
|
app[routes.KB_CLAIM_LIST_LOADER_KEY] = lambda _filters: {"total": 3, "rows": rows}
|
|
assert routes.CLAIM_LIST_MAX_RESPONSE_BYTES < 512_000
|
|
byte_limit = 100_000
|
|
monkeypatch.setattr(routes, "CLAIM_LIST_MAX_RESPONSE_BYTES", byte_limit)
|
|
request = _claim_list_request("/api/kb/claims?limit=3&status=all", app=app)
|
|
|
|
response = asyncio.run(routes.handle_api_kb_claims(request))
|
|
data = json.loads(response.body.decode())
|
|
|
|
assert response.status == 200
|
|
assert len(response.body) <= byte_limit
|
|
assert data["page"]["limit"] == 3
|
|
assert data["page"]["returned"] == 1
|
|
assert data["page"]["has_more"] is True
|
|
assert routes.decode_claim_cursor(data["page"]["next_cursor"])[1] == CLAIM_ID
|
|
assert [claim["id"] for claim in data["claims"]] == [CLAIM_ID]
|
|
|
|
|
|
def test_canonical_claim_list_clamps_limit_and_rejects_bad_cursor() -> None:
|
|
app = _claim_list_app()
|
|
observed = {}
|
|
app[routes.KB_CLAIM_LIST_LOADER_KEY] = lambda filters: observed.update(filters) or {"total": 0, "rows": []}
|
|
large_req = _claim_list_request("/api/kb/claims?limit=999", app=app)
|
|
|
|
large_response = asyncio.run(routes.handle_api_kb_claims(large_req))
|
|
large_data = json.loads(large_response.body.decode())
|
|
|
|
assert large_response.status == 200
|
|
assert large_data["page"]["limit"] == routes.CLAIM_LIST_MAX_LIMIT
|
|
assert observed["limit"] == routes.CLAIM_LIST_MAX_LIMIT
|
|
|
|
bad_req = _claim_list_request("/api/kb/claims?cursor=not-a-cursor", app=app)
|
|
bad_response = asyncio.run(routes.handle_api_kb_claims(bad_req))
|
|
bad_data = json.loads(bad_response.body.decode())
|
|
|
|
assert bad_response.status == 400
|
|
assert bad_data == {"error": "invalid cursor"}
|
|
|
|
long_req = _claim_list_request(f"/api/kb/claims?cursor={'x' * 513}", app=app)
|
|
long_response = asyncio.run(routes.handle_api_kb_claims(long_req))
|
|
|
|
assert long_response.status == 400
|
|
|
|
|
|
def test_canonical_claim_list_failure_is_sanitized() -> None:
|
|
app = _claim_list_app()
|
|
|
|
def fail(_filters):
|
|
raise SystemExit("database password would have appeared here")
|
|
|
|
app[routes.KB_CLAIM_LIST_LOADER_KEY] = fail
|
|
req = _claim_list_request("/api/kb/claims", app=app)
|
|
|
|
response = asyncio.run(routes.handle_api_kb_claims(req))
|
|
data = json.loads(response.body.decode())
|
|
|
|
assert response.status == 503
|
|
assert data == {"error": "canonical_claims_unavailable"}
|
|
assert b"password" not in response.body
|
|
|
|
|
|
def test_canonical_claim_auth_boundaries_are_exact() -> None:
|
|
list_path = "/api/kb/claims"
|
|
|
|
assert list_path in routes.KB_CLAIM_SELF_AUTH_PATHS
|
|
assert f"{list_path}/{CLAIM_ID}" not in routes.KB_CLAIM_SELF_AUTH_PATHS
|
|
assert f"/api/kb/claims/{CLAIM_ID}".startswith(routes.KB_CLAIM_GLOBAL_AUTH_PREFIXES)
|
|
assert f"/kb/claims/{CLAIM_ID}".startswith(routes.KB_CLAIM_GLOBAL_AUTH_PREFIXES)
|
|
|
|
|
|
def test_canonical_claim_list_authentication_fails_closed() -> None:
|
|
app_without_key = web.Application()
|
|
unconfigured_req = make_mocked_request("GET", "/api/kb/claims", app=app_without_key)
|
|
unconfigured_response = asyncio.run(routes.handle_api_kb_claims(unconfigured_req))
|
|
|
|
assert unconfigured_response.status == 503
|
|
assert json.loads(unconfigured_response.body.decode()) == {"error": "canonical_claims_auth_unconfigured"}
|
|
|
|
configured_app = _claim_list_app()
|
|
unauthorized_req = make_mocked_request("GET", "/api/kb/claims", app=configured_app)
|
|
unauthorized_response = asyncio.run(routes.handle_api_kb_claims(unauthorized_req))
|
|
|
|
assert unauthorized_response.status == 401
|
|
assert json.loads(unauthorized_response.body.decode()) == {"error": "unauthorized"}
|
|
|
|
non_ascii_request = make_mocked_request(
|
|
"GET",
|
|
"/api/kb/claims",
|
|
app=configured_app,
|
|
headers={"X-Api-Key": "é" * 24},
|
|
)
|
|
non_ascii_response = asyncio.run(routes.handle_api_kb_claims(non_ascii_request))
|
|
|
|
assert non_ascii_response.status == 401
|
|
assert json.loads(non_ascii_response.body.decode()) == {"error": "unauthorized"}
|
|
|
|
|
|
def test_argus_middleware_only_bypasses_global_key_for_exact_list_path() -> None:
|
|
async def ok_handler(_request):
|
|
return web.Response(status=204)
|
|
|
|
app = web.Application()
|
|
list_request = make_mocked_request("GET", "/api/kb/claims", app=app)
|
|
detail_request = make_mocked_request(
|
|
"GET", f"/api/kb/claims/{CLAIM_ID}", app=app
|
|
)
|
|
html_request = make_mocked_request("GET", f"/kb/claims/{CLAIM_ID}", app=app)
|
|
|
|
list_response = asyncio.run(argus_app.auth_middleware(list_request, ok_handler))
|
|
detail_response = asyncio.run(argus_app.auth_middleware(detail_request, ok_handler))
|
|
html_response = asyncio.run(argus_app.auth_middleware(html_request, ok_handler))
|
|
|
|
assert list_response.status == 204
|
|
assert detail_response.status == 503
|
|
assert html_response.status == 503
|
|
|
|
|
|
def test_canonical_claim_list_loader_runs_off_the_event_loop() -> None:
|
|
app = _claim_list_app()
|
|
event_loop_thread = threading.get_ident()
|
|
loader_thread = None
|
|
|
|
def loader(_filters):
|
|
nonlocal loader_thread
|
|
loader_thread = threading.get_ident()
|
|
return {"total": 0, "rows": []}
|
|
|
|
app[routes.KB_CLAIM_LIST_LOADER_KEY] = loader
|
|
request = _claim_list_request("/api/kb/claims", app=app)
|
|
|
|
response = asyncio.run(routes.handle_api_kb_claims(request))
|
|
|
|
assert response.status == 200
|
|
assert loader_thread is not None
|
|
assert loader_thread != event_loop_thread
|
|
|
|
|
|
def test_canonical_claim_list_loader_timeout_is_sanitized(monkeypatch) -> None:
|
|
app = _claim_list_app()
|
|
|
|
def slow_loader(_filters):
|
|
time.sleep(0.03)
|
|
return {"total": 0, "rows": []}
|
|
|
|
app[routes.KB_CLAIM_LIST_LOADER_KEY] = slow_loader
|
|
monkeypatch.setattr(routes, "CLAIM_LIST_HANDLER_TIMEOUT_SECONDS", 0.001)
|
|
request = _claim_list_request("/api/kb/claims", app=app)
|
|
|
|
response = asyncio.run(routes.handle_api_kb_claims(request))
|
|
|
|
assert response.status == 503
|
|
assert json.loads(response.body.decode()) == {"error": "canonical_claims_unavailable"}
|
|
|
|
|
|
def test_canonical_claim_list_rejects_duplicate_or_out_of_order_pages() -> None:
|
|
filters = {
|
|
"q": "",
|
|
"status": "all",
|
|
"type": "",
|
|
"tag": "",
|
|
"limit": 2,
|
|
"cursor": "",
|
|
"cursor_values": None,
|
|
}
|
|
newest = _claim_summary(CLAIM_ID, timestamp="2026-07-13T21:12:00+00:00")
|
|
older = _claim_summary(SECOND_CLAIM_ID, timestamp="2026-07-13T20:12:00+00:00")
|
|
|
|
with pytest.raises(ValueError, match="duplicate ids"):
|
|
routes._claim_list_payload({"total": 2, "rows": [newest, newest]}, filters)
|
|
|
|
with pytest.raises(ValueError, match="strictly ordered"):
|
|
routes._claim_list_payload({"total": 2, "rows": [older, newest]}, filters)
|
|
|
|
|
|
def test_canonical_claim_list_rejects_rows_at_or_before_wrong_cursor_side() -> None:
|
|
cursor_timestamp = "2026-07-13T20:12:00+00:00"
|
|
filters = {
|
|
"q": "",
|
|
"status": "all",
|
|
"type": "",
|
|
"tag": "",
|
|
"limit": 1,
|
|
"cursor": routes.encode_claim_cursor(cursor_timestamp, SECOND_CLAIM_ID),
|
|
"cursor_values": (cursor_timestamp, SECOND_CLAIM_ID),
|
|
}
|
|
newer = _claim_summary(CLAIM_ID, timestamp="2026-07-13T21:12:00+00:00")
|
|
|
|
with pytest.raises(ValueError, match="crossed its request cursor"):
|
|
routes._claim_list_payload({"total": 2, "rows": [newer]}, filters)
|
|
|
|
|
|
def test_legacy_claim_errors_do_not_expose_exception_text() -> None:
|
|
app = web.Application()
|
|
|
|
def fail(_claim_id):
|
|
raise RuntimeError("PGPASSWORD=must-not-leak")
|
|
|
|
app[routes.KB_CLAIM_LOADER_KEY] = fail
|
|
api_request = make_mocked_request(
|
|
"GET",
|
|
f"/api/kb/claims/{CLAIM_ID}",
|
|
app=app,
|
|
match_info={"claim_id": CLAIM_ID},
|
|
)
|
|
html_request = make_mocked_request(
|
|
"GET",
|
|
f"/kb/claims/{CLAIM_ID}",
|
|
app=app,
|
|
match_info={"claim_id": CLAIM_ID},
|
|
)
|
|
|
|
api_response = asyncio.run(routes.handle_api_kb_claim(api_request))
|
|
html_response = asyncio.run(routes.handle_kb_claim_page(html_request))
|
|
|
|
assert api_response.status == 503
|
|
assert html_response.status == 500
|
|
assert b"PGPASSWORD" not in api_response.body
|
|
assert b"PGPASSWORD" not in html_response.body
|