reuse-surface/tests/test_plan_check.py

from __future__ import annotations

import json

from reuse_surface.plan_check import (
    MatchQuery,
    load_query_from_intent,
    load_query_from_workplan,
    maybe_file_capability_request,
    match_query,
    record_outcome,
    run_plan_check,
    verdict_for_score,
)

SAMPLE_CAPABILITIES = [
    {
        "id": "capability.infotech.issue-tracking",
        "name": "Universal Issue Tracking Coordination",
        "summary": "Unified interface for issue tracking coordination across Gitea, GitHub, GitLab.",
        "vector": "D4 / A2 / C2 / R1",
        "owner": "issue-core",
        "tags": ["issue-tracking", "coordination"],
    },
    {
        "id": "capability.audit.event-retain",
        "name": "Audit Event Retention",
        "summary": "Collect, normalize, retain, and search audit events with integrity evidence across tenants.",
        "vector": "D4 / A2 / C2 / R1",
        "owner": "audit-core",
        "tags": ["audit"],
    },
]


def test_verdict_thresholds():
    assert verdict_for_score(0.5) == "reuse"
    assert verdict_for_score(0.3) == "extend"
    assert verdict_for_score(0.1) == "new"
    assert verdict_for_score(0.45) == "reuse"
    assert verdict_for_score(0.22) == "extend"


def test_match_query_finds_close_match():
    query = load_query_from_intent(
        "unified issue tracking coordination across Gitea GitHub GitLab"
    )
    matches = match_query(query, SAMPLE_CAPABILITIES)
    assert matches
    assert matches[0].id == "capability.infotech.issue-tracking"
    assert matches[0].score > 0.3


def test_match_query_empty_when_no_overlap():
    query = load_query_from_intent("something entirely unrelated xyzzy plugh")
    matches = match_query(query, SAMPLE_CAPABILITIES)
    assert matches == []


def test_match_query_no_capabilities():
    query = load_query_from_intent("anything")
    assert match_query(query, []) == []


def test_load_query_from_workplan(tmp_path):
    workplan = tmp_path / "TEST-WP-0001-thing.md"
    workplan.write_text(
        """---
id: TEST-WP-0001
title: "Unified issue tracking coordination"
status: proposed
---

# Unified issue tracking coordination

## Core Idea

Build a single interface for issue tracking across Gitea, GitHub, and GitLab.
"""
    )
    query = load_query_from_workplan(workplan)
    assert query.source == "workplan"
    assert query.workplan_id == "TEST-WP-0001"
    assert "issue tracking" in query.text.lower()
    matches = match_query(query, SAMPLE_CAPABILITIES)
    assert matches[0].id == "capability.infotech.issue-tracking"


def test_run_plan_check_reuse_verdict(monkeypatch):
    monkeypatch.setattr(
        "reuse_surface.plan_check.load_federated_capabilities",
        lambda: (SAMPLE_CAPABILITIES, "2026-07-06"),
    )
    query = MatchQuery(
        source="intent",
        text="unified issue tracking coordination gitea github gitlab",
        tokens={"unified", "issue", "tracking", "coordination", "gitea", "github", "gitlab"},
    )
    result = run_plan_check(query)
    assert result["verdict"] == "reuse"
    assert result["matches"][0]["id"] == "capability.infotech.issue-tracking"
    assert result["federated_index_stale_warning"] is None


def test_run_plan_check_stale_warning(monkeypatch):
    monkeypatch.setattr(
        "reuse_surface.plan_check.load_federated_capabilities",
        lambda: (SAMPLE_CAPABILITIES, "2020-01-01"),
    )
    query = load_query_from_intent("nothing matches this at all zzz")
    result = run_plan_check(query)
    assert result["verdict"] == "new"
    assert "days old" in result["federated_index_stale_warning"]


def test_record_outcome_appends_jsonl(tmp_path, monkeypatch):
    telemetry_path = tmp_path / "plan-check-events.jsonl"
    monkeypatch.setattr("reuse_surface.plan_check.TELEMETRY_PATH", telemetry_path)
    result = {
        "verdict": "reuse",
        "matches": [{"id": "capability.infotech.issue-tracking"}],
    }
    record_outcome(result, "reused", consumer_repo="some-repo")
    lines = telemetry_path.read_text().splitlines()
    assert len(lines) == 1
    event = json.loads(lines[0])
    assert event["consumer_repo"] == "some-repo"
    assert event["capability_id"] == "capability.infotech.issue-tracking"
    assert event["outcome"] == "reused"
    assert event["source"] == "plan-check"


def test_maybe_file_capability_request_only_on_new_verdict(monkeypatch):
    called = []
    monkeypatch.setattr(
        "reuse_surface.plan_check.file_capability_request",
        lambda **kwargs: called.append(kwargs) or {"id": "req-1"},
    )
    reuse_result = {"verdict": "reuse", "query": {"text": "x", "workplan_id": None}}
    assert maybe_file_capability_request(
        reuse_result, requesting_domain="infotech", requesting_agent="a"
    ) is None
    assert called == []

    new_result = {"verdict": "new", "query": {"text": "some new intent", "workplan_id": None}}
    filed = maybe_file_capability_request(
        new_result, requesting_domain="infotech", requesting_agent="a"
    )
    assert filed == {"id": "req-1"}
    assert called[0]["requesting_domain"] == "infotech"


def test_cmd_plan_check_file_request_flag(monkeypatch):
    from reuse_surface.cli import main

    monkeypatch.setattr(
        "reuse_surface.plan_check.load_federated_capabilities",
        lambda: (SAMPLE_CAPABILITIES, "2026-07-06"),
    )
    filed_calls = []
    monkeypatch.setattr(
        "reuse_surface.cli.maybe_file_capability_request",
        lambda result, **kwargs: filed_calls.append(kwargs) or {"id": "req-9"},
    )
    exit_code = main(
        ["plan-check", "--intent", "totally unrelated xyzzy plugh", "--file-request"]
    )
    assert exit_code == 0
    assert filed_calls


def test_cmd_plan_check_intent_json(monkeypatch):
    from reuse_surface.cli import main

    monkeypatch.setattr(
        "reuse_surface.plan_check.load_federated_capabilities",
        lambda: (SAMPLE_CAPABILITIES, "2026-07-06"),
    )
    exit_code = main(
        ["plan-check", "--intent", "issue tracking gitea github gitlab", "--format", "json"]
    )
    assert exit_code == 0


def test_cmd_plan_check_requires_input():
    from reuse_surface.cli import main

    exit_code = main(["plan-check"])
    assert exit_code == 1


def test_cmd_plan_check_rejects_both_inputs(tmp_path):
    from reuse_surface.cli import main

    workplan = tmp_path / "x.md"
    workplan.write_text("---\nid: X\n---\nbody")
    exit_code = main(["plan-check", str(workplan), "--intent", "also this"])
    assert exit_code == 1