Test email
" + ) + + mock_smtp.assert_called_once() + +@patch('app.core.config.settings.SMTP_HOST', 'localhost') +def test_email_configuration(): + """Test email configuration settings.""" + from app.core.config import settings + assert settings.SMTP_HOST == 'localhost' + +def test_email_template_rendering(): + """Test email template rendering with variables.""" + template_data = { + "username": "Test User", + "reset_link": "https://example.com/reset" + } + + # Mock template rendering + with patch('app.email_templates.render_template') as mock_render: + mock_render.return_value = "Hello Test User
" + result = mock_render(template_data) + assert "Test User" in result + +def test_email_sending_failure(): + """Test handling email sending failures.""" + with patch('smtplib.SMTP') as mock_smtp: + mock_smtp.side_effect = Exception("SMTP connection failed") + + result = send_email("test@example.com", "Test", "Test
") + # Should handle the error gracefully + assert result is False or result is None diff --git a/backend/tests/test_error_handling.py b/backend/tests/test_error_handling.py new file mode 100644 index 0000000000..c9dbbdc281 --- /dev/null +++ b/backend/tests/test_error_handling.py @@ -0,0 +1,42 @@ +import pytest +from unittest.mock import patch, MagicMock +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session +from app.models import User + +def test_database_integrity_error(db: Session): + """Test database integrity constraints.""" + # Create first user + user1 = User(email="duplicate@example.com", hashed_password="test123") + db.add(user1) + db.commit() + + # Try to create duplicate email (should fail) + user2 = User(email="duplicate@example.com", hashed_password="test456") + db.add(user2) + + with pytest.raises(IntegrityError): + db.commit() + +def test_database_connection_error(): + """Test handling database connection errors.""" + with patch('app.core.db.SessionLocal') as mock_session: + mock_session.side_effect = Exception("Database connection failed") + + # Test that the error is properly handled + with pytest.raises(Exception): + mock_session() + +def test_transaction_rollback_on_error(db: Session): + """Test that transactions are properly rolled back on errors.""" + try: + user = User(email="error@example.com", hashed_password="test123") + db.add(user) + # Simulate an error + raise Exception("Simulated error") + except Exception: + db.rollback() + + # Verify the user was not saved + user_check = db.query(User).filter(User.email == "error@example.com").first() + assert user_check is None diff --git a/backend/tests/test_external_services.py b/backend/tests/test_external_services.py new file mode 100644 index 0000000000..e8ae540b04 --- /dev/null +++ b/backend/tests/test_external_services.py @@ -0,0 +1,51 @@ +import pytest +from unittest.mock import patch, MagicMock +from app.core.config import settings + +def test_external_api_integration(): + """Test integration with external APIs.""" + with patch('requests.get') as mock_get: + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"status": "ok"} + mock_get.return_value = mock_response + + # Simulate external API call + import requests + response = requests.get("https://api.example.com/status") + assert response.status_code == 200 + +def test_third_party_service_timeout(): + """Test handling of third-party service timeouts.""" + with patch('requests.get') as mock_get: + mock_get.side_effect = requests.exceptions.Timeout("Request timed out") + + try: + import requests + requests.get("https://api.example.com/status", timeout=5) + except requests.exceptions.Timeout: + # Should handle timeout gracefully + assert True + +def test_api_rate_limiting(): + """Test API rate limiting behavior.""" + with patch('requests.get') as mock_get: + mock_response = MagicMock() + mock_response.status_code = 429 # Too Many Requests + mock_get.return_value = mock_response + + import requests + response = requests.get("https://api.example.com/data") + assert response.status_code == 429 + +def test_service_health_check(): + """Test external service health checks.""" + with patch('requests.get') as mock_get: + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"healthy": True} + mock_get.return_value = mock_response + + import requests + response = requests.get("https://service.example.com/health") + assert response.json()["healthy"] is True diff --git a/backend/tests/test_file_handling.py b/backend/tests/test_file_handling.py new file mode 100644 index 0000000000..d381575af8 --- /dev/null +++ b/backend/tests/test_file_handling.py @@ -0,0 +1,49 @@ +import pytest +from fastapi.testclient import TestClient +from unittest.mock import patch, MagicMock +from app.main import app + +client = TestClient(app) + +def test_file_upload_validation(): + """Test file upload validation and security.""" + # Test file size limits + large_file_data = {"file": ("test.txt", b"x" * 1000000)} # 1MB file + response = client.post("/api/v1/upload/", files=large_file_data) + # Should handle large files appropriately + assert response.status_code in [200, 413, 422] + +def test_file_type_validation(): + """Test file type validation.""" + # Test with potentially dangerous file type + malicious_file = {"file": ("script.exe", b"MZ\x90\x00")} + response = client.post("/api/v1/upload/", files=malicious_file) + # Should reject executable files + assert response.status_code in [400, 422] + +def test_image_processing(): + """Test image file processing capabilities.""" + with patch('PIL.Image.open') as mock_image: + mock_img = MagicMock() + mock_img.size = (100, 100) + mock_image.return_value = mock_img + + # Simulate image upload + image_file = {"file": ("test.jpg", b"\xff\xd8\xff\xe0")} # JPEG header + response = client.post("/api/v1/upload/", files=image_file) + + # Should process image successfully + assert response.status_code in [200, 404] # 404 if endpoint doesn't exist + +def test_file_storage_integration(): + """Test file storage integration.""" + with patch('os.path.exists') as mock_exists: + with patch('builtins.open', create=True) as mock_open: + mock_exists.return_value = True + mock_file = MagicMock() + mock_open.return_value.__enter__.return_value = mock_file + + # Test file retrieval + response = client.get("/api/v1/files/test-file.txt") + # Should handle file retrieval appropriately + assert response.status_code in [200, 404] diff --git a/backend/tests/test_input_validation.py b/backend/tests/test_input_validation.py new file mode 100644 index 0000000000..662bc14787 --- /dev/null +++ b/backend/tests/test_input_validation.py @@ -0,0 +1,38 @@ +import pytest +from fastapi.testclient import TestClient +from app.main import app + +client = TestClient(app) + +def test_invalid_json_payload(): + """Test API response to invalid JSON payload.""" + headers = {"Content-Type": "application/json"} + response = client.post("/api/v1/login/access-token", + data="invalid json", headers=headers) + assert response.status_code == 422 + +def test_missing_required_fields(): + """Test API response when required fields are missing.""" + incomplete_data = {"email": "test@example.com"} # Missing password + response = client.post("/api/v1/login/access-token", json=incomplete_data) + assert response.status_code == 422 + +def test_field_validation_email(): + """Test email field validation.""" + invalid_data = {"email": "not-an-email", "password": "test123"} + response = client.post("/api/v1/users/", json=invalid_data) + assert response.status_code == 422 + +def test_field_validation_password_length(): + """Test password length validation.""" + short_password_data = {"email": "test@example.com", "password": "123"} + response = client.post("/api/v1/users/", json=short_password_data) + assert response.status_code == 422 + +def test_response_schema_validation(): + """Test that API responses match expected schema.""" + response = client.get("/api/v1/utils/health-check/") + assert response.status_code == 200 + data = response.json() + assert isinstance(data, dict) + assert "message" in data diff --git a/backend/tests/test_items_api.py b/backend/tests/test_items_api.py new file mode 100644 index 0000000000..2759c7dfb7 --- /dev/null +++ b/backend/tests/test_items_api.py @@ -0,0 +1,40 @@ +import pytest +from fastapi.testclient import TestClient +from app.main import app +from app.core.security import create_access_token + +client = TestClient(app) + +def test_create_item(): + """Test creating a new item.""" + token = create_access_token(subject="test@example.com") + headers = {"Authorization": f"Bearer {token}"} + + item_data = { + "title": "Test Item", + "description": "Test Description" + } + + response = client.post("/api/v1/items/", json=item_data, headers=headers) + assert response.status_code == 200 + data = response.json() + assert data["title"] == "Test Item" + +def test_get_items(): + """Test retrieving items list.""" + token = create_access_token(subject="test@example.com") + headers = {"Authorization": f"Bearer {token}"} + + response = client.get("/api/v1/items/", headers=headers) + assert response.status_code == 200 + data = response.json() + assert isinstance(data, list) + +def test_get_item_by_id(): + """Test retrieving a specific item by ID.""" + token = create_access_token(subject="test@example.com") + headers = {"Authorization": f"Bearer {token}"} + + response = client.get("/api/v1/items/1", headers=headers) + # Should return 200 if item exists, 404 if not + assert response.status_code in [200, 404] diff --git a/backend/tests/test_logging.py b/backend/tests/test_logging.py new file mode 100644 index 0000000000..7ed4d81fcc --- /dev/null +++ b/backend/tests/test_logging.py @@ -0,0 +1,42 @@ +import pytest +from fastapi.testclient import TestClient +from unittest.mock import patch +from app.main import app + +client = TestClient(app) + +def test_logging_configuration(): + """Test that logging is properly configured.""" + import logging + logger = logging.getLogger("app") + assert logger.level <= logging.INFO + +def test_request_logging(): + """Test that API requests are logged.""" + with patch('app.main.logger') as mock_logger: + response = client.get("/api/v1/utils/health-check/") + assert response.status_code == 200 + # Verify logging was called (implementation dependent) + +def test_error_logging(): + """Test that errors are properly logged.""" + with patch('app.main.logger') as mock_logger: + # Make a request that might cause an error + response = client.get("/api/v1/nonexistent-endpoint") + assert response.status_code == 404 + +def test_log_level_filtering(): + """Test log level filtering works correctly.""" + import logging + + # Test different log levels + logger = logging.getLogger("test") + logger.setLevel(logging.WARNING) + + with patch.object(logger, 'info') as mock_info: + with patch.object(logger, 'warning') as mock_warning: + logger.info("This should not be logged") + logger.warning("This should be logged") + + mock_info.assert_called_once() + mock_warning.assert_called_once() diff --git a/backend/tests/test_main.py b/backend/tests/test_main.py new file mode 100644 index 0000000000..262c12cdde --- /dev/null +++ b/backend/tests/test_main.py @@ -0,0 +1,28 @@ +import pytest +from fastapi.testclient import TestClient +from app.main import app + +client = TestClient(app) + +def test_health_check(): + """Test the health check endpoint.""" + response = client.get("/api/v1/utils/health-check/") + assert response.status_code == 200 + assert response.json() == {"message": "OK"} + +def test_root_endpoint(): + """Test the root endpoint redirect.""" + response = client.get("/") + # Should redirect to docs or return some response + assert response.status_code in [200, 307, 308] + +def test_docs_endpoint(): + """Test the OpenAPI docs endpoint.""" + response = client.get("/docs") + assert response.status_code == 200 + +def test_openapi_json(): + """Test the OpenAPI JSON schema endpoint.""" + response = client.get("/openapi.json") + assert response.status_code == 200 + assert response.headers["content-type"] == "application/json" diff --git a/backend/tests/test_models_validation.py b/backend/tests/test_models_validation.py new file mode 100644 index 0000000000..c94070013f --- /dev/null +++ b/backend/tests/test_models_validation.py @@ -0,0 +1,38 @@ +import pytest +from pydantic import ValidationError +from app.models import User, Item + +def test_user_model_validation(): + """Test user model field validation.""" + # Valid user data + valid_data = { + "email": "test@example.com", + "hashed_password": "hashedpass123", + "full_name": "Test User" + } + user = User(**valid_data) + assert user.email == "test@example.com" + +def test_user_model_invalid_email(): + """Test user model with invalid email.""" + with pytest.raises(ValidationError): + User(email="invalid-email", hashed_password="test123") + +def test_item_model_creation(): + """Test item model creation and validation.""" + item_data = { + "title": "Test Item", + "description": "Test Description", + "owner_id": 1 + } + item = Item(**item_data) + assert item.title == "Test Item" + assert item.owner_id == 1 + +def test_model_relationships(): + """Test model relationships and foreign keys.""" + user = User(email="owner@example.com", hashed_password="test123") + item = Item(title="User's Item", owner_id=1) + + assert item.owner_id == 1 + assert item.title == "User's Item" diff --git a/backend/tests/test_pagination.py b/backend/tests/test_pagination.py new file mode 100644 index 0000000000..630846091f --- /dev/null +++ b/backend/tests/test_pagination.py @@ -0,0 +1,40 @@ +import pytest +from fastapi.testclient import TestClient +from app.main import app +from app.core.security import create_access_token + +client = TestClient(app) + +def test_pagination_default_params(): + """Test pagination with default parameters.""" + token = create_access_token(subject="test@example.com") + headers = {"Authorization": f"Bearer {token}"} + + response = client.get("/api/v1/items/", headers=headers) + assert response.status_code == 200 + data = response.json() + assert isinstance(data, list) or "items" in data + +def test_pagination_custom_limit(): + """Test pagination with custom limit parameter.""" + token = create_access_token(subject="test@example.com") + headers = {"Authorization": f"Bearer {token}"} + + response = client.get("/api/v1/items/?limit=5", headers=headers) + assert response.status_code == 200 + +def test_pagination_skip_parameter(): + """Test pagination with skip parameter.""" + token = create_access_token(subject="test@example.com") + headers = {"Authorization": f"Bearer {token}"} + + response = client.get("/api/v1/items/?skip=10&limit=5", headers=headers) + assert response.status_code == 200 + +def test_pagination_invalid_parameters(): + """Test pagination with invalid parameters.""" + token = create_access_token(subject="test@example.com") + headers = {"Authorization": f"Bearer {token}"} + + response = client.get("/api/v1/items/?limit=-1", headers=headers) + assert response.status_code in [200, 422] # Should handle invalid params gracefully diff --git a/backend/tests/test_password_recovery.py b/backend/tests/test_password_recovery.py new file mode 100644 index 0000000000..d36ea8e328 --- /dev/null +++ b/backend/tests/test_password_recovery.py @@ -0,0 +1,41 @@ +import pytest +from fastapi.testclient import TestClient +from unittest.mock import patch, MagicMock +from app.main import app + +client = TestClient(app) + +def test_password_recovery_request(): + """Test requesting password recovery.""" + recovery_data = {"email": "test@example.com"} + response = client.post("/api/v1/password-recovery/", json=recovery_data) + assert response.status_code in [200, 404] + +@patch('app.utils.send_email') +def test_password_recovery_email_sent(mock_send_email): + """Test that password recovery email is sent.""" + mock_send_email.return_value = True + + recovery_data = {"email": "test@example.com"} + response = client.post("/api/v1/password-recovery/", json=recovery_data) + + if response.status_code == 200: + mock_send_email.assert_called_once() + +def test_reset_password_with_token(): + """Test resetting password with valid token.""" + reset_data = { + "token": "valid_reset_token", + "new_password": "newpassword123" + } + response = client.post("/api/v1/reset-password/", json=reset_data) + assert response.status_code in [200, 400] + +def test_reset_password_invalid_token(): + """Test resetting password with invalid token.""" + reset_data = { + "token": "invalid_token", + "new_password": "newpassword123" + } + response = client.post("/api/v1/reset-password/", json=reset_data) + assert response.status_code == 400 diff --git a/backend/tests/test_performance.py b/backend/tests/test_performance.py new file mode 100644 index 0000000000..09bb4b1938 --- /dev/null +++ b/backend/tests/test_performance.py @@ -0,0 +1,58 @@ +import pytest +from fastapi.testclient import TestClient +from unittest.mock import patch +from app.main import app +import time + +client = TestClient(app) + +def test_api_response_time(): + """Test API response time is within acceptable limits.""" + start_time = time.time() + response = client.get("/api/v1/utils/health-check/") + end_time = time.time() + + assert response.status_code == 200 + assert (end_time - start_time) < 2.0 # Should respond within 2 seconds + +def test_concurrent_requests(): + """Test handling of concurrent requests.""" + import concurrent.futures + import threading + + def make_request(): + return client.get("/api/v1/utils/health-check/") + + with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: + futures = [executor.submit(make_request) for _ in range(5)] + results = [future.result() for future in futures] + + # All requests should succeed + for result in results: + assert result.status_code == 200 + +def test_memory_usage_stability(): + """Test that repeated requests don't cause memory leaks.""" + import gc + + # Make multiple requests + for _ in range(10): + response = client.get("/api/v1/utils/health-check/") + assert response.status_code == 200 + + # Force garbage collection + gc.collect() + # This is a basic test - more sophisticated memory monitoring would be needed in practice +e +def test_database_connection_pool(): + """Test database connection pooling under load.""" + responses = [] + + # Make multiple database-dependent requests + for _ in range(5): + response = client.get("/api/v1/utils/health-check/") + responses.append(response) + + # All should succeed without connection pool exhaustion + for response in responses: + assert response.status_code == 200 diff --git a/backend/tests/test_security.py b/backend/tests/test_security.py new file mode 100644 index 0000000000..224a92b846 --- /dev/null +++ b/backend/tests/test_security.py @@ -0,0 +1,28 @@ +import pytest +from unittest.mock import patch, MagicMock +from app.core.security import create_access_token, verify_password, get_password_hash + +def test_create_access_token(): + """Test JWT token creation.""" + token = create_access_token(subject="test@example.com") + assert isinstance(token, str) + assert len(token) > 0 + +def test_verify_password_correct(): + """Test password verification with correct password.""" + plain_password = "testpassword" + hashed_password = get_password_hash(plain_password) + assert verify_password(plain_password, hashed_password) is True + +def test_verify_password_incorrect(): + """Test password verification with incorrect password.""" + plain_password = "testpassword" + hashed_password = get_password_hash(plain_password) + assert verify_password("wrongpassword", hashed_password) is False + +def test_get_password_hash(): + """Test password hashing.""" + password = "testpassword" + hashed = get_password_hash(password) + assert hashed != password + assert len(hashed) > 0 diff --git a/backend/tests/test_security_vulnerabilities.py b/backend/tests/test_security_vulnerabilities.py new file mode 100644 index 0000000000..3efbb02216 --- /dev/null +++ b/backend/tests/test_security_vulnerabilities.py @@ -0,0 +1,58 @@ +import pytest +from fastapi.testclient import TestClient +from app.main import app +from app.core.security import create_access_token + +client = TestClient(app) + +def test_sql_injection_prevention(): + """Test that SQL injection attempts are prevented.""" + token = create_access_token(subject="test@example.com") + headers = {"Authorization": f"Bearer {token}"} + + # Attempt SQL injection in query parameters + malicious_query = "1' OR '1'='1" + response = client.get(f"/api/v1/items/{malicious_query}", headers=headers) + + # Should return 404 or 422, not expose database errors + assert response.status_code in [404, 422] + +def test_xss_prevention(): + """Test XSS prevention in API responses.""" + token = create_access_token(subject="test@example.com") + headers = {"Authorization": f"Bearer {token}"} + + xss_payload = { + "title": "", + "description": "Test description" + } + + response = client.post("/api/v1/items/", json=xss_payload, headers=headers) + + if response.status_code == 200: + # Check that script tags are not returned as-is + data = response.json() + assert "