tests/test_topic_routing.py

from __future__ import annotations

from pathlib import Path

import pytest

from gcmd_classifier.classification import build_topic_candidates, route_topics
from gcmd_classifier.config import ModelSettings
from gcmd_classifier.errors import StructuredModelResponseError, UnknownCandidateIDError
from gcmd_classifier.llm import FakeModelClient
from gcmd_classifier.models import ArticleRecord, SupportType
from gcmd_classifier.vocabulary import load_vocabulary

FIXTURE_PATH = Path("tests/fixtures/gcmd_hierarchy_small.json")
FULL_HIERARCHY_PATH = Path("data/gcmd_hierarchy.json")


def _index():
    return load_vocabulary(FIXTURE_PATH)


def _article(
    abstract: str = "Atmospheric chemistry and ocean carbon are discussed.",
) -> ArticleRecord:
    return ArticleRecord(
        DOI="10.example/topic-routing",
        Title="Atmospheric and ocean observations",
        Year=2025,
        Abstract=abstract,
    )


def _decision(candidate_id: str, confidence: float | None = 0.82) -> dict:
    return {
        "candidate_id": candidate_id,
        "confidence": confidence,
        "evidence": "The title and abstract support this Topic.",
        "support_type": "explicit",
        "reason": "Primary subject is represented by this Topic.",
    }


def test_topic_candidates_are_built_from_vocabulary_index() -> None:
    index = _index()
    candidates = build_topic_candidates(index)

    assert len(candidates) == 2
    assert [candidate.prompt_candidate.name for candidate in candidates] == [
        record.name for record in index.topics()
    ]


def test_root_category_is_never_included_as_candidate() -> None:
    candidates = build_topic_candidates(_index())

    assert all(candidate.prompt_candidate.level == "Topic" for candidate in candidates)
    assert all(candidate.prompt_candidate.name != "EARTH SCIENCE" for candidate in candidates)


def test_every_topic_candidate_maps_to_uuid_bearing_topic_record() -> None:
    index = _index()

    for candidate in build_topic_candidates(index):
        record = index.get(candidate.topic_uuid)
        assert record.UUID
        assert record.level == "Topic"
        assert candidate.prompt_candidate.candidate_id == candidate.candidate_id


def test_candidate_ids_are_unique_and_stable() -> None:
    first = build_topic_candidates(_index())
    second = build_topic_candidates(_index())

    first_ids = [candidate.candidate_id for candidate in first]
    second_ids = [candidate.candidate_id for candidate in second]
    assert first_ids == second_ids
    assert len(first_ids) == len(set(first_ids))
    assert first_ids == ["topic_0001", "topic_0002"]


def test_candidate_construction_does_not_hard_code_topic_names_or_uuids() -> None:
    index = _index()
    candidates = build_topic_candidates(index)

    assert [candidate.topic_uuid for candidate in candidates] == [
        topic.UUID for topic in index.topics()
    ]
    assert [candidate.prompt_candidate.name for candidate in candidates] == [
        topic.name for topic in index.topics()
    ]


def test_single_topic_selection() -> None:
    index = _index()
    candidates = build_topic_candidates(index)
    client = FakeModelClient([{"selected": [_decision(candidates[0].candidate_id)]}])

    result = route_topics(
        article=_article(),
        vocabulary=index,
        model_client=client,
        settings=ModelSettings(),
    )

    assert result.selected_count == 1
    assert result.is_no_topic is False
    assert result.branches[0].topic_uuid == candidates[0].topic_uuid


def test_multiple_topic_selection() -> None:
    index = _index()
    candidates = build_topic_candidates(index)
    client = FakeModelClient(
        [
            {
                "selected": [
                    _decision(candidates[0].candidate_id),
                    _decision(candidates[1].candidate_id, confidence=None),
                ]
            }
        ]
    )

    result = route_topics(
        article=_article(),
        vocabulary=index,
        model_client=client,
        settings=ModelSettings(),
    )

    assert [branch.topic_uuid for branch in result.branches] == [
        candidates[0].topic_uuid,
        candidates[1].topic_uuid,
    ]
    assert result.branches[1].confidence is None


def test_no_topic_selection_with_reason() -> None:
    client = FakeModelClient(
        [
            {
                "selected": [],
                "ambiguous_alternatives": [],
                "no_selection_reason": "No supplied Topic is defensible.",
            }
        ]
    )

    result = route_topics(
        article=_article(),
        vocabulary=_index(),
        model_client=client,
        settings=ModelSettings(),
    )

    assert result.branches == ()
    assert result.is_no_topic is True
    assert result.no_selection_reason == "No supplied Topic is defensible."


def test_invalid_candidate_id_returned_by_fake_model_is_rejected() -> None:
    client = FakeModelClient([{"selected": [_decision("unknown-topic")]}])

    with pytest.raises(UnknownCandidateIDError):
        route_topics(
            article=_article(),
            vocabulary=_index(),
            model_client=client,
            settings=ModelSettings(),
        )


def test_empty_or_malformed_candidate_id_rejected_by_structured_schema() -> None:
    client = FakeModelClient([{"selected": [_decision("")]}])

    with pytest.raises(StructuredModelResponseError):
        route_topics(
            article=_article(),
            vocabulary=_index(),
            model_client=client,
            settings=ModelSettings(),
        )


def test_duplicate_candidate_id_returned_by_model_is_rejected() -> None:
    candidate_id = build_topic_candidates(_index())[0].candidate_id
    client = FakeModelClient([{"selected": [_decision(candidate_id), _decision(candidate_id)]}])

    with pytest.raises(UnknownCandidateIDError):
        route_topics(
            article=_article(),
            vocabulary=_index(),
            model_client=client,
            settings=ModelSettings(),
        )


def test_selected_topic_branch_seed_is_populated_from_vocabulary_index() -> None:
    index = _index()
    candidate = build_topic_candidates(index)[0]
    record = index.get(candidate.topic_uuid)
    client = FakeModelClient([{"selected": [_decision(candidate.candidate_id, confidence=0.73)]}])

    result = route_topics(
        article=_article(),
        vocabulary=index,
        model_client=client,
        settings=ModelSettings(model_name="unit-test-model", prompt_version_topic="topic-test"),
    )
    branch = result.branches[0]

    assert branch.branch_id == f"topic:{candidate.candidate_id}"
    assert branch.topic_uuid == record.UUID
    assert branch.topic_name == record.name
    assert branch.topic_level == "Topic"
    assert branch.topic_canonical_path == record.canonical_path
    assert branch.evidence == "The title and abstract support this Topic."
    assert branch.support_type is SupportType.EXPLICIT
    assert branch.confidence == 0.73
    assert branch.reason == "Primary subject is represented by this Topic."
    assert branch.candidate_id == candidate.candidate_id
    assert branch.prompt_version == "topic-test"
    assert branch.model_name == "unit-test-model"


def test_confidence_is_preserved_as_uncalibrated_metadata_only() -> None:
    candidate = build_topic_candidates(_index())[0]
    client = FakeModelClient([{"selected": [_decision(candidate.candidate_id, confidence=0.01)]}])

    result = route_topics(
        article=_article(),
        vocabulary=_index(),
        model_client=client,
        settings=ModelSettings(),
    )

    assert result.branches[0].confidence == 0.01
    assert result.branches[0].topic_uuid == candidate.topic_uuid


def test_no_fake_fallback_topic_is_created() -> None:
    client = FakeModelClient([{"selected": [], "no_selection_reason": "No Topic."}])

    result = route_topics(
        article=_article(),
        vocabulary=_index(),
        model_client=client,
        settings=ModelSettings(),
    )

    assert result.branches == ()
    assert "FALLBACK" not in str(result.model_dump())


def test_no_term_or_variable_routing_is_performed() -> None:
    candidate = build_topic_candidates(_index())[0]
    client = FakeModelClient([{"selected": [_decision(candidate.candidate_id)]}])

    route_topics(
        article=_article(),
        vocabulary=_index(),
        model_client=client,
        settings=ModelSettings(),
    )

    assert len(client.requests) == 1
    assert client.requests[0].stage.value == "topic"


def test_topic_router_uses_provider_neutral_model_interface() -> None:
    candidate = build_topic_candidates(_index())[0]
    client = FakeModelClient([{"selected": [_decision(candidate.candidate_id)]}])

    route_topics(
        article=_article(),
        vocabulary=_index(),
        model_client=client,
        settings=ModelSettings(provider="fake-provider", model_name="fake-model"),
    )

    request = client.requests[0]
    assert request.provider == "fake-provider"
    assert request.model_name == "fake-model"
    assert request.response_schema.__name__ == "TopicResponse"


def test_fake_model_receives_article_fields_and_topic_candidates() -> None:
    index = _index()
    candidate = build_topic_candidates(index)[0]
    article = _article()
    client = FakeModelClient([{"selected": [_decision(candidate.candidate_id)]}])

    route_topics(
        article=article,
        vocabulary=index,
        model_client=client,
        settings=ModelSettings(),
    )

    prompt = client.requests[0].prompt
    assert article.DOI in prompt
    assert article.Title in prompt
    assert article.Abstract in prompt
    assert candidate.candidate_id in prompt
    assert index.get(candidate.topic_uuid).name in prompt


def test_empty_abstract_remains_valid_and_is_passed_to_prompt() -> None:
    candidate = build_topic_candidates(_index())[0]
    client = FakeModelClient([{"selected": [_decision(candidate.candidate_id)]}])

    route_topics(
        article=_article(abstract=""),
        vocabulary=_index(),
        model_client=client,
        settings=ModelSettings(),
    )

    assert "<ABSTRACT>\n\n</ABSTRACT>" in client.requests[0].prompt


def test_full_data_current_vocabulary_has_14_topic_candidates() -> None:
    index = load_vocabulary(FULL_HIERARCHY_PATH)
    candidates = build_topic_candidates(index)

    assert len(candidates) == 14
    assert all(index.get(candidate.topic_uuid).UUID for candidate in candidates)
    assert all(index.get(candidate.topic_uuid).level == "Topic" for candidate in candidates)