Spaces:
Running
Running
| """Focused P0 tests for EngineeringState's safety and rollout contract.""" | |
| from __future__ import annotations | |
| import os | |
| import sys | |
| import unittest | |
| from unittest.mock import patch | |
| _BACKEND = os.path.join(os.path.dirname(__file__), "..") | |
| if _BACKEND not in sys.path: | |
| sys.path.insert(0, _BACKEND) | |
| from agents.engineering_state import ( # noqa: E402 | |
| EngineeringState, | |
| EngineeringStateConfig, | |
| EngineeringStateMode, | |
| SCHEMA_VERSION, | |
| redact_text, | |
| ) | |
| class TestEngineeringState(unittest.TestCase): | |
| def test_default_rollout_is_authoritative_and_invalid_mode_fails_closed(self) -> None: | |
| with patch.dict(os.environ, {}, clear=True): | |
| self.assertEqual(EngineeringStateConfig.from_env().mode, EngineeringStateMode.AUTHORITATIVE) | |
| with patch.dict(os.environ, {"ENGINEERING_STATE_MODE": "unsafe"}, clear=False): | |
| self.assertEqual(EngineeringStateConfig.from_env().mode, EngineeringStateMode.OFF) | |
| def test_redaction_removes_common_credentials(self) -> None: | |
| value = "Authorization: Bearer abcdefghijkl token=ghp_1234567890abcdef hf_1234567890" | |
| result = redact_text(value) | |
| self.assertNotIn("abcdefghijkl", result) | |
| self.assertNotIn("ghp_1234567890abcdef", result) | |
| self.assertNotIn("hf_1234567890", result) | |
| self.assertIn("[REDACTED]", result) | |
| def test_transitions_are_validated_and_idempotent(self) -> None: | |
| state = EngineeringState.start("build a safe agent", run_id="run-1", now_ms=100) | |
| self.assertTrue(state.transition("CLASSIFYING", now_ms=101)) | |
| self.assertFalse(state.transition("CLASSIFYING", now_ms=102)) | |
| with self.assertRaises(ValueError): | |
| state.transition("IDLE", now_ms=103) | |
| self.assertEqual(state.revision, 1) | |
| self.assertEqual(state.sequence, 1) | |
| def test_round_trip_is_bounded_and_does_not_store_raw_goal(self) -> None: | |
| goal = "use token=super-secret-value to build this agent" | |
| state = EngineeringState.start(goal, run_id="run-2", session_id="session-2", now_ms=100) | |
| for target in ("CLASSIFYING", "THINKING", "COMPLETED"): | |
| state.transition(target, now_ms=101) | |
| snapshot = state.snapshot() | |
| restored = EngineeringState.from_snapshot(snapshot) | |
| self.assertEqual(restored.snapshot(), snapshot) | |
| self.assertEqual(snapshot["schema_version"], SCHEMA_VERSION) | |
| self.assertNotIn("super-secret-value", str(snapshot)) | |
| self.assertLessEqual(len(snapshot["history"]), 64) | |
| def test_corrupt_schema_and_revision_are_rejected(self) -> None: | |
| state = EngineeringState.start("goal", run_id="run-3") | |
| snapshot = state.snapshot() | |
| snapshot["schema_version"] = 999 | |
| with self.assertRaises(ValueError): | |
| EngineeringState.from_snapshot(snapshot) | |
| snapshot = state.snapshot() | |
| snapshot["revision"] = -1 | |
| with self.assertRaises(ValueError): | |
| EngineeringState.from_snapshot(snapshot) | |
| def test_canary_selection_is_deterministic_and_requires_session(self) -> None: | |
| config = EngineeringStateConfig(EngineeringStateMode.CANARY, 0.5) | |
| self.assertFalse(config.selects_canary("run", "")) | |
| self.assertEqual( | |
| config.selects_canary("run", "session"), | |
| config.selects_canary("run", "session"), | |
| ) | |
| def test_resume_normalizes_terminal_state_and_preserves_history(self) -> None: | |
| state = EngineeringState.start("resume this task", run_id="run-4", session_id="session-4") | |
| for target in ("CLASSIFYING", "THINKING", "COMPLETED"): | |
| state.transition(target) | |
| history_before_resume = list(state.history) | |
| state.prepare_for_resume() | |
| self.assertEqual(state.current_state, "IDLE") | |
| self.assertEqual(state.status, "active") | |
| self.assertEqual(state.history[:len(history_before_resume)], history_before_resume) | |
| self.assertTrue(any("resume normalized state to IDLE" in item for item in state.diagnostics)) | |
| if __name__ == "__main__": | |
| unittest.main(verbosity=2) | |