129 lines
4.5 KiB
Python
129 lines
4.5 KiB
Python
"""Tests for the Argus KB proposal review routes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
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 kb_proposal_routes as routes # noqa: E402
|
|
|
|
|
|
def _approved_freeform_proposal():
|
|
return {
|
|
"id": "14fa5ecc-ac7a-41c1-807d-a2e85b936617",
|
|
"proposal_type": "attach_evidence",
|
|
"status": "approved",
|
|
"proposed_by_handle": "leo",
|
|
"reviewed_by_handle": "m3ta",
|
|
"channel": "telegram",
|
|
"source_ref": "telegram:leo:m3taversal:claim-review",
|
|
"payload": {
|
|
"proposal_kind": "supersede_compressed_claim_with_split_claim_cluster",
|
|
"old_claim": {"id": "d0000000-0000-0000-0000-000000000005"},
|
|
"claim_candidates": [{"headline": "A"}, {"headline": "B"}],
|
|
"source_candidates": [{"source_key": "hayek"}],
|
|
"proposed_supersession_edges": [{"edge_type": "supersedes_component_of"}],
|
|
"proposed_concept_map_dependency": {"headline": "Distributed market intelligence"},
|
|
"schema_gap_note": "bridge cannot propose claims directly",
|
|
},
|
|
}
|
|
|
|
|
|
def _approved_applyable_proposal():
|
|
return {
|
|
"id": "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
|
|
"proposal_type": "add_edge",
|
|
"status": "approved",
|
|
"payload": {
|
|
"apply_payload": {
|
|
"from_claim": "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
|
|
"to_claim": "cccccccc-cccc-cccc-cccc-cccccccccccc",
|
|
"edge_type": "supports",
|
|
}
|
|
},
|
|
}
|
|
|
|
|
|
def _call_api(loader, path="/api/kb-proposals?status=approved&limit=2"):
|
|
app = web.Application()
|
|
app[routes.KB_PROPOSAL_LOADER_KEY] = loader
|
|
req = make_mocked_request("GET", path, app=app)
|
|
response = asyncio.run(routes.handle_api_kb_proposals(req))
|
|
return response.status, json.loads(response.body.decode())
|
|
|
|
|
|
def test_api_kb_proposals_returns_review_packets_and_counts():
|
|
seen = {}
|
|
|
|
def loader(args):
|
|
seen["status"] = args.status
|
|
seen["limit"] = args.limit
|
|
return [_approved_freeform_proposal(), _approved_applyable_proposal()]
|
|
|
|
status, data = _call_api(loader)
|
|
|
|
assert status == 200
|
|
assert seen == {"status": "approved", "limit": 2}
|
|
assert data["read_only"] is True
|
|
assert data["total"] == 2
|
|
assert data["worker_applyable_count"] == 1
|
|
assert data["review_state_counts"]["approved_needs_apply_payload"] == 1
|
|
assert data["review_state_counts"]["approved_applyable"] == 1
|
|
assert data["packets"][0]["identity_and_authorization"]["source_ref"] == "telegram:leo:m3taversal:claim-review"
|
|
assert data["packets"][0]["normalization_preview"]["normalization_state"] == "blocked_missing_canonical_ids"
|
|
assert data["packets"][1]["normalization_preview"]["normalization_state"] == "already_strict_apply_payload"
|
|
|
|
|
|
def test_page_renders_applyability_and_next_admin_action():
|
|
packet = routes.proposal_review.classify_proposal(_approved_freeform_proposal())
|
|
packet["normalization_preview"] = routes.proposal_normalize.normalize_proposal(_approved_freeform_proposal())
|
|
data = routes.build_kb_proposal_response(
|
|
[packet],
|
|
{"status": "approved", "proposal_id": "", "limit": 20},
|
|
)
|
|
|
|
html = routes.render_kb_proposals_page(data)
|
|
|
|
assert "KB Proposals" in html
|
|
assert "approved_needs_apply_payload" in html
|
|
assert "strict payload.apply_payload" in html
|
|
assert "claim-row normalization" in html
|
|
assert "normalize into one or more strict apply_payload" in html
|
|
assert "Normalization" in html
|
|
assert "blocked_missing_canonical_ids" in html
|
|
|
|
|
|
def test_api_kb_proposals_supports_proposal_id_filter():
|
|
seen = {}
|
|
|
|
def loader(args):
|
|
seen["status"] = args.status
|
|
seen["proposal_id"] = args.proposal_id
|
|
seen["limit"] = args.limit
|
|
return [_approved_freeform_proposal()]
|
|
|
|
status, data = _call_api(
|
|
loader,
|
|
"/api/kb-proposals?proposal_id=14fa5ecc-ac7a-41c1-807d-a2e85b936617&status=pending_review&limit=999",
|
|
)
|
|
|
|
assert status == 200
|
|
assert seen == {
|
|
"status": None,
|
|
"proposal_id": "14fa5ecc-ac7a-41c1-807d-a2e85b936617",
|
|
"limit": routes.MAX_LIMIT,
|
|
}
|
|
assert data["filters"]["status"] == ""
|
|
assert data["filters"]["proposal_id"] == "14fa5ecc-ac7a-41c1-807d-a2e85b936617"
|