add tests

Author: ybdarrenwang

tests/strands_evals/chaos/__init__.py

@@ -0,0 +1 @@
+

tests/strands_evals/chaos/test_context.py

@@ -0,0 +1,33 @@
+"""Unit tests for the chaos _context module."""
+
+from strands_evals.chaos import ChaosScenario
+from strands_evals.chaos._context import _current_scenario
+
+
+class TestContextVar:
+    """Tests for the _current_scenario ContextVar."""
+
+    def test_set_and_get(self):
+        scenario = ChaosScenario(name="test_scenario")
+        token = _current_scenario.set(scenario)
+        try:
+            assert _current_scenario.get() is scenario
+            assert _current_scenario.get().name == "test_scenario"
+        finally:
+            _current_scenario.reset(token)
+
+    def test_nested_set_and_reset(self):
+        s1 = ChaosScenario(name="outer")
+        s2 = ChaosScenario(name="inner")
+
+        token1 = _current_scenario.set(s1)
+        try:
+            assert _current_scenario.get().name == "outer"
+            token2 = _current_scenario.set(s2)
+            try:
+                assert _current_scenario.get().name == "inner"
+            finally:
+                _current_scenario.reset(token2)
+            assert _current_scenario.get().name == "outer"
+        finally:
+            _current_scenario.reset(token1)

tests/strands_evals/chaos/test_effects.py

@@ -0,0 +1,163 @@
+"""Unit tests for chaos effect classes."""
+
+import random
+
+import pytest
+
+from strands_evals.chaos.effects import (
+    CorruptValues,
+    RemoveFields,
+    ToolCallFailure,
+    TruncateFields,
+)
+
+
+class TestToolCallFailure:
+    """Tests for the ToolCallFailure pre-hook effect."""
+
+    @pytest.mark.parametrize(
+        "error_type,expected_message",
+        [
+            ("timeout", "Tool call timed out"),
+            ("network_error", "Network unreachable"),
+            ("execution_error", "Tool execution failed"),
+            ("validation_error", "Tool input validation failed"),
+        ],
+    )
+    def test_apply_returns_default_message(self, error_type, expected_message):
+        effect = ToolCallFailure(error_type=error_type)
+        assert effect.apply() == expected_message
+
+    def test_apply_returns_custom_message_when_provided(self):
+        effect = ToolCallFailure(error_type="timeout", error_message="Custom timeout msg")
+        assert effect.apply() == "Custom timeout msg"
+
+    def test_apply_rate_defaults_to_one(self):
+        effect = ToolCallFailure()
+        assert effect.apply_rate == 1.0
+
+
+class TestTruncateFields:
+    """Tests for the TruncateFields post-hook effect."""
+
+    def test_truncates_long_strings(self):
+        effect = TruncateFields(max_length=5)
+        response = {"name": "hello world", "short": "hi"}
+        result = effect.apply(response)
+        assert result["name"] == "hello"
+        assert result["short"] == "hi"
+
+    def test_preserves_non_string_values(self):
+        effect = TruncateFields(max_length=3)
+        response = {"count": 42, "flag": True, "items": [1, 2, 3]}
+        result = effect.apply(response)
+        assert result["count"] == 42
+        assert result["flag"] is True
+        assert result["items"] == [1, 2, 3]
+
+    def test_truncates_nested_dicts(self):
+        effect = TruncateFields(max_length=3)
+        response = {"nested": {"deep_value": "abcdef"}}
+        result = effect.apply(response)
+        assert result["nested"]["deep_value"] == "abc"
+
+    def test_empty_dict_returns_empty(self):
+        effect = TruncateFields(max_length=5)
+        assert effect.apply({}) == {}
+
+    def test_non_dict_input_returned_as_is(self):
+        effect = TruncateFields(max_length=5)
+        assert effect.apply("not a dict") == "not a dict"
+        assert effect.apply(None) is None
+
+    def test_zero_max_length_truncates_all_strings(self):
+        effect = TruncateFields(max_length=0)
+        response = {"text": "hello"}
+        result = effect.apply(response)
+        assert result["text"] == ""
+
+
+class TestRemoveFields:
+    """Tests for the RemoveFields post-hook effect."""
+
+    def test_removes_at_least_one_field(self):
+        random.seed(42)
+        effect = RemoveFields(remove_ratio=0.1)
+        response = {"a": 1, "b": 2, "c": 3, "d": 4}
+        result = effect.apply(response)
+        assert len(result) < len(response)
+
+    def test_removes_half_fields(self):
+        random.seed(42)
+        effect = RemoveFields(remove_ratio=0.5)
+        response = {"a": 1, "b": 2, "c": 3, "d": 4}
+        result = effect.apply(response)
+        assert len(result) == 2
+
+    def test_removes_all_fields_at_ratio_one(self):
+        random.seed(42)
+        effect = RemoveFields(remove_ratio=1.0)
+        response = {"a": 1, "b": 2, "c": 3}
+        result = effect.apply(response)
+        assert len(result) == 0
+
+    def test_empty_dict_returns_empty(self):
+        effect = RemoveFields(remove_ratio=0.5)
+        assert effect.apply({}) == {}
+
+    def test_non_dict_input_returned_as_is(self):
+        effect = RemoveFields(remove_ratio=0.5)
+        assert effect.apply("not a dict") == "not a dict"
+        assert effect.apply(None) is None
+
+    def test_single_field_always_removed(self):
+        random.seed(42)
+        effect = RemoveFields(remove_ratio=0.5)
+        response = {"only_key": "value"}
+        result = effect.apply(response)
+        assert len(result) == 0
+
+
+class TestCorruptValues:
+    """Tests for the CorruptValues post-hook effect."""
+
+    def test_corrupts_at_least_one_field(self):
+        random.seed(42)
+        effect = CorruptValues(corrupt_ratio=0.1)
+        response = {"a": "original_a", "b": "original_b", "c": "original_c", "d": "original_d"}
+        result = effect.apply(response)
+        corrupted_count = sum(1 for k in response if result[k] != response[k])
+        assert corrupted_count >= 1
+
+    def test_corrupted_values_come_from_corruption_pool(self):
+        random.seed(42)
+        effect = CorruptValues(corrupt_ratio=1.0)
+        response = {"a": "original", "b": "data"}
+        result = effect.apply(response)
+        corruption_pool = [None, 99999, "", True, [], "CORRUPTED_DATA"]
+        for key in response:
+            assert result[key] in corruption_pool
+
+    def test_corrupts_nested_dicts_recursively(self):
+        random.seed(42)
+        effect = CorruptValues(corrupt_ratio=1.0)
+        response = {"top": "value", "nested": {"inner": "deep_value"}}
+        result = effect.apply(response)
+        # The nested dict should also be processed
+        assert "nested" in result or "top" in result
+
+    def test_empty_dict_returns_empty(self):
+        effect = CorruptValues(corrupt_ratio=0.5)
+        assert effect.apply({}) == {}
+
+    def test_non_dict_input_returned_as_is(self):
+        effect = CorruptValues(corrupt_ratio=0.5)
+        assert effect.apply("not a dict") == "not a dict"
+        assert effect.apply(None) is None
+
+    def test_corrupted_value_differs_from_original(self):
+        random.seed(42)
+        effect = CorruptValues(corrupt_ratio=1.0)
+        response = {"key": "unique_original_value"}
+        result = effect.apply(response)
+        assert result["key"] != "unique_original_value"

tests/strands_evals/chaos/test_experiment.py

@@ -0,0 +1,161 @@
+"""Unit tests for ChaosExperiment."""
+
+import pytest
+
+from strands_evals import Case
+from strands_evals.chaos import ChaosExperiment, ChaosScenario
+from strands_evals.chaos._context import _current_scenario
+from strands_evals.chaos.effects import CorruptValues, ToolCallFailure
+from strands_evals.evaluators.evaluator import Evaluator
+from strands_evals.types import EvaluationData, EvaluationOutput
+
+
+class MockChaosEvaluator(Evaluator):
+    """Simple evaluator that always passes."""
+
+    def evaluate(self, evaluation_case: EvaluationData) -> list[EvaluationOutput]:
+        return [EvaluationOutput(score=1.0, test_pass=True, reason="Mock pass")]
+
+
+@pytest.fixture
+def cases():
+    return [
+        Case(name="case_a", input="hello"),
+        Case(name="case_b", input="world"),
+    ]
+
+
+@pytest.fixture
+def scenarios():
+    return [
+        ChaosScenario(
+            name="search_timeout",
+            effects={"search_tool": [ToolCallFailure(error_type="timeout")]},
+        ),
+        ChaosScenario(
+            name="db_corrupt",
+            effects={"db_tool": [CorruptValues(corrupt_ratio=0.8)]},
+        ),
+    ]
+
+
+@pytest.fixture
+def evaluator():
+    return MockChaosEvaluator()
+
+
+class TestChaosExperiment:
+    """Tests for ChaosExperiment initialization and execution."""
+
+    def test_expanded_cases_count_with_baseline(self, cases, scenarios, evaluator):
+        experiment = ChaosExperiment(cases=cases, scenarios=scenarios, evaluators=[evaluator], include_baseline=True)
+        # 2 cases × (2 scenarios + 1 baseline) = 6
+        assert len(experiment._expanded_cases) == 6
+
+    def test_expanded_cases_count_without_baseline(self, cases, scenarios, evaluator):
+        experiment = ChaosExperiment(cases=cases, scenarios=scenarios, evaluators=[evaluator], include_baseline=False)
+        # 2 cases × 2 scenarios = 4
+        assert len(experiment._expanded_cases) == 4
+
+    def test_expanded_case_names_include_scenario(self, cases, scenarios, evaluator):
+        experiment = ChaosExperiment(cases=cases, scenarios=scenarios, evaluators=[evaluator], include_baseline=True)
+        names = [c.name for c in experiment._expanded_cases]
+        assert "case_a|baseline" in names
+        assert "case_a|search_timeout" in names
+        assert "case_a|db_corrupt" in names
+        assert "case_b|baseline" in names
+        assert "case_b|search_timeout" in names
+        assert "case_b|db_corrupt" in names
+
+    def test_each_expanded_case_has_unique_session_id(self, cases, scenarios, evaluator):
+        experiment = ChaosExperiment(cases=cases, scenarios=scenarios, evaluators=[evaluator])
+        session_ids = [c.session_id for c in experiment._expanded_cases]
+        assert len(session_ids) == len(set(session_ids))
+
+    def test_get_scenario_for_session(self, cases, scenarios, evaluator):
+        experiment = ChaosExperiment(cases=cases, scenarios=scenarios, evaluators=[evaluator], include_baseline=True)
+        # Pick an expanded case and verify its scenario maps correctly
+        for expanded_case in experiment._expanded_cases:
+            scenario = experiment.get_scenario_for_session(expanded_case.session_id)
+            assert scenario is not None
+            assert scenario.name in expanded_case.name
+
+    def test_get_scenario_for_unknown_session(self, cases, scenarios, evaluator):
+        experiment = ChaosExperiment(cases=cases, scenarios=scenarios, evaluators=[evaluator])
+        assert experiment.get_scenario_for_session("nonexistent-id") is None
+
+    def test_get_original_case_name(self, cases, scenarios, evaluator):
+        experiment = ChaosExperiment(cases=cases, scenarios=scenarios, evaluators=[evaluator], include_baseline=True)
+        for expanded_case in experiment._expanded_cases:
+            original_name = experiment.get_original_case_name(expanded_case.session_id)
+            assert original_name in ("case_a", "case_b")
+
+    def test_get_original_case_name_unknown_session(self, cases, scenarios, evaluator):
+        experiment = ChaosExperiment(cases=cases, scenarios=scenarios, evaluators=[evaluator])
+        assert experiment.get_original_case_name("nonexistent-id") is None
+
+    def test_context_var_set_and_reset(self, cases, scenarios, evaluator):
+        """Verify the ContextVar is set to the correct scenario during task execution and reset after."""
+        observed_scenarios = []
+
+        def capturing_task(case: Case):
+            scenario = _current_scenario.get()
+            observed_scenarios.append((case.name, scenario.name if scenario else None))
+            return "output"
+
+        experiment = ChaosExperiment(cases=cases, scenarios=scenarios, evaluators=[evaluator], include_baseline=True)
+        experiment.run_evaluations(task=capturing_task)
+
+        # Should have 6 observations (2 cases × 3 scenarios)
+        assert len(observed_scenarios) == 6
+
+        # Verify baseline scenarios observed
+        baseline_obs = [(name, sn) for name, sn in observed_scenarios if sn == "baseline"]
+        assert len(baseline_obs) == 2
+
+        # Verify chaos scenarios observed
+        timeout_obs = [(name, sn) for name, sn in observed_scenarios if sn == "search_timeout"]
+        assert len(timeout_obs) == 2
+
+        # After all runs, the ContextVar should be back to None
+        assert _current_scenario.get() is None
+
+    def test_context_var_reset_on_task_exception(self, evaluator):
+        """Verify the ContextVar is reset even if the task raises."""
+        cases = [Case(name="failing", input="x")]
+        scenarios_list = [ChaosScenario(name="chaos", effects={"t": [ToolCallFailure()]})]
+
+        call_count = [0]
+
+        def failing_task(case: Case):
+            call_count[0] += 1
+            if call_count[0] == 1:
+                raise RuntimeError("Task failed")
+            return "output"
+
+        experiment = ChaosExperiment(
+            cases=cases, scenarios=scenarios_list, evaluators=[evaluator], include_baseline=True
+        )
+
+        # The base Experiment should handle the exception internally
+        # ContextVar should still be reset
+        try:
+            experiment.run_evaluations(task=failing_task)
+        except Exception:
+            pass
+
+        assert _current_scenario.get() is None
+
+    def test_returns_evaluation_reports(self, cases, scenarios, evaluator):
+        """Verify run_evaluations returns reports."""
+
+        def task(case: Case):
+            return "output"
+
+        experiment = ChaosExperiment(cases=cases, scenarios=scenarios, evaluators=[evaluator], include_baseline=True)
+        reports = experiment.run_evaluations(task=task)
+
+        assert len(reports) >= 1
+        report = reports[0]
+        # 2 cases × 3 scenarios = 6 scores
+        assert len(report.scores) == 6