Upload 8 files

tests/test_endpoints.py

@@ -0,0 +1,97 @@
+"""Tests for the FastAPI server endpoints."""
+
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+# Skip entire module if environment can't be created (e.g. missing datasets)
+# This allows CI to run without HF dataset access
+pytestmark = pytest.mark.skipif(
+    os.getenv("SKIP_SERVER_TESTS", "1") == "1",
+    reason="Set SKIP_SERVER_TESTS=0 to run server tests (requires dataset access)"
+)
+
+
+class TestHealthEndpoint:
+    def test_health_returns_200(self):
+        from server.app import app
+        client = TestClient(app)
+        resp = client.get("/health")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["status"] == "healthy"
+        assert "version" in data
+
+
+class TestTasksEndpoint:
+    def test_tasks_returns_three_tasks(self):
+        from server.app import app
+        client = TestClient(app)
+        resp = client.get("/tasks")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert len(data["tasks"]) == 3
+        task_ids = [t["task_id"] for t in data["tasks"]]
+        assert "task_1_factual_grounding" in task_ids
+        assert "task_2_multi_hop_synthesis" in task_ids
+        assert "task_3_adversarial_resistance" in task_ids
+
+    def test_tasks_has_action_schema(self):
+        from server.app import app
+        client = TestClient(app)
+        resp = client.get("/tasks")
+        data = resp.json()
+        assert "action_schema" in data
+        assert "answer" in data["action_schema"]["properties"]
+
+
+class TestGraderEndpoint:
+    def test_grader_requires_task_id(self):
+        from server.app import app
+        client = TestClient(app)
+        resp = client.post("/grader", json={"step_rewards": [0.5], "step_infos": [{}]})
+        assert resp.status_code == 422
+
+    def test_grader_invalid_task_id(self):
+        from server.app import app
+        client = TestClient(app)
+        resp = client.post("/grader", json={
+            "task_id": "nonexistent",
+            "step_rewards": [0.5],
+            "step_infos": [{}],
+        })
+        assert resp.status_code == 404
+
+    def test_grader_returns_score(self):
+        from server.app import app
+        client = TestClient(app)
+        resp = client.post("/grader", json={
+            "task_id": "task_1_factual_grounding",
+            "step_rewards": [0.7, 0.5, 0.3],
+            "step_infos": [
+                {"correctness": 0.7, "grounding": 0.6, "calibration": 0.8,
+                 "hallucination_score": 0.1, "is_hallucination": False},
+                {"correctness": 0.5, "grounding": 0.4, "calibration": 0.7,
+                 "hallucination_score": 0.2, "is_hallucination": False},
+                {"correctness": 0.3, "grounding": 0.3, "calibration": 0.6,
+                 "hallucination_score": 0.5, "is_hallucination": True},
+            ],
+        })
+        assert resp.status_code == 200
+        data = resp.json()
+        assert 0.0 <= data["score"] <= 1.0
+        assert "breakdown" in data
+
+
+class TestMetadataEndpoint:
+    def test_metadata(self):
+        from server.app import app
+        client = TestClient(app)
+        resp = client.get("/metadata")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["name"] == "hallucination-guard-env"
+        assert "version" in data

tests/test_grader.py

@@ -0,0 +1,247 @@
+"""Tests for the 9-component reward system and hallucination detection."""
+
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import pytest
+from server.grader import (
+    calculate_reward,
+    detect_hallucination_advanced,
+    compute_calibration_error,
+    is_refusal_answer,
+    normalize_text,
+    check_quote_in_context_advanced,
+    check_factual_accuracy_advanced,
+    compute_rouge,
+    compute_bertscore,
+    HallucinationType,
+    HallucinationSeverity,
+)
+
+
+class TestRewardRange:
+    """Rewards must always be in [0, 1]."""
+
+    @pytest.mark.parametrize("difficulty", ["beginner", "intermediate", "advanced", "expert"])
+    def test_reward_in_range_correct_answer(self, difficulty):
+        reward, info = calculate_reward(
+            answer="Paris is the capital of France.",
+            confidence=0.9,
+            source_quote="Paris is the capital of France.",
+            context="Paris is the capital of France. It is located in northern France.",
+            ground_truth="Paris",
+            difficulty_level=difficulty,
+        )
+        assert 0.0 <= reward <= 1.0, f"Reward {reward} out of range for {difficulty}"
+
+    def test_reward_in_range_wrong_answer(self):
+        reward, info = calculate_reward(
+            answer="London is the capital of France.",
+            confidence=0.9,
+            source_quote="London is the capital of France.",
+            context="Paris is the capital of France.",
+            ground_truth="Paris",
+        )
+        assert 0.0 <= reward <= 1.0
+
+    def test_reward_in_range_empty_answer(self):
+        reward, info = calculate_reward(
+            answer="",
+            confidence=0.5,
+            source_quote="",
+            context="Some context here.",
+            ground_truth="Some answer",
+        )
+        assert 0.0 <= reward <= 1.0
+
+    def test_reward_in_range_refusal(self):
+        reward, info = calculate_reward(
+            answer="I cannot answer from the provided context.",
+            confidence=0.3,
+            source_quote="",
+            context="Some unrelated context.",
+            ground_truth="not mentioned in context",
+        )
+        assert 0.0 <= reward <= 1.0
+
+
+class TestRefusalHandling:
+    """Proper refusals on unanswerable questions should be rewarded."""
+
+    def test_proper_refusal_rewarded(self):
+        reward, info = calculate_reward(
+            answer="I cannot answer from the provided context.",
+            confidence=0.3,
+            source_quote="",
+            context="The sky is blue.",
+            ground_truth="not mentioned in context",
+        )
+        assert reward >= 0.5, f"Proper refusal should get reward >= 0.5, got {reward}"
+        assert info.get("is_refusal") is True
+
+    def test_underconfident_refusal_penalized(self):
+        """Refusing when the answer IS in context should be penalized."""
+        reward, info = calculate_reward(
+            answer="I cannot determine the answer from the context.",
+            confidence=0.3,
+            source_quote="",
+            context="The capital of France is Paris.",
+            ground_truth="Paris",
+        )
+        assert reward <= 0.4, f"Underconfident refusal should be penalized, got {reward}"
+
+    def test_overconfident_refusal(self):
+        """High confidence refusal on answerable question should be penalized."""
+        reward, info = calculate_reward(
+            answer="I don't know the answer.",
+            confidence=0.9,
+            source_quote="",
+            context="The capital of France is Paris.",
+            ground_truth="Paris",
+        )
+        assert reward <= 0.5
+
+
+class TestHallucinationDetection:
+    """Hallucination detection should classify types correctly."""
+
+    def test_no_hallucination_for_grounded_answer(self):
+        score, htype, severity, analysis = detect_hallucination_advanced(
+            answer="Paris is the capital of France.",
+            context="Paris is the capital of France.",
+            ground_truth="Paris",
+            confidence=0.9,
+        )
+        assert score < 0.3, f"Grounded answer should have low hallucination score, got {score}"
+
+    def test_fabricated_fact_detected(self):
+        score, htype, severity, analysis = detect_hallucination_advanced(
+            answer="Berlin is the capital of France.",
+            context="Paris is the capital of France.",
+            ground_truth="Paris",
+            confidence=0.9,
+        )
+        assert score > 0.3, f"Fabricated fact should have high hallucination score, got {score}"
+
+    def test_numerical_fabrication_detected(self):
+        score, htype, severity, analysis = detect_hallucination_advanced(
+            answer="The population is 8.7 million.",
+            context="The population is 2.1 million people.",
+            ground_truth="2.1 million",
+            confidence=0.8,
+        )
+        assert analysis.get("numerical_fabrication", 0) > 0, \
+            f"Fabricated number 8.7 should be detected, got {analysis}"
+
+
+class TestCitationAccuracy:
+    """Source quote verification should work correctly."""
+
+    def test_exact_quote_match(self):
+        score, analysis = check_quote_in_context_advanced(
+            "Paris is the capital of France.",
+            "Paris is the capital of France. It is a beautiful city.",
+        )
+        assert score == 1.0, f"Exact quote should score 1.0, got {score}"
+
+    def test_no_quote(self):
+        score, analysis = check_quote_in_context_advanced(
+            "",
+            "Some context here.",
+        )
+        assert score == 0.0
+
+    def test_partial_quote(self):
+        score, analysis = check_quote_in_context_advanced(
+            "capital of France",
+            "Paris is the capital of France.",
+        )
+        assert score > 0.5, f"Partial quote should score > 0.5, got {score}"
+
+
+class TestCalibrationError:
+    """Calibration error should penalize overconfidence."""
+
+    def test_perfect_calibration(self):
+        error = compute_calibration_error(0.9, 0.9)
+        assert error == 0.0
+
+    def test_overconfidence_penalized(self):
+        error = compute_calibration_error(0.95, 0.3)
+        assert error > 0.5, f"Overconfidence should be heavily penalized, got {error}"
+
+    def test_underconfidence_safe(self):
+        error = compute_calibration_error(0.3, 0.9)
+        assert error < compute_calibration_error(0.95, 0.3), \
+            "Overconfidence should be penalized more than underconfidence"
+
+
+class TestBERTScoreEdgeCases:
+    """BERTScore should not crash on edge cases."""
+
+    def test_empty_strings(self):
+        result = compute_bertscore("", "")
+        assert result["f1"] == 0.0
+
+    def test_identical_strings(self):
+        result = compute_bertscore("The cat sat on the mat.", "The cat sat on the mat.")
+        assert result["f1"] > 0.8, f"Identical strings should have high BERTScore, got {result['f1']}"
+
+    def test_short_strings(self):
+        result = compute_bertscore("yes", "no")
+        assert "f1" in result  # Should not crash
+
+
+class TestROUGE:
+    """ROUGE scores should be computed correctly."""
+
+    def test_identical_strings(self):
+        result = compute_rouge("The cat sat on the mat.", "The cat sat on the mat.")
+        assert result["rougeL"] == 1.0
+
+    def test_completely_different(self):
+        result = compute_rouge("The cat sat on the mat.", "Dogs run in the park.")
+        assert result["rougeL"] < 0.5
+
+    def test_empty_strings(self):
+        result = compute_rouge("", "")
+        assert result["rouge1"] == 0.0
+
+
+class TestFactualAccuracy:
+    """Factual accuracy should handle various answer types."""
+
+    def test_exact_match(self):
+        score, analysis = check_factual_accuracy_advanced(
+            "Paris", "Paris", "Paris is the capital of France."
+        )
+        assert score >= 0.9, f"Exact match should score high, got {score}"
+
+    def test_wrong_answer(self):
+        score, analysis = check_factual_accuracy_advanced(
+            "London", "Paris", "Paris is the capital of France."
+        )
+        assert score < 0.5, f"Wrong answer should score low, got {score}"
+
+    def test_contains_truth(self):
+        score, analysis = check_factual_accuracy_advanced(
+            "The capital is Paris, which is in northern France.",
+            "Paris",
+            "Paris is the capital of France.",
+        )
+        assert score >= 0.8, f"Answer containing truth should score high, got {score}"
+
+
+class TestNormalizeText:
+    """Text normalization should handle edge cases."""
+
+    def test_empty_string(self):
+        assert normalize_text("") == ""
+
+    def test_whitespace_normalization(self):
+        result = normalize_text("  The   cat   sat  ")
+        assert "  " not in result
+
+    def test_case_normalization(self):
+        result = normalize_text("PARIS IS THE CAPITAL")
+        assert result == result.lower()

tests/test_tasks.py

@@ -0,0 +1,116 @@
+"""Tests for the OpenEnv task registry and grader."""
+
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import pytest
+from server.tasks import (
+    ALL_TASKS, TASK_1, TASK_2, TASK_3,
+    get_task, task_id_for_difficulty,
+    compute_task_score, ACTION_SCHEMA,
+)
+
+
+class TestTaskRegistry:
+    """Task registry should contain all required tasks."""
+
+    def test_three_tasks_exist(self):
+        assert len(ALL_TASKS) == 3
+        assert "task_1_factual_grounding" in ALL_TASKS
+        assert "task_2_multi_hop_synthesis" in ALL_TASKS
+        assert "task_3_adversarial_resistance" in ALL_TASKS
+
+    def test_task_difficulties(self):
+        assert TASK_1.difficulty == "beginner"
+        assert TASK_2.difficulty == "intermediate"
+        assert TASK_3.difficulty == "advanced"
+
+    def test_get_task(self):
+        assert get_task("task_1_factual_grounding") is TASK_1
+        assert get_task("nonexistent") is None
+
+    def test_difficulty_mapping(self):
+        assert task_id_for_difficulty("beginner") == TASK_1.task_id
+        assert task_id_for_difficulty("intermediate") == TASK_2.task_id
+        assert task_id_for_difficulty("advanced") == TASK_3.task_id
+        assert task_id_for_difficulty("expert") == TASK_3.task_id
+
+    def test_action_schema_has_required_fields(self):
+        props = ACTION_SCHEMA["properties"]
+        assert "answer" in props
+        assert "confidence" in props
+        assert "source_quote" in props
+        assert ACTION_SCHEMA["required"] == ["answer"]
+
+
+class TestTaskGrader:
+    """Per-episode task grader should produce scores in [0, 1]."""
+
+    def test_empty_steps_score_zero(self):
+        result = compute_task_score(TASK_1, [], [])
+        assert result["score"] == 0.0
+
+    def test_perfect_scores(self):
+        step_rewards = [1.0, 1.0, 1.0, 1.0, 1.0]
+        step_infos = [
+            {"correctness": 1.0, "grounding": 1.0, "calibration": 1.0,
+             "hallucination_score": 0.0, "is_hallucination": False}
+            for _ in range(5)
+        ]
+        result = compute_task_score(TASK_1, step_rewards, step_infos)
+        assert 0.9 <= result["score"] <= 1.05, f"Perfect answers should score ~1.0, got {result['score']}"
+
+    def test_zero_rewards_score_low(self):
+        step_rewards = [0.0, 0.0, 0.0]
+        step_infos = [
+            {"correctness": 0.0, "grounding": 0.0, "calibration": 0.0,
+             "hallucination_score": 1.0, "is_hallucination": True}
+            for _ in range(3)
+        ]
+        result = compute_task_score(TASK_1, step_rewards, step_infos)
+        assert result["score"] <= 0.1, f"All-wrong should score ~0, got {result['score']}"
+
+    def test_task3_overconfidence_penalty(self):
+        """Task 3 should penalize overconfident wrong answers."""
+        step_rewards = [0.3, 0.3, 0.3]
+        step_infos = [
+            {"correctness": 0.2, "grounding": 0.3, "calibration": 0.9,
+             "hallucination_score": 0.7, "is_hallucination": True}
+            for _ in range(3)
+        ]
+        result_t3 = compute_task_score(TASK_3, step_rewards, step_infos)
+        # Same data on task 1 should score higher than task 3
+        result_t1 = compute_task_score(TASK_1, step_rewards, step_infos)
+        assert result_t3["score"] <= result_t1["score"], \
+            f"Task 3 should penalize overconfidence more than Task 1"
+
+    def test_completion_bonus(self):
+        """5+ steps should get a completion bonus."""
+        short_infos = [{"correctness": 0.5, "grounding": 0.5, "calibration": 0.5,
+                        "hallucination_score": 0.0, "is_hallucination": False}]
+        long_infos = [{"correctness": 0.5, "grounding": 0.5, "calibration": 0.5,
+                       "hallucination_score": 0.0, "is_hallucination": False}
+                      for _ in range(6)]
+        short_result = compute_task_score(TASK_1, [0.5], short_infos)
+        long_result = compute_task_score(TASK_1, [0.5] * 6, long_infos)
+        assert long_result["score"] > short_result["score"], \
+            "Longer episodes should get completion bonus"
+
+    def test_score_always_in_range(self):
+        """Any combination of rewards/infos should produce score in [0, 1]."""
+        import random
+        random.seed(42)
+        for _ in range(100):
+            n = random.randint(1, 10)
+            rewards = [random.uniform(0, 1) for _ in range(n)]
+            infos = [{
+                "correctness": random.uniform(0, 1),
+                "grounding": random.uniform(0, 1),
+                "calibration": random.uniform(0, 1),
+                "hallucination_score": random.uniform(0, 1),
+                "is_hallucination": random.random() > 0.5,
+            } for _ in range(n)]
+            for task in [TASK_1, TASK_2, TASK_3]:
+                result = compute_task_score(task, rewards, infos)
+                assert 0.0 <= result["score"] <= 1.0, \
+                    f"Score {result['score']} out of range for {task.task_id}"