Spaces:
Paused
Paused
| import pytest | |
| from fastapi.testclient import TestClient | |
| from unittest.mock import patch | |
| from constants import MAX_FILE_NAME_LENGTH | |
| from main import app # Adjust import based on your structure | |
| client = TestClient(app) | |
| def delete_with_body(url: str, json_data: dict): | |
| """Helper to send DELETE request with JSON body""" | |
| return client.request("DELETE", url, json=json_data) | |
| class TestDeleteFileEndpoint: | |
| """Test the DELETE /file endpoint""" | |
| def base_required_fields(self): | |
| """Base fields required by IdentifierBase and ProfileBase""" | |
| return { | |
| "user_id": "test-user-123", | |
| "participant_id": "participant-456", | |
| "session_id": "test-session-123", | |
| "consent": True, | |
| "age_group": "25-34", | |
| "gender": "M", | |
| "roles": ["patient"], | |
| } | |
| def valid_payload(self, base_required_fields): | |
| return {**base_required_fields, "file_name": "document.txt"} | |
| def mock_dependencies(self): | |
| """Mock external dependencies""" | |
| with ( | |
| patch("main.session_document_store") as mock_store, | |
| patch("main.replace_spaces_in_filename") as mock_replace, | |
| ): | |
| # Setup default behavior | |
| mock_replace.side_effect = lambda x: ( | |
| x | |
| ) # Return filename unchanged by default | |
| mock_store.delete_document.return_value = True | |
| yield {"store": mock_store, "replace_spaces": mock_replace} | |
| # ==================== Successful Deletion Tests ==================== | |
| def test_successful_file_deletion(self, valid_payload, mock_dependencies): | |
| """Test successful file deletion with valid inputs""" | |
| response = delete_with_body("/file", valid_payload) | |
| assert response.status_code == 200 | |
| # Verify workflow | |
| mock_dependencies["replace_spaces"].assert_called_once_with("document.txt") | |
| mock_dependencies["store"].delete_document.assert_called_once_with( | |
| "test-session-123", "document.txt" | |
| ) | |
| def test_delete_file_with_spaces_in_filename( | |
| self, base_required_fields, mock_dependencies | |
| ): | |
| """Test that spaces in filename are replaced""" | |
| payload = {**base_required_fields, "file_name": "my document.txt"} | |
| # Mock replace_spaces to return expected result | |
| mock_dependencies["replace_spaces"].side_effect = None | |
| mock_dependencies["replace_spaces"].return_value = "my_document.txt" | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 200 | |
| mock_dependencies["replace_spaces"].assert_called_once_with("my document.txt") | |
| mock_dependencies["store"].delete_document.assert_called_once_with( | |
| "test-session-123", "my_document.txt" | |
| ) | |
| def test_delete_file_different_filenames( | |
| self, base_required_fields, mock_dependencies | |
| ): | |
| """Test deleting files with various filename formats""" | |
| filenames = [ | |
| "document.txt", | |
| "report.pdf", | |
| "data.csv", | |
| "file_with_underscores.docx", | |
| "file-with-dashes.xlsx", | |
| "file.multiple.dots.txt", | |
| ] | |
| for filename in filenames: | |
| payload = {**base_required_fields, "file_name": filename} | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 200 | |
| def test_delete_file_different_session_ids( | |
| self, base_required_fields, mock_dependencies | |
| ): | |
| """Test deleting files from different sessions""" | |
| session_ids = [ | |
| "session-1", | |
| "session-2", | |
| "session_abc_123", | |
| "a1b2c3", | |
| ] | |
| for session_id in session_ids: | |
| payload = { | |
| **base_required_fields, | |
| "session_id": session_id, | |
| "file_name": "document.txt", | |
| } | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 200 | |
| # ==================== Request Validation Tests ==================== | |
| def test_delete_file_missing_session_id( | |
| self, base_required_fields, mock_dependencies | |
| ): | |
| """Test that missing session_id returns validation error""" | |
| payload = {**base_required_fields, "file_name": "document.txt"} | |
| del payload["session_id"] | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 422 | |
| # Store should not be called | |
| assert not mock_dependencies["store"].delete_document.called | |
| def test_delete_file_missing_file_name( | |
| self, base_required_fields, mock_dependencies | |
| ): | |
| """Test that missing file_name returns validation error""" | |
| payload = {**base_required_fields} | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 422 | |
| assert not mock_dependencies["store"].delete_document.called | |
| def test_delete_file_missing_both_fields(self, mock_dependencies): | |
| """Test that missing session_id and file_name returns validation error""" | |
| payload = { | |
| "user_id": "test-user", | |
| "participant_id": "participant-123", | |
| "consent": True, | |
| "age_group": "25-34", | |
| "gender": "M", | |
| "roles": ["patient"], | |
| } | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 422 | |
| assert not mock_dependencies["store"].delete_document.called | |
| def test_delete_file_empty_session_id( | |
| self, base_required_fields, mock_dependencies | |
| ): | |
| """Test handling of empty session_id""" | |
| payload = { | |
| **base_required_fields, | |
| "session_id": "", | |
| "file_name": "document.txt", | |
| } | |
| response = delete_with_body("/file", payload) | |
| # Empty string violates pattern and min_length | |
| assert response.status_code == 422 | |
| def test_delete_file_empty_file_name(self, base_required_fields, mock_dependencies): | |
| """Test handling of empty file_name""" | |
| payload = {**base_required_fields, "file_name": ""} | |
| response = delete_with_body("/file", payload) | |
| # Empty string violates min_length=1 | |
| assert response.status_code == 422 | |
| def test_delete_file_extra_fields_ignored( | |
| self, base_required_fields, mock_dependencies | |
| ): | |
| """Test that extra fields in payload are ignored""" | |
| payload = { | |
| **base_required_fields, | |
| "file_name": "document.txt", | |
| "extra_field": "should be ignored", | |
| "another_field": 123, | |
| } | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 200 | |
| mock_dependencies["store"].delete_document.assert_called_once() | |
| # ==================== Store Behavior Tests ==================== | |
| def test_delete_file_store_returns_true(self, valid_payload, mock_dependencies): | |
| """Test when store successfully deletes (returns True)""" | |
| mock_dependencies["store"].delete_document.return_value = True | |
| response = delete_with_body("/file", valid_payload) | |
| assert response.status_code == 200 | |
| def test_delete_file_store_returns_false(self, valid_payload, mock_dependencies): | |
| """Test when store deletion fails (returns False)""" | |
| mock_dependencies["store"].delete_document.return_value = False | |
| response = delete_with_body("/file", valid_payload) | |
| # Endpoint doesn't check return value, so still 200 | |
| assert response.status_code == 200 | |
| def test_delete_file_nonexistent_file( | |
| self, base_required_fields, mock_dependencies | |
| ): | |
| """Test deleting a file that doesn't exist""" | |
| payload = {**base_required_fields, "file_name": "nonexistent.txt"} | |
| # Store returns False for nonexistent file | |
| mock_dependencies["store"].delete_document.return_value = False | |
| response = delete_with_body("/file", payload) | |
| # Endpoint still returns 200 (idempotent DELETE) | |
| assert response.status_code == 200 | |
| def test_delete_file_nonexistent_session( | |
| self, base_required_fields, mock_dependencies | |
| ): | |
| """Test deleting from a session that doesn't exist""" | |
| payload = { | |
| **base_required_fields, | |
| "session_id": "nonexistent-session", | |
| "file_name": "document.txt", | |
| } | |
| mock_dependencies["store"].delete_document.return_value = False | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 200 | |
| # ==================== Filename Replacement Tests ==================== | |
| def test_replace_spaces_called_with_correct_argument( | |
| self, base_required_fields, mock_dependencies | |
| ): | |
| """Test that replace_spaces_in_filename is called with the right argument""" | |
| payload = {**base_required_fields, "file_name": "my file.txt"} | |
| delete_with_body("/file", payload) | |
| mock_dependencies["replace_spaces"].assert_called_once_with("my file.txt") | |
| # ==================== Rate Limiting Tests ==================== | |
| def test_invalid_filename_pattern_double_dots( | |
| self, base_required_fields, mock_dependencies | |
| ): | |
| """Test that filenames with double dots are rejected""" | |
| payload = {**base_required_fields, "file_name": "file..txt"} | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 422 | |
| def test_invalid_filename_pattern_starting_dot( | |
| self, base_required_fields, mock_dependencies | |
| ): | |
| """Test that filenames starting with dot are rejected""" | |
| payload = {**base_required_fields, "file_name": ".hidden.txt"} | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 422 | |
| def test_invalid_filename_pattern_starting_space( | |
| self, base_required_fields, mock_dependencies | |
| ): | |
| """Test that filenames starting with space are rejected""" | |
| payload = {**base_required_fields, "file_name": " file.txt"} | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 422 | |
| def test_valid_filename_with_parentheses( | |
| self, base_required_fields, mock_dependencies | |
| ): | |
| """Test that filenames with parentheses are accepted""" | |
| payload = {**base_required_fields, "file_name": "file(1).txt"} | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 200 | |
| def test_invalid_session_id_with_special_chars( | |
| self, base_required_fields, mock_dependencies | |
| ): | |
| """Test that session IDs with invalid characters are rejected""" | |
| invalid_ids = ["session@123", "session.123", "session/123", "session 123"] | |
| for invalid_id in invalid_ids: | |
| payload = { | |
| **base_required_fields, | |
| "session_id": invalid_id, | |
| "file_name": "document.txt", | |
| } | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 422 | |
| def test_invalid_age_group(self, base_required_fields, mock_dependencies): | |
| """Test that invalid age groups are rejected""" | |
| payload = { | |
| **base_required_fields, | |
| "age_group": "99-100", # Invalid | |
| "file_name": "document.txt", | |
| } | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 422 | |
| def test_invalid_gender(self, base_required_fields, mock_dependencies): | |
| """Test that invalid gender values are rejected""" | |
| payload = { | |
| **base_required_fields, | |
| "gender": "X", # Invalid - must be M or F | |
| "file_name": "document.txt", | |
| } | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 422 | |
| def test_missing_consent(self, base_required_fields, mock_dependencies): | |
| """Test that missing consent field is rejected""" | |
| payload = {**base_required_fields, "file_name": "document.txt"} | |
| del payload["consent"] | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 422 | |
| def test_invalid_roles_empty_set(self, base_required_fields, mock_dependencies): | |
| """Test that empty roles set is rejected""" | |
| payload = { | |
| **base_required_fields, | |
| "roles": [], # Empty - violates min_length=1 | |
| "file_name": "document.txt", | |
| } | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 422 | |
| def test_invalid_roles_too_many(self, base_required_fields, mock_dependencies): | |
| """Test that more than 5 roles is rejected""" | |
| payload = { | |
| **base_required_fields, | |
| "roles": [ | |
| "patient", | |
| "clinician", | |
| "computer-scientist", | |
| "researcher", | |
| "other", | |
| "extra", | |
| ], | |
| "file_name": "document.txt", | |
| } | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 422 | |
| def test_invalid_role_value(self, base_required_fields, mock_dependencies): | |
| """Test that invalid role values are rejected""" | |
| payload = { | |
| **base_required_fields, | |
| "roles": ["invalid-role"], | |
| "file_name": "document.txt", | |
| } | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 422 | |
| def test_valid_multiple_roles(self, base_required_fields, mock_dependencies): | |
| """Test that multiple valid roles are accepted""" | |
| payload = { | |
| **base_required_fields, | |
| "roles": ["patient", "clinician", "researcher"], | |
| "file_name": "document.txt", | |
| } | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 200 | |
| # ==================== Rate Limiting Tests ==================== | |
| def test_rate_limiting(self, valid_payload, mock_dependencies): | |
| """Test that rate limiting works (20 requests per minute)""" | |
| from fastapi.testclient import TestClient | |
| from main import app | |
| # Create fresh client with rate limiting enabled | |
| rate_limit_client = TestClient(app) | |
| # Make 21 rapid requests | |
| responses = [] | |
| for i in range(21): | |
| response = rate_limit_client.request("DELETE", "/file", json=valid_payload) | |
| responses.append(response) | |
| # First 20 should succeed | |
| # 21st should be rate limited | |
| assert responses[-1].status_code == 429 | |
| # ==================== Integration Tests ==================== | |
| def test_delete_same_file_twice_idempotent(self, valid_payload, mock_dependencies): | |
| """Test that deleting the same file twice is idempotent""" | |
| # First delete | |
| response1 = delete_with_body("/file", valid_payload) | |
| assert response1.status_code == 200 | |
| # Second delete (file already gone) | |
| mock_dependencies["store"].delete_document.return_value = False | |
| response2 = delete_with_body("/file", valid_payload) | |
| assert response2.status_code == 200 | |
| def test_delete_multiple_files_same_session( | |
| self, base_required_fields, mock_dependencies | |
| ): | |
| """Test deleting multiple files from the same session""" | |
| session_id = "test-session" | |
| files = ["file1.txt", "file2.txt", "file3.txt"] | |
| for filename in files: | |
| payload = { | |
| **base_required_fields, | |
| "session_id": session_id, | |
| "file_name": filename, | |
| } | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 200 | |
| def test_delete_files_from_multiple_sessions( | |
| self, base_required_fields, mock_dependencies | |
| ): | |
| """Test deleting files from different sessions""" | |
| sessions_and_files = [ | |
| ("session-1", "file1.txt"), | |
| ("session-2", "file2.txt"), | |
| ("session-3", "file3.txt"), | |
| ] | |
| for session_id, filename in sessions_and_files: | |
| payload = { | |
| **base_required_fields, | |
| "session_id": session_id, | |
| "file_name": filename, | |
| } | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 200 | |
| def test_workflow_order(self, valid_payload, mock_dependencies): | |
| """Test that operations happen in correct order""" | |
| call_order = [] | |
| def track_replace(filename): | |
| call_order.append("replace") | |
| return filename | |
| def track_delete(session_id, filename): | |
| call_order.append("delete") | |
| return True | |
| mock_dependencies["replace_spaces"].side_effect = track_replace | |
| mock_dependencies["store"].delete_document.side_effect = track_delete | |
| delete_with_body("/file", valid_payload) | |
| # replace_spaces should be called before delete_document | |
| assert call_order == ["replace", "delete"] | |
| def test_very_long_filename(self, base_required_fields, mock_dependencies): | |
| """Test handling of very long filenames""" | |
| long_filename = "a" * MAX_FILE_NAME_LENGTH + ".txt" | |
| payload = {**base_required_fields, "file_name": long_filename} | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 422 | |
| def test_very_long_session_id(self, base_required_fields, mock_dependencies): | |
| """Test handling of very long session IDs""" | |
| long_session_id = "s" * 51 | |
| payload = { | |
| **base_required_fields, | |
| "session_id": long_session_id, | |
| "file_name": "document.txt", | |
| } | |
| response = delete_with_body("/file", payload) | |
| assert response.status_code == 422 | |