atlas2-mcp/tests/test_resolve_strategies.py

import pytest

from app.resolve import ResolveService


@pytest.mark.anyio
async def test_cache_hit_is_resolved_when_confidence_satisfies_requested_mode():
    async def load_entity(_subject):
        return {
            "atlas_id": "atlas-1",
            "label": "Australia",
            "type": "atlas:Location",
            "description": "country in Oceania",
            "needs_curation": True,
            "confidence": 0.91,
        }

    async def fail_lookup(*_args, **_kwargs):
        raise AssertionError("wikidata lookup should not run")

    async def no_persist(_entity):
        return None

    svc = ResolveService(load_entity_fn=load_entity, wikidata_lookup_fn=fail_lookup, persist_entity_fn=no_persist)
    result = await svc.resolve(
        subject="Australia",
        constraints={"min_confidence": 0.85},
        strategy={"mode": "ranked"},
        debug={"include_explanations": True},
    )

    assert result["status"] == "resolved"
    assert result["confidence"] == 0.91
    assert result["meta"]["debug"]["used_cache"] is True


@pytest.mark.anyio
async def test_low_confidence_cache_triggers_reresolution_for_ranked_mode():
    calls = {"lookup": 0, "persist": 0}

    async def load_entity(_subject):
        return {
            "atlas_id": "atlas-1",
            "label": "Australia",
            "type": "atlas:Location",
            "description": "country in Oceania",
            "needs_curation": True,
            "confidence": 0.55,
        }

    async def wikidata_lookup(_subject, _language="en", _limit=5):
        calls["lookup"] += 1
        return [
            {"id": "Q408", "label": "Australia", "description": "country in Oceania", "type": "Q6256"}
        ]

    async def persist(_entity):
        calls["persist"] += 1

    svc = ResolveService(load_entity_fn=load_entity, wikidata_lookup_fn=wikidata_lookup, persist_entity_fn=persist)
    result = await svc.resolve(
        subject="Australia",
        context={"language": "en", "realm": "external"},
        constraints={"min_confidence": 0.85},
        hints={"expected_type": "location", "aliases": []},
        strategy={"mode": "ranked", "auto_accept_threshold": 0.85},
        debug={"include_explanations": True},
    )

    assert calls["lookup"] == 1
    assert calls["persist"] == 1
    assert result["status"] == "resolved"
    assert result["meta"]["debug"]["used_cache"] is False


@pytest.mark.anyio
async def test_interactive_returns_all_candidates_without_auto_accepting_below_threshold():
    async def no_hit(_subject):
        return None

    async def wikidata_lookup(_subject, _language="en", _limit=5):
        return [
            {"id": "Q1225", "label": "Georgia", "description": "country in Eastern Europe and West Asia", "type": "Q6256"},
            {"id": "Q1428", "label": "Georgia", "description": "state of the United States of America", "type": "Q35657"},
        ]

    async def no_persist(_entity):
        return None

    svc = ResolveService(load_entity_fn=no_hit, wikidata_lookup_fn=wikidata_lookup, persist_entity_fn=no_persist)
    result = await svc.resolve(
        subject="Georgia",
        context={"language": "en", "realm": "external"},
        constraints={"max_candidates": 5},
        hints={"expected_type": "location", "aliases": []},
        strategy={"mode": "interactive", "auto_accept_threshold": 0.95, "interactive_below_threshold": True},
        debug={"include_candidates": True, "include_explanations": True},
    )

    assert result["status"] == "ambiguous"
    assert len(result["candidates"]) == 2
    assert result["meta"]["debug"]["decision"] == "ambiguous_below_threshold"