137 lines
No EOL
5 KiB
Python
137 lines
No EOL
5 KiB
Python
import unittest
|
|
from unittest.mock import patch, MagicMock
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import datetime
|
|
from video_processor import VideoProcessor
|
|
|
|
class TestWebhookIntegration(unittest.TestCase):
|
|
"""Test cases for webhook integration in VideoProcessor."""
|
|
|
|
def setUp(self):
|
|
"""Set up test environment."""
|
|
# Create a VideoProcessor instance with a mock API key
|
|
self.video_processor = VideoProcessor(api_key="test_api_key")
|
|
|
|
# Create a temporary file to simulate a video
|
|
self.temp_file = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
|
|
self.temp_file.close()
|
|
|
|
# Write some dummy data to the file
|
|
with open(self.temp_file.name, 'wb') as f:
|
|
f.write(b'test video content')
|
|
|
|
def tearDown(self):
|
|
"""Clean up after tests."""
|
|
# Remove the temporary file
|
|
if os.path.exists(self.temp_file.name):
|
|
os.unlink(self.temp_file.name)
|
|
|
|
@patch('video_processor.genai')
|
|
@patch('video_processor.requests.post')
|
|
def test_webhook_called_on_successful_processing(self, mock_post, mock_genai):
|
|
"""Test that the webhook is called when video processing is successful."""
|
|
# Mock the genai API responses
|
|
mock_file = MagicMock()
|
|
mock_file.uri = "test_uri"
|
|
mock_file.name = "test_name"
|
|
mock_file.state.name = "ACTIVE"
|
|
mock_genai.upload_file.return_value = mock_file
|
|
|
|
# Mock the generate_content response
|
|
mock_response = MagicMock()
|
|
mock_part = MagicMock()
|
|
mock_part.text = "Test response content"
|
|
mock_response.parts = [mock_part]
|
|
mock_genai.GenerativeModel.return_value.generate_content.return_value = mock_response
|
|
|
|
# Set up the mock for the requests.post call
|
|
mock_post.return_value.status_code = 200
|
|
|
|
# Test data
|
|
test_prompt = "Test prompt for video processing"
|
|
test_email = "test.user@example.com"
|
|
|
|
# Call the process_video method
|
|
result = self.video_processor.process_video(
|
|
self.temp_file.name,
|
|
test_prompt,
|
|
test_email
|
|
)
|
|
|
|
# Verify the result is successful
|
|
self.assertTrue(result["success"])
|
|
self.assertEqual(result["content"], "Test response content")
|
|
|
|
# Verify webhook was called with correct data
|
|
mock_post.assert_called_once()
|
|
|
|
# Get the arguments the mock was called with
|
|
call_args = mock_post.call_args
|
|
|
|
# Verify URL
|
|
self.assertEqual(call_args[0][0], "https://hook.us1.make.celonis.com/8ri1h8b2he4wudp2jku69mgcxumzxf3v")
|
|
|
|
# Verify headers
|
|
self.assertEqual(call_args[1]["headers"], {"Content-Type": "application/json"})
|
|
|
|
# Verify timeout
|
|
self.assertEqual(call_args[1]["timeout"], 10)
|
|
|
|
# Parse and verify the payload
|
|
payload = json.loads(call_args[1]["data"])
|
|
self.assertEqual(payload["tool"], "VIDEOQUERY")
|
|
self.assertEqual(payload["user"], test_email)
|
|
self.assertEqual(payload["model"], "GEMINI")
|
|
self.assertEqual(payload["prompt"], test_prompt)
|
|
|
|
# Verify date format (should be ISO format)
|
|
try:
|
|
datetime.datetime.fromisoformat(payload["date"])
|
|
date_valid = True
|
|
except ValueError:
|
|
date_valid = False
|
|
self.assertTrue(date_valid, "Date should be in ISO format")
|
|
|
|
@patch('video_processor.genai')
|
|
@patch('video_processor.requests.post')
|
|
def test_webhook_error_does_not_affect_processing(self, mock_post, mock_genai):
|
|
"""Test that errors in the webhook don't affect the main processing flow."""
|
|
# Mock the genai API responses
|
|
mock_file = MagicMock()
|
|
mock_file.uri = "test_uri"
|
|
mock_file.name = "test_name"
|
|
mock_file.state.name = "ACTIVE"
|
|
mock_genai.upload_file.return_value = mock_file
|
|
|
|
# Mock the generate_content response
|
|
mock_response = MagicMock()
|
|
mock_part = MagicMock()
|
|
mock_part.text = "Test response content"
|
|
mock_response.parts = [mock_part]
|
|
mock_genai.GenerativeModel.return_value.generate_content.return_value = mock_response
|
|
|
|
# Set up the mock for the requests.post call to raise an exception
|
|
mock_post.side_effect = Exception("Webhook connection error")
|
|
|
|
# Test data
|
|
test_prompt = "Test prompt for video processing"
|
|
test_email = "test.user@example.com"
|
|
|
|
# Call the process_video method
|
|
result = self.video_processor.process_video(
|
|
self.temp_file.name,
|
|
test_prompt,
|
|
test_email
|
|
)
|
|
|
|
# Verify the result is still successful despite webhook error
|
|
self.assertTrue(result["success"])
|
|
self.assertEqual(result["content"], "Test response content")
|
|
|
|
# Verify webhook was called
|
|
mock_post.assert_called_once()
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main() |