tests/test_workers.py

import pytest

from workers.adu_parser import ADUParserWorker
from workers.fallacy_detector import FallacyDetectorWorker
from workers.stance_tracker import StanceTrackerWorker
from workers.evidence_scorer import EvidenceScorerWorker


def _assert_schema(output, required_keys):
    assert isinstance(output, dict)
    assert required_keys.issubset(output.keys())


def test_adu_valid_input_schema_and_values(loaded_workers, sample_test_data):
    worker = loaded_workers["adu"]
    result = worker.predict(sample_test_data["adu"]["valid"])

    _assert_schema(
        result,
        {
            "tokens",
            "labels",
            "segments",
            "structure",
            "completeness",
            "score",
            "confidence",
            "uncertain",
        },
    )
    assert result["tokens"]
    assert isinstance(result["labels"], list)
    assert result["uncertain"] is False


@pytest.mark.parametrize("text_key", ["empty", "short", "long"])
def test_adu_edge_cases(sample_test_data, worker_factory, text_key):
    worker = worker_factory("adu", confidence_mode="high")
    result = worker.predict(sample_test_data["adu"][text_key])

    _assert_schema(
        result,
        {
            "tokens",
            "labels",
            "segments",
            "structure",
            "completeness",
            "score",
            "confidence",
            "uncertain",
        },
    )

    if text_key == "empty":
        assert result["tokens"] == []
        assert result["score"] == 0.0


def test_adu_confidence_threshold_applied(worker_factory, sample_test_data):
    low_conf_worker = worker_factory("adu", confidence_mode="low", confidence_threshold=0.70)
    high_conf_worker = worker_factory("adu", confidence_mode="high", confidence_threshold=0.70)

    low_result = low_conf_worker.predict(sample_test_data["adu"]["valid"])
    high_result = high_conf_worker.predict(sample_test_data["adu"]["valid"])

    assert low_result["confidence"] < 0.70
    assert low_result["uncertain"] is True
    assert high_result["confidence"] >= 0.70
    assert high_result["uncertain"] is False


def test_adu_error_handling(worker_factory, sample_test_data):
    worker = worker_factory("adu", confidence_mode="high")

    class FailingModel:
        def parameters(self):
            import torch
            parameter = torch.nn.Parameter(torch.zeros(1))
            return iter([parameter])

        def __call__(self, **enc):
            raise RuntimeError("ADU inference failed")

    worker.model = FailingModel()

    with pytest.raises(RuntimeError, match="ADU inference failed"):
        worker.predict(sample_test_data["adu"]["valid"])


def test_fallacy_valid_input_schema_and_values(loaded_workers, sample_test_data):
    worker = loaded_workers["fallacy"]
    result = worker.predict(sample_test_data["fallacy"]["valid"])

    _assert_schema(
        result,
        {
            "label",
            "label_name",
            "confidence",
            "is_fallacy",
            "score",
            "detail",
            "uncertain",
        },
    )
    assert isinstance(result["label"], int)
    assert isinstance(result["detail"], str)


@pytest.mark.parametrize("text_key", ["empty", "short", "long"])
def test_fallacy_edge_cases(sample_test_data, worker_factory, text_key):
    worker = worker_factory("fallacy", confidence_mode="high")
    result = worker.predict(sample_test_data["fallacy"][text_key])

    _assert_schema(
        result,
        {
            "label",
            "label_name",
            "confidence",
            "is_fallacy",
            "score",
            "detail",
            "uncertain",
        },
    )


def test_fallacy_confidence_threshold_applied(worker_factory, sample_test_data):
    low_conf_worker = worker_factory("fallacy", confidence_mode="low", confidence_threshold=0.55)
    high_conf_worker = worker_factory("fallacy", confidence_mode="high", confidence_threshold=0.55)

    low_result = low_conf_worker.predict(sample_test_data["fallacy"]["valid"])
    high_result = high_conf_worker.predict(sample_test_data["fallacy"]["valid"])

    assert low_result["confidence"] < 0.55
    assert low_result["uncertain"] is True
    assert high_result["confidence"] >= 0.55
    assert high_result["uncertain"] is False


def test_fallacy_error_handling(worker_factory, sample_test_data):
    worker = worker_factory("fallacy", confidence_mode="high")

    class FailingTokenizer:
        def __call__(self, *args, **kwargs):
            raise RuntimeError("Fallacy tokenizer failed")

    worker.tokenizer = FailingTokenizer()

    with pytest.raises(RuntimeError, match="Fallacy tokenizer failed"):
        worker.predict(sample_test_data["fallacy"]["valid"])


def test_stance_valid_input_schema_and_values(loaded_workers, sample_test_data):
    worker = loaded_workers["stance"]
    result = worker.predict(
        sample_test_data["stance"]["valid"],
        sample_test_data["stance"]["topic"],
    )

    _assert_schema(
        result,
        {
            "label",
            "stance",
            "confidence",
            "score",
            "uncertain",
        },
    )
    assert result["stance"] in {"FAVOR", "AGAINST", "NONE"}


@pytest.mark.parametrize("text_key", ["empty", "short", "long"])
def test_stance_edge_cases(sample_test_data, worker_factory, text_key):
    worker = worker_factory("stance", confidence_mode="high")
    result = worker.predict(
        sample_test_data["stance"][text_key],
        sample_test_data["stance"]["topic"],
    )

    _assert_schema(
        result,
        {
            "label",
            "stance",
            "confidence",
            "score",
            "uncertain",
        },
    )


def test_stance_confidence_threshold_applied(worker_factory, sample_test_data):
    low_conf_worker = worker_factory("stance", confidence_mode="low", confidence_threshold=0.60)
    high_conf_worker = worker_factory("stance", confidence_mode="high", confidence_threshold=0.60)

    low_result = low_conf_worker.predict(
        sample_test_data["stance"]["valid"],
        sample_test_data["stance"]["topic"],
    )
    high_result = high_conf_worker.predict(
        sample_test_data["stance"]["valid"],
        sample_test_data["stance"]["topic"],
    )

    assert low_result["confidence"] < 0.60
    assert low_result["uncertain"] is True
    assert low_result["stance"] == "NONE"
    assert high_result["confidence"] >= 0.60
    assert high_result["uncertain"] is False


def test_stance_error_handling(worker_factory, sample_test_data):
    worker = worker_factory("stance", confidence_mode="high")

    class FailingModel:
        def parameters(self):
            import torch
            parameter = torch.nn.Parameter(torch.zeros(1))
            return iter([parameter])

        def __call__(self, **enc):
            raise RuntimeError("Stance model crashed")

    worker.model = FailingModel()

    with pytest.raises(RuntimeError, match="Stance model crashed"):
        worker.predict(
            sample_test_data["stance"]["valid"],
            sample_test_data["stance"]["topic"],
        )


def test_evidence_valid_input_schema_and_values(loaded_workers, sample_test_data):
    worker = loaded_workers["evidence"]
    result = worker.predict(sample_test_data["evidence"]["valid"])

    _assert_schema(
        result,
        {
            "label",
            "class",
            "confidence",
            "score",
            "uncertain",
        },
    )
    assert result["class"] in {"strong", "weak", "none"}


@pytest.mark.parametrize("text_key", ["empty", "short", "long"])
def test_evidence_edge_cases(sample_test_data, worker_factory, text_key):
    worker = worker_factory("evidence", confidence_mode="high")
    result = worker.predict(sample_test_data["evidence"][text_key])

    _assert_schema(
        result,
        {
            "label",
            "class",
            "confidence",
            "score",
            "uncertain",
        },
    )


def test_evidence_confidence_threshold_applied(worker_factory, sample_test_data):
    low_conf_worker = worker_factory("evidence", confidence_mode="low", confidence_threshold=0.60)
    high_conf_worker = worker_factory("evidence", confidence_mode="high", confidence_threshold=0.60)

    low_result = low_conf_worker.predict(sample_test_data["evidence"]["valid"])
    high_result = high_conf_worker.predict(sample_test_data["evidence"]["valid"])

    assert low_result["confidence"] < 0.60
    assert low_result["uncertain"] is True
    assert high_result["confidence"] >= 0.60
    assert high_result["uncertain"] is False


def test_evidence_error_handling(worker_factory, sample_test_data):
    worker = worker_factory("evidence", confidence_mode="high")

    class FailingTokenizer:
        def __call__(self, *args, **kwargs):
            raise RuntimeError("Evidence tokenizer failed")

    worker.tokenizer = FailingTokenizer()

    with pytest.raises(RuntimeError, match="Evidence tokenizer failed"):
        worker.predict(sample_test_data["evidence"]["valid"])


def test_worker_load_error_handling(monkeypatch):
    def failed_load(_self):
        raise RuntimeError("forced load failure")

    monkeypatch.setattr(ADUParserWorker, "_load", failed_load)
    monkeypatch.setattr(FallacyDetectorWorker, "_load", failed_load)
    monkeypatch.setattr(StanceTrackerWorker, "_load", failed_load)
    monkeypatch.setattr(EvidenceScorerWorker, "_load", failed_load)

    with pytest.raises(RuntimeError, match="forced load failure"):
        ADUParserWorker("mock://adu")

    with pytest.raises(RuntimeError, match="forced load failure"):
        FallacyDetectorWorker("mock://fallacy")

    with pytest.raises(RuntimeError, match="forced load failure"):
        StanceTrackerWorker("mock://stance")

    with pytest.raises(RuntimeError, match="forced load failure"):
        EvidenceScorerWorker("mock://evidence")