| from __future__ import annotations | |
| import unittest | |
| from turn_detection.runtime import ControllerConfig, Prediction, TurnController, TurnState | |
| class TurnControllerTest(unittest.TestCase): | |
| def setUp(self) -> None: | |
| self.controller = TurnController( | |
| ControllerConfig( | |
| endpoint_threshold=0.7, | |
| long_pause_threshold=0.4, | |
| min_silence_ms=200, | |
| relax_after_ms=800, | |
| max_silence_ms=1800, | |
| required_confirmations=2, | |
| ) | |
| ) | |
| def test_minimum_silence_always_holds(self) -> None: | |
| decision = self.controller.evaluate_pause(0.99, silence_ms=199) | |
| self.assertEqual(decision.state, TurnState.HOLD) | |
| self.assertEqual(decision.reason, "minimum_silence_not_reached") | |
| def test_endpoint_requires_configured_confirmations(self) -> None: | |
| first = self.controller.evaluate_pause(0.9, silence_ms=300, timestamp_ms=300) | |
| second = self.controller.evaluate_pause(0.9, silence_ms=500, timestamp_ms=500) | |
| self.assertEqual(first.state, TurnState.HOLD) | |
| self.assertEqual(second.state, TurnState.END) | |
| self.assertTrue(second.emit_response) | |
| def test_endpoint_latches_without_duplicate_response(self) -> None: | |
| self.controller.evaluate_pause(0.9, 300, timestamp_ms=300) | |
| endpoint = self.controller.evaluate_pause(0.9, 500, timestamp_ms=500) | |
| repeated = self.controller.evaluate_pause(0.01, 700, timestamp_ms=700) | |
| self.assertTrue(endpoint.emit_response) | |
| self.assertEqual(repeated.state, TurnState.END) | |
| self.assertEqual(repeated.reason, "endpoint_latched") | |
| self.assertFalse(repeated.emit_response) | |
| def test_new_speech_resets_endpoint(self) -> None: | |
| self.controller.evaluate_pause(0.9, 300) | |
| self.controller.evaluate_pause(0.9, 500) | |
| decision = self.controller.observe_speech(600) | |
| self.assertEqual(decision.state, TurnState.SPEAKING) | |
| def test_max_timeout_is_bounded(self) -> None: | |
| decision = self.controller.evaluate_pause(0.01, silence_ms=1800) | |
| self.assertEqual(decision.state, TurnState.END) | |
| self.assertEqual(decision.reason, "maximum_timeout") | |
| self.assertTrue(decision.emit_response) | |
| def test_threshold_relaxes_linearly(self) -> None: | |
| self.assertAlmostEqual(self.controller.threshold_for_silence(800), 0.7) | |
| self.assertAlmostEqual(self.controller.threshold_for_silence(1300), 0.55) | |
| self.assertAlmostEqual(self.controller.threshold_for_silence(1800), 0.4) | |
| def test_prediction_validation(self) -> None: | |
| with self.assertRaises(ValueError): | |
| Prediction(endpoint_probability=1.1) | |
| with self.assertRaises(ValueError): | |
| self.controller.evaluate_pause(-0.1, 300) | |
| def test_timestamps_must_be_monotonic(self) -> None: | |
| self.controller.observe_speech(100) | |
| with self.assertRaises(ValueError): | |
| self.controller.evaluate_pause(0.5, 300, timestamp_ms=99) | |
| if __name__ == "__main__": | |
| unittest.main() | |