"""Performance tests for Reklamator Tests performance requirements from spec.md: - SC-012: System handles 100 concurrent feedback submissions - SC-008: Dashboard loads 1000 feedback items in less than 3 seconds """ import os import time import pytest import threading from concurrent.futures import ThreadPoolExecutor, as_completed from app.services.feedback_storage import FeedbackStorageService from app.models.product import Product from io import BytesIO @pytest.mark.performance def test_concurrent_submissions(client, temp_data_dir, sample_product): """Test handling 100 concurrent submissions (SC-012) This test verifies the system can handle high concurrent load without errors or data corruption. """ product_slug = sample_product.submission_url_slug num_submissions = 100 success_count = 0 error_count = 0 submission_times = [] def submit_feedback(thread_id): """Submit a single feedback item""" start_time = time.time() try: response = client.post( f'/submit/{product_slug}', data={ 'feedback_text': f'Concurrent test feedback #{thread_id}', 'csrf_token': 'test_csrf_token' }, follow_redirects=False ) elapsed = time.time() - start_time return response.status_code, elapsed except Exception as e: print(f"Error in thread {thread_id}: {e}") return 500, 0 # Execute concurrent submissions start_time = time.time() with ThreadPoolExecutor(max_workers=20) as executor: futures = [executor.submit(submit_feedback, i) for i in range(num_submissions)] for future in as_completed(futures): status_code, elapsed = future.result() submission_times.append(elapsed) if status_code in [200, 302]: # Success or redirect success_count += 1 else: error_count += 1 total_time = time.time() - start_time # Calculate statistics avg_time = sum(submission_times) / len(submission_times) max_time = max(submission_times) min_time = min(submission_times) print(f"\n=== Concurrent Submission Test Results ===") print(f"Total submissions: {num_submissions}") print(f"Successful: {success_count}") print(f"Failed: {error_count}") print(f"Total time: {total_time:.2f}s") print(f"Throughput: {num_submissions / total_time:.2f} submissions/second") print(f"Average response time: {avg_time:.3f}s") print(f"Min response time: {min_time:.3f}s") print(f"Max response time: {max_time:.3f}s") # Assertions assert success_count >= 95, f"Too many failures: {error_count} out of {num_submissions}" assert avg_time < 5.0, f"Average response time too high: {avg_time:.2f}s" # Verify data integrity - check that submissions were actually saved feedback_dir = os.path.join(temp_data_dir, 'products', sample_product.product_id, 'feedback') if os.path.exists(feedback_dir): saved_count = len([d for d in os.listdir(feedback_dir) if os.path.isdir(os.path.join(feedback_dir, d))]) print(f"Feedback items saved: {saved_count}") assert saved_count >= 95, f"Not all submissions were saved: {saved_count} out of {num_submissions}" @pytest.mark.performance def test_dashboard_load_performance(client, temp_data_dir, sample_product, auth_user): """Test dashboard loads 1000 items in <3 seconds (SC-008) This test creates 1000 feedback items and measures dashboard load time. """ # Login first client.post('/login', data={ 'username': auth_user.username, 'password': 'admin123', 'csrf_token': 'test_csrf_token' }) # Create 1000 feedback items print("\n=== Creating 1000 feedback items for performance test ===") create_start = time.time() for i in range(1000): FeedbackStorageService.save_complete_feedback( product_id=sample_product.product_id, content_text=f"Performance test feedback #{i}", files=None ) if (i + 1) % 100 == 0: print(f"Created {i + 1} items...") create_time = time.time() - create_start print(f"Creation completed in {create_time:.2f}s") # Measure dashboard load time (cold load - first request) print("\n=== Testing dashboard load time ===") start_time = time.time() response = client.get('/dashboard') cold_load_time = time.time() - start_time assert response.status_code == 200 print(f"Cold load time (first request): {cold_load_time:.3f}s") # Measure warm load time (subsequent requests) warm_times = [] for i in range(3): start_time = time.time() response = client.get('/dashboard') elapsed = time.time() - start_time warm_times.append(elapsed) print(f"Warm load time (request {i+2}): {elapsed:.3f}s") avg_warm_time = sum(warm_times) / len(warm_times) print(f"Average warm load time: {avg_warm_time:.3f}s") # Test with pagination (page 2) start_time = time.time() response = client.get('/dashboard?page=2') page2_time = time.time() - start_time print(f"Page 2 load time: {page2_time:.3f}s") # Test with filters start_time = time.time() response = client.get('/dashboard?status=new') filter_time = time.time() - start_time print(f"Filtered view load time: {filter_time:.3f}s") # Assertions - SC-008 requires <3 seconds for 1000 items assert cold_load_time < 3.0, f"Dashboard load time exceeds 3s: {cold_load_time:.2f}s" assert avg_warm_time < 3.0, f"Average warm load time exceeds 3s: {avg_warm_time:.2f}s" assert page2_time < 3.0, f"Page 2 load time exceeds 3s: {page2_time:.2f}s" assert filter_time < 3.0, f"Filtered view load time exceeds 3s: {filter_time:.2f}s" print(f"\n✓ All dashboard performance tests passed!") print(f"✓ Cold load: {cold_load_time:.3f}s < 3.0s") print(f"✓ Warm load: {avg_warm_time:.3f}s < 3.0s") print(f"✓ Pagination: {page2_time:.3f}s < 3.0s") print(f"✓ Filtering: {filter_time:.3f}s < 3.0s") @pytest.mark.performance def test_large_file_upload_performance(client, sample_product): """Test performance with maximum size file uploads Verifies system can handle 3x10MB files without timeout. """ product_slug = sample_product.submission_url_slug # Create 3 files of 10MB each (at the limit) file_size = 10 * 1024 * 1024 # 10MB files = [] for i in range(3): file_data = b'x' * file_size files.append( (BytesIO(file_data), f'large_file_{i}.txt') ) start_time = time.time() response = client.post( f'/submit/{product_slug}', data={ 'feedback_text': 'Testing large file upload performance', 'files': files, 'csrf_token': 'test_csrf_token' }, content_type='multipart/form-data', follow_redirects=False ) upload_time = time.time() - start_time print(f"\n=== Large File Upload Test ===") print(f"Total size: {3 * file_size / (1024*1024):.1f}MB") print(f"Upload time: {upload_time:.2f}s") print(f"Upload speed: {(3 * file_size / (1024*1024)) / upload_time:.2f}MB/s") # Should complete within reasonable time (30s for 30MB) assert upload_time < 30.0, f"Upload took too long: {upload_time:.2f}s" assert response.status_code in [200, 302], f"Upload failed with status {response.status_code}" print(f"✓ Large file upload completed successfully in {upload_time:.2f}s") @pytest.mark.performance def test_rapid_sequential_submissions(client, sample_product): """Test rapid sequential submissions from single client Verifies rate limiting works correctly. """ product_slug = sample_product.submission_url_slug num_submissions = 15 # More than the rate limit (10/hour) success_count = 0 rate_limited_count = 0 print(f"\n=== Rapid Sequential Submission Test ===") for i in range(num_submissions): response = client.post( f'/submit/{product_slug}', data={ 'feedback_text': f'Rapid submission #{i}', 'csrf_token': 'test_csrf_token' }, follow_redirects=False ) if response.status_code in [200, 302]: success_count += 1 elif response.status_code == 429: # Too Many Requests rate_limited_count += 1 print(f"Rate limited at submission {i + 1}") print(f"Successful submissions: {success_count}") print(f"Rate limited: {rate_limited_count}") # Should allow at least the configured number (10) but then start rate limiting # Note: In testing, rate limiting might be disabled assert success_count > 0, "No submissions succeeded" print(f"✓ Rate limiting test completed (success: {success_count}, limited: {rate_limited_count})")