Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """ | |
| Test script for OFP endpoint | |
| Tests getManifests and utterance events | |
| """ | |
| import requests | |
| import json | |
| from datetime import datetime, timezone | |
| import uuid | |
| # OFP endpoint URL | |
| OFP_URL = "https://bladeszasza-talker.hf.space/api/ofp" | |
| MANIFEST_URL = "https://bladeszasza-talker.hf.space/api/manifest" | |
| def test_manifest_endpoint(): | |
| """Test the /manifest endpoint""" | |
| print("\n" + "="*60) | |
| print("TEST 1: GET /manifest endpoint") | |
| print("="*60) | |
| try: | |
| response = requests.get(MANIFEST_URL, timeout=10) | |
| print(f"Status Code: {response.status_code}") | |
| print(f"Response:\n{json.dumps(response.json(), indent=2)}") | |
| return response.status_code == 200 | |
| except Exception as e: | |
| print(f"❌ Error: {e}") | |
| return False | |
| def test_get_manifests_event(): | |
| """Test getManifests event via OFP endpoint""" | |
| print("\n" + "="*60) | |
| print("TEST 2: POST /ofp - getManifests event") | |
| print("="*60) | |
| envelope = { | |
| "openFloor": { | |
| "schema": {"version": "1.0.0"}, | |
| "conversation": {"id": f"conv:test-{uuid.uuid4()}"}, | |
| "sender": { | |
| "speakerUri": "tag:test,2025:tester", | |
| "serviceUrl": "https://test.example.com/ofp" | |
| }, | |
| "events": [{ | |
| "eventType": "getManifests", | |
| "to": { | |
| "speakerUri": "tag:talker.service,2025:agent-01" | |
| }, | |
| "parameters": {} | |
| }] | |
| } | |
| } | |
| print(f"Sending envelope:\n{json.dumps(envelope, indent=2)}") | |
| try: | |
| response = requests.post( | |
| OFP_URL, | |
| json=envelope, | |
| headers={"Content-Type": "application/json"}, | |
| timeout=30 | |
| ) | |
| print(f"\nStatus Code: {response.status_code}") | |
| print(f"Response:\n{json.dumps(response.json(), indent=2)}") | |
| # Verify response structure | |
| resp_data = response.json() | |
| if "openFloor" in resp_data: | |
| events = resp_data["openFloor"].get("events", []) | |
| if events and events[0].get("eventType") == "publishManifest": | |
| print("\n✅ SUCCESS: Received publishManifest response") | |
| return True | |
| else: | |
| print("\n❌ FAIL: Expected publishManifest event") | |
| return False | |
| else: | |
| print("\n❌ FAIL: Invalid response structure") | |
| return False | |
| except Exception as e: | |
| print(f"\n❌ Error: {e}") | |
| return False | |
| def test_utterance_event(): | |
| """Test utterance event via OFP endpoint""" | |
| print("\n" + "="*60) | |
| print("TEST 3: POST /ofp - utterance event") | |
| print("="*60) | |
| test_message = "Hello! Can you hear me?" | |
| envelope = { | |
| "openFloor": { | |
| "schema": {"version": "1.0.0"}, | |
| "conversation": {"id": f"conv:test-{uuid.uuid4()}"}, | |
| "sender": { | |
| "speakerUri": "tag:test,2025:tester", | |
| "serviceUrl": "https://test.example.com/ofp" | |
| }, | |
| "events": [{ | |
| "eventType": "utterance", | |
| "to": { | |
| "speakerUri": "tag:talker.service,2025:agent-01" | |
| }, | |
| "parameters": { | |
| "dialogEvent": { | |
| "id": f"de:{uuid.uuid4()}", | |
| "speakerUri": "tag:test,2025:tester", | |
| "span": { | |
| "startTime": datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z') | |
| }, | |
| "features": { | |
| "text": { | |
| "mimeType": "text/plain", | |
| "tokens": [{ | |
| "value": test_message, | |
| "confidence": 1.0 | |
| }] | |
| } | |
| } | |
| } | |
| } | |
| }] | |
| } | |
| } | |
| print(f"Sending envelope with message: '{test_message}'") | |
| print(f"Full envelope:\n{json.dumps(envelope, indent=2)}") | |
| try: | |
| response = requests.post( | |
| OFP_URL, | |
| json=envelope, | |
| headers={"Content-Type": "application/json"}, | |
| timeout=60 # Longer timeout for LLM generation | |
| ) | |
| print(f"\nStatus Code: {response.status_code}") | |
| print(f"Response:\n{json.dumps(response.json(), indent=2)}") | |
| # Verify response structure | |
| resp_data = response.json() | |
| if "openFloor" in resp_data: | |
| events = resp_data["openFloor"].get("events", []) | |
| if events and events[0].get("eventType") == "utterance": | |
| # Extract the response text | |
| params = events[0].get("parameters", {}) | |
| dialog_event = params.get("dialogEvent", {}) | |
| features = dialog_event.get("features", {}) | |
| text_feature = features.get("text", {}) | |
| tokens = text_feature.get("tokens", []) | |
| if tokens: | |
| response_text = tokens[0].get("value", "") | |
| print(f"\n✅ SUCCESS: Received utterance response") | |
| print(f"Agent Response: '{response_text}'") | |
| return True | |
| else: | |
| print("\n❌ FAIL: No tokens in response") | |
| return False | |
| else: | |
| print("\n⚠️ WARNING: Received acknowledgment (empty events array)") | |
| print("This may be expected if auto_respond is disabled") | |
| return True # Still consider it a pass | |
| else: | |
| print("\n❌ FAIL: Invalid response structure") | |
| return False | |
| except Exception as e: | |
| print(f"\n❌ Error: {e}") | |
| return False | |
| def test_complex_utterance(): | |
| """Test a more complex utterance to verify LLM integration""" | |
| print("\n" + "="*60) | |
| print("TEST 4: POST /ofp - Complex utterance (LLM test)") | |
| print("="*60) | |
| test_message = "What is the Open Floor Protocol and why is it important?" | |
| envelope = { | |
| "openFloor": { | |
| "schema": {"version": "1.0.0"}, | |
| "conversation": {"id": f"conv:test-{uuid.uuid4()}"}, | |
| "sender": { | |
| "speakerUri": "tag:test,2025:tester", | |
| "serviceUrl": "https://test.example.com/ofp" | |
| }, | |
| "events": [{ | |
| "eventType": "utterance", | |
| "to": { | |
| "speakerUri": "tag:talker.service,2025:agent-01" | |
| }, | |
| "parameters": { | |
| "dialogEvent": { | |
| "id": f"de:{uuid.uuid4()}", | |
| "speakerUri": "tag:test,2025:tester", | |
| "span": { | |
| "startTime": datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z') | |
| }, | |
| "features": { | |
| "text": { | |
| "mimeType": "text/plain", | |
| "tokens": [{ | |
| "value": test_message, | |
| "confidence": 1.0 | |
| }] | |
| } | |
| } | |
| } | |
| } | |
| }] | |
| } | |
| } | |
| print(f"Sending complex question: '{test_message}'") | |
| try: | |
| response = requests.post( | |
| OFP_URL, | |
| json=envelope, | |
| headers={"Content-Type": "application/json"}, | |
| timeout=60 | |
| ) | |
| print(f"\nStatus Code: {response.status_code}") | |
| resp_data = response.json() | |
| if "openFloor" in resp_data: | |
| events = resp_data["openFloor"].get("events", []) | |
| if events and events[0].get("eventType") == "utterance": | |
| params = events[0].get("parameters", {}) | |
| dialog_event = params.get("dialogEvent", {}) | |
| features = dialog_event.get("features", {}) | |
| text_feature = features.get("text", {}) | |
| tokens = text_feature.get("tokens", []) | |
| if tokens: | |
| response_text = tokens[0].get("value", "") | |
| print(f"\n✅ SUCCESS: LLM generated response") | |
| print(f"Agent Response ({len(response_text)} chars):") | |
| print(f"'{response_text[:500]}{'...' if len(response_text) > 500 else ''}'") | |
| return True | |
| print("\n⚠️ No utterance response received") | |
| return True # Still pass | |
| except Exception as e: | |
| print(f"\n❌ Error: {e}") | |
| return False | |
| def main(): | |
| """Run all tests""" | |
| print("\n" + "="*60) | |
| print("🧪 OFP ENDPOINT TEST SUITE") | |
| print("="*60) | |
| print(f"Target URL: {OFP_URL}") | |
| print(f"Manifest URL: {MANIFEST_URL}") | |
| results = { | |
| "Manifest Endpoint": test_manifest_endpoint(), | |
| "GetManifests Event": test_get_manifests_event(), | |
| "Utterance Event": test_utterance_event(), | |
| "Complex Utterance": test_complex_utterance() | |
| } | |
| # Summary | |
| print("\n" + "="*60) | |
| print("📊 TEST RESULTS SUMMARY") | |
| print("="*60) | |
| for test_name, passed in results.items(): | |
| status = "✅ PASS" if passed else "❌ FAIL" | |
| print(f"{status} - {test_name}") | |
| total = len(results) | |
| passed = sum(results.values()) | |
| print(f"\n{passed}/{total} tests passed ({passed/total*100:.1f}%)") | |
| print("="*60 + "\n") | |
| return all(results.values()) | |
| if __name__ == "__main__": | |
| success = main() | |
| exit(0 if success else 1) |