"""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" assert data["packets"][0]["apply_preview"]["state"] == "not_applyable_yet" assert data["packets"][0]["apply_preview"]["blocked_fragments"][0]["missing"] == [ "canonical from_claim", "canonical to_claim", ] assert data["packets"][1]["apply_preview"]["state"] == "strict_apply_payload_ready" assert data["packets"][1]["apply_preview"]["rows"] == [ { "action": "insert_if_missing", "table": "public.claim_edges", "target": "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb -> cccccccc-cccc-cccc-cccc-cccccccccccc (supports)", "claim_ids": [ "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb", "cccccccc-cccc-cccc-cccc-cccccccccccc", ], "details": { "from_claim": "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb", "to_claim": "cccccccc-cccc-cccc-cccc-cccccccccccc", "edge_type": "supports", "weight": None, }, "source": "proposal.apply_payload", } ] def test_page_renders_applyability_and_next_admin_action(): proposal = _approved_freeform_proposal() packet = routes.proposal_review.classify_proposal(_approved_freeform_proposal()) packet["normalization_preview"] = routes.proposal_normalize.normalize_proposal(proposal) packet["apply_preview"] = routes.build_apply_preview(proposal, packet) 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 assert 'href="/kb/claims/d0000000-0000-0000-0000-000000000005"' in html assert "d0000000-0000-0000-0000-000000000005" in html assert "Apply preview" in html assert "not_applyable_yet" in html assert "No canonical row preview is executable yet." in html assert "canonical from_claim" in html def test_page_renders_strict_apply_payload_claim_links(): proposal = _approved_applyable_proposal() packet = routes.proposal_review.classify_proposal(proposal) packet["normalization_preview"] = routes.proposal_normalize.normalize_proposal(proposal) packet["apply_preview"] = routes.build_apply_preview(proposal, packet) data = routes.build_kb_proposal_response( [packet], {"status": "approved", "proposal_id": "", "limit": 20}, ) html = routes.render_kb_proposals_page(data) assert "strict_apply_payload_ready" in html assert "public.claim_edges" in html assert "insert_if_missing" in html assert 'href="/kb/claims/bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"' in html assert 'href="/kb/claims/cccccccc-cccc-cccc-cccc-cccccccccccc"' 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"