Spaces:
Sleeping
Sleeping
| """ | |
| Unit tests for OFP client | |
| """ | |
| import unittest | |
| from unittest.mock import Mock, patch, MagicMock | |
| from src.ofp_client import OFPClient | |
| from src.models import Envelope | |
| class TestOFPClient(unittest.TestCase): | |
| """Test cases for OFPClient class""" | |
| def setUp(self): | |
| """Set up test fixtures""" | |
| self.client = OFPClient( | |
| speaker_uri="tag:test,2025:sentinel", | |
| service_url="http://test.com", | |
| manifest={"test": "manifest"} | |
| ) | |
| def test_initialization(self): | |
| """Test client initialization""" | |
| self.assertEqual(self.client.speaker_uri, "tag:test,2025:sentinel") | |
| self.assertEqual(self.client.service_url, "http://test.com") | |
| self.assertIsNotNone(self.client.manifest) | |
| def test_send_envelope_success(self, mock_post): | |
| """Test successful envelope sending""" | |
| mock_response = Mock() | |
| mock_response.status_code = 200 | |
| mock_post.return_value = mock_response | |
| envelope = Envelope( | |
| schema={"version": "1.0.0"}, | |
| conversation={"id": "test"}, | |
| sender={"speakerUri": "tag:test,2025:sentinel"}, | |
| events=[] | |
| ) | |
| result = self.client.send_envelope("http://recipient.com", envelope) | |
| self.assertTrue(result) | |
| mock_post.assert_called_once() | |
| def test_send_envelope_failure(self, mock_post): | |
| """Test envelope sending failure""" | |
| mock_post.side_effect = Exception("Network error") | |
| envelope = Envelope( | |
| schema={"version": "1.0.0"}, | |
| conversation={"id": "test"}, | |
| sender={"speakerUri": "tag:test,2025:sentinel"}, | |
| events=[] | |
| ) | |
| result = self.client.send_envelope("http://recipient.com", envelope) | |
| self.assertFalse(result) | |
| def test_send_envelope_timeout(self, mock_post): | |
| """Test envelope sending timeout""" | |
| import requests | |
| mock_post.side_effect = requests.exceptions.Timeout() | |
| envelope = Envelope( | |
| schema={"version": "1.0.0"}, | |
| conversation={"id": "test"}, | |
| sender={"speakerUri": "tag:test,2025:sentinel"}, | |
| events=[] | |
| ) | |
| result = self.client.send_envelope("http://recipient.com", envelope) | |
| self.assertFalse(result) | |
| def test_send_private_alert(self, mock_post): | |
| """Test sending private alert to convener""" | |
| mock_response = Mock() | |
| mock_response.status_code = 200 | |
| mock_post.return_value = mock_response | |
| alert_data = { | |
| "alertType": "content_violation", | |
| "severity": "high", | |
| "message": "Test alert" | |
| } | |
| result = self.client.send_private_alert( | |
| convener_uri="tag:convener,2025:test", | |
| convener_url="http://convener.com", | |
| conversation_id="conv:123", | |
| alert_data=alert_data | |
| ) | |
| self.assertTrue(result) | |
| mock_post.assert_called_once() | |
| # Verify the envelope structure | |
| call_args = mock_post.call_args | |
| payload = call_args[1]['json'] | |
| self.assertIn('openFloor', payload) | |
| self.assertEqual(len(payload['openFloor']['events']), 1) | |
| event = payload['openFloor']['events'][0] | |
| self.assertEqual(event['eventType'], 'utterance') | |
| self.assertTrue(event['to']['private']) | |
| def test_send_public_message(self, mock_post): | |
| """Test sending public message""" | |
| mock_response = Mock() | |
| mock_response.status_code = 200 | |
| mock_post.return_value = mock_response | |
| result = self.client.send_public_message( | |
| conversation_id="conv:123", | |
| recipient_url="http://recipient.com", | |
| text="Hello everyone" | |
| ) | |
| self.assertTrue(result) | |
| mock_post.assert_called_once() | |
| def test_get_manifest(self): | |
| """Test manifest retrieval""" | |
| manifest = self.client.get_manifest() | |
| self.assertEqual(manifest, {"test": "manifest"}) | |
| if __name__ == '__main__': | |
| unittest.main() | |