270 lines
11 KiB
Python
270 lines
11 KiB
Python
import pytest
|
|
from unittest.mock import patch, Mock
|
|
from app import create_app, db
|
|
from app.models import User, Folder, AIRuleCache
|
|
from app.ai_service import AIService
|
|
|
|
class TestAIRuleEndpoints:
|
|
"""Test cases for AI rule generation API endpoints."""
|
|
|
|
@pytest.fixture
|
|
def app(self):
|
|
"""Create and configure a test app."""
|
|
app = create_app('testing')
|
|
|
|
with app.app_context():
|
|
db.create_all()
|
|
yield app
|
|
db.drop_all()
|
|
|
|
@pytest.fixture
|
|
def client(self, app):
|
|
"""Create a test client."""
|
|
return app.test_client()
|
|
|
|
@pytest.fixture
|
|
def user(self, app):
|
|
"""Create a test user."""
|
|
with app.app_context():
|
|
user = User(
|
|
first_name='Test',
|
|
last_name='User',
|
|
email='test@example.com',
|
|
password_hash='hashed_password'
|
|
)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
# Refresh the user to ensure it's attached to the session
|
|
db.session.refresh(user)
|
|
return user
|
|
|
|
@pytest.fixture
|
|
def authenticated_client(self, client, user):
|
|
"""Create a test client with authenticated user."""
|
|
with client.session_transaction() as sess:
|
|
sess['_user_id'] = str(user.id)
|
|
sess['_fresh'] = True
|
|
return client
|
|
|
|
def test_generate_rule_success(self, authenticated_client, user):
|
|
"""Test successful rule generation."""
|
|
with patch('app.routes.folders.ai_service') as mock_ai_service:
|
|
# Mock AI service response
|
|
mock_ai_service.generate_multiple_rules.return_value = (
|
|
[{'text': "Move emails from 'boss@company.com' to this folder", 'quality_score': 85}],
|
|
{'total_generated': 1}
|
|
)
|
|
|
|
response = authenticated_client.post('/api/folders/generate-rule', data={
|
|
'name': 'Work',
|
|
'folder_type': 'destination',
|
|
'rule_type': 'multiple'
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
# Check that the response contains HTML
|
|
assert b'generated-rule-text' in response.data
|
|
assert b'Move emails from' in response.data
|
|
|
|
def test_generate_rule_missing_folder_name(self, authenticated_client, user):
|
|
"""Test rule generation with missing folder name."""
|
|
response = authenticated_client.post('/api/folders/generate-rule', data={
|
|
'folder_type': 'destination',
|
|
'rule_type': 'multiple'
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
# Should return error in HTML format
|
|
assert b'Folder name is required' in response.data
|
|
|
|
def test_generate_rule_invalid_folder_type(self, authenticated_client, user):
|
|
"""Test rule generation with invalid folder type."""
|
|
response = authenticated_client.post('/api/folders/generate-rule', data={
|
|
'name': 'Work',
|
|
'folder_type': 'invalid',
|
|
'rule_type': 'multiple'
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
# Should return error in HTML format
|
|
assert b'Invalid folder type' in response.data
|
|
|
|
def test_generate_rule_multiple_options(self, authenticated_client, user):
|
|
"""Test multiple rule options generation."""
|
|
with patch('app.routes.folders.ai_service') as mock_ai_service:
|
|
# Mock AI service response
|
|
mock_ai_service.generate_multiple_rules.return_value = (
|
|
[
|
|
{'text': 'Rule 1', 'quality_score': 85},
|
|
{'text': 'Rule 2', 'quality_score': 75}
|
|
],
|
|
{'total_generated': 2}
|
|
)
|
|
|
|
response = authenticated_client.post('/api/folders/generate-rule', data={
|
|
'name': 'Work',
|
|
'folder_type': 'destination',
|
|
'rule_type': 'multiple'
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
# Check that multiple rules are displayed
|
|
assert b'Rule 1' in response.data
|
|
assert b'Rule 2' in response.data
|
|
|
|
def test_generate_rule_ai_service_failure(self, authenticated_client, user):
|
|
"""Test rule generation when AI service fails."""
|
|
with patch('app.routes.folders.ai_service') as mock_ai_service:
|
|
# Mock AI service failure
|
|
mock_ai_service.generate_multiple_rules.return_value = (None, {'error': 'Service unavailable'})
|
|
mock_ai_service.get_fallback_rule.return_value = 'Fallback rule'
|
|
|
|
response = authenticated_client.post('/api/folders/generate-rule', data={
|
|
'name': 'Work',
|
|
'folder_type': 'destination',
|
|
'rule_type': 'multiple'
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
# Should return fallback rule
|
|
assert b'Fallback rule' in response.data
|
|
assert b'AI service unavailable' in response.data
|
|
|
|
def test_assess_rule_success(self, authenticated_client, user):
|
|
"""Test successful rule assessment."""
|
|
with patch('app.routes.folders.ai_service') as mock_ai_service:
|
|
# Mock AI service response
|
|
mock_ai_service.assess_rule_quality.return_value = {
|
|
'score': 85,
|
|
'grade': 'good',
|
|
'feedback': 'Good rule with room for improvement'
|
|
}
|
|
|
|
response = authenticated_client.post('/api/folders/assess-rule', data={
|
|
'rule_text': 'Move emails from boss@company.com to this folder',
|
|
'folder_name': 'Work',
|
|
'folder_type': 'destination'
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
# Check that the response contains HTML
|
|
assert b'generated-rule-text' in response.data
|
|
assert b'85%' in response.data
|
|
|
|
def test_assess_rule_missing_inputs(self, authenticated_client, user):
|
|
"""Test rule assessment with missing inputs."""
|
|
response = authenticated_client.post('/api/folders/assess-rule', data={
|
|
'folder_name': 'Work',
|
|
'folder_type': 'destination'
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
# Should return error in HTML format
|
|
assert b'Rule text is required' in response.data
|
|
|
|
def test_cache_functionality(self, authenticated_client, user):
|
|
"""Test rule caching functionality."""
|
|
with patch('app.routes.folders.ai_service') as mock_ai_service:
|
|
# Mock AI service response
|
|
mock_ai_service.generate_multiple_rules.return_value = (
|
|
[{'text': "Cached rule", 'quality_score': 90}],
|
|
{'total_generated': 1}
|
|
)
|
|
|
|
# First request - should generate new rule
|
|
response1 = authenticated_client.post('/api/folders/generate-rule', data={
|
|
'name': 'Work',
|
|
'folder_type': 'destination',
|
|
'rule_type': 'multiple'
|
|
})
|
|
|
|
assert response1.status_code == 200
|
|
assert b'Cached rule' in response1.data
|
|
|
|
# Verify cache entry was created
|
|
cache_entry = AIRuleCache.query.filter_by(
|
|
user_id=user.id,
|
|
folder_name='Work',
|
|
folder_type='destination'
|
|
).first()
|
|
assert cache_entry is not None
|
|
assert cache_entry.rule_text == 'Cached rule'
|
|
|
|
# Second request - should use cached rule
|
|
with patch('app.routes.folders.ai_service') as mock_ai_service:
|
|
response2 = authenticated_client.post('/api/folders/generate-rule', data={
|
|
'name': 'Work',
|
|
'folder_type': 'destination',
|
|
'rule_type': 'multiple'
|
|
})
|
|
|
|
assert response2.status_code == 200
|
|
assert b'Using cached rule' in response2.data
|
|
|
|
def test_cache_expiration(self, authenticated_client, user):
|
|
"""Test cache expiration functionality."""
|
|
with patch('app.routes.folders.ai_service') as mock_ai_service:
|
|
# Mock AI service response
|
|
mock_ai_service.generate_multiple_rules.return_value = (
|
|
[{'text': "Expired rule", 'quality_score': 90}],
|
|
{'total_generated': 1}
|
|
)
|
|
|
|
# Create expired cache entry
|
|
from datetime import datetime, timedelta
|
|
expired_entry = AIRuleCache(
|
|
user_id=user.id,
|
|
folder_name='Work',
|
|
folder_type='destination',
|
|
rule_text='Expired rule',
|
|
rule_metadata={'quality_score': 90},
|
|
cache_key='test-key',
|
|
expires_at=datetime.utcnow() - timedelta(hours=1),
|
|
is_active=True
|
|
)
|
|
db.session.add(expired_entry)
|
|
db.session.commit()
|
|
|
|
# Request should generate new rule despite cache entry
|
|
response = authenticated_client.post('/api/folders/generate-rule', data={
|
|
'name': 'Work',
|
|
'folder_type': 'destination',
|
|
'rule_type': 'multiple'
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
# Should not show cached message
|
|
assert b'Using cached rule' not in response.data
|
|
|
|
def test_unauthorized_access(self, client):
|
|
"""Test unauthorized access to AI rule endpoints."""
|
|
response = client.post('/api/folders/generate-rule', data={
|
|
'folder_name': 'Work',
|
|
'folder_type': 'destination',
|
|
'rule_type': 'multiple'
|
|
})
|
|
|
|
# Should be redirected to login
|
|
assert response.status_code == 302
|
|
|
|
def test_database_error_handling(self, authenticated_client, user):
|
|
"""Test handling of database errors."""
|
|
with patch('app.routes.folders.db.session.commit') as mock_commit:
|
|
# Mock database commit failure
|
|
mock_commit.side_effect = Exception("Database error")
|
|
|
|
with patch('app.routes.folders.ai_service') as mock_ai_service:
|
|
mock_ai_service.generate_multiple_rules.return_value = (
|
|
[{'text': "Test rule", 'quality_score': 85}],
|
|
{'total_generated': 1}
|
|
)
|
|
|
|
response = authenticated_client.post('/api/folders/generate-rule', data={
|
|
'name': 'Work',
|
|
'folder_type': 'destination',
|
|
'rule_type': 'multiple'
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
# Should return error message
|
|
assert b'An unexpected error occurred' in response.data |