Spaces:
Paused
Paused
| import pytest | |
| import sys | |
| from unittest.mock import patch | |
| from classes.session_document_store import SessionDocumentStore | |
| from constants import MAX_FILE_SIZES_PER_SESSION | |
| class TestSessionDocumentStore: | |
| """Unit tests for SessionDocumentStore class""" | |
| def store(self): | |
| """Fresh store instance for each test""" | |
| return SessionDocumentStore() | |
| def sample_session_id(self): | |
| return "test-session-123" | |
| def sample_file_text(self): | |
| return "This is sample file content" | |
| def sample_file_name(self): | |
| return "document.txt" | |
| # ==================== create_document Tests ==================== | |
| def test_create_document_new_session( | |
| self, store, sample_session_id, sample_file_text, sample_file_name | |
| ): | |
| """Test creating a document in a new session""" | |
| result = store.create_document( | |
| sample_session_id, sample_file_text, sample_file_name | |
| ) | |
| assert result is True | |
| assert sample_session_id in store.session_document_map | |
| assert sample_file_name in store.session_document_map[sample_session_id] | |
| stored_text, stored_size = store.session_document_map[sample_session_id][ | |
| sample_file_name | |
| ] | |
| assert stored_text == sample_file_text | |
| assert stored_size == sys.getsizeof(sample_file_text) | |
| def test_create_document_existing_session(self, store, sample_session_id): | |
| """Test adding multiple documents to the same session""" | |
| file1_text = "First document" | |
| file2_text = "Second document" | |
| result1 = store.create_document(sample_session_id, file1_text, "file1.txt") | |
| result2 = store.create_document(sample_session_id, file2_text, "file2.txt") | |
| assert result1 is True | |
| assert result2 is True | |
| assert len(store.session_document_map[sample_session_id]) == 2 | |
| assert "file1.txt" in store.session_document_map[sample_session_id] | |
| assert "file2.txt" in store.session_document_map[sample_session_id] | |
| def test_create_document_duplicate_filename_overwrites( | |
| self, store, sample_session_id | |
| ): | |
| """Test that creating a document with same filename overwrites the previous one""" | |
| first_text = "Original content" | |
| updated_text = "Updated content" | |
| filename = "document.txt" | |
| store.create_document(sample_session_id, first_text, filename) | |
| store.create_document(sample_session_id, updated_text, filename) | |
| stored_text, _ = store.session_document_map[sample_session_id][filename] | |
| assert stored_text == updated_text | |
| assert len(store.session_document_map[sample_session_id]) == 1 | |
| def test_create_document_exceeds_size_limit(self, store, sample_session_id): | |
| """Test that documents exceeding size limit are rejected""" | |
| # Create a large text that exceeds the limit | |
| large_text = "x" * (MAX_FILE_SIZES_PER_SESSION + 1000) | |
| result = store.create_document(sample_session_id, large_text, "large.txt") | |
| assert result is False | |
| assert sample_session_id not in store.session_document_map | |
| def test_create_document_cumulative_size_limit(self, store, sample_session_id): | |
| """Test that cumulative size across multiple files is enforced""" | |
| # Calculate size that would fit individually but not together | |
| half_limit_size = MAX_FILE_SIZES_PER_SESSION // 2 | |
| # Create text slightly over half the limit | |
| text1 = "a" * (half_limit_size - 100) | |
| text2 = "b" * (half_limit_size - 100) | |
| text3 = "c" * 1000 # This should push over the limit | |
| result1 = store.create_document(sample_session_id, text1, "file1.txt") | |
| result2 = store.create_document(sample_session_id, text2, "file2.txt") | |
| result3 = store.create_document(sample_session_id, text3, "file3.txt") | |
| assert result1 is True | |
| assert result2 is True | |
| assert result3 is False # Should be rejected | |
| assert len(store.session_document_map[sample_session_id]) == 2 | |
| def test_create_document_exactly_at_limit(self, store, sample_session_id): | |
| """Test edge case where document is exactly at the size limit""" | |
| # Mock sys.getsizeof to return exact limit | |
| with patch("sys.getsizeof", return_value=MAX_FILE_SIZES_PER_SESSION): | |
| result = store.create_document(sample_session_id, "text", "file.txt") | |
| assert result is True | |
| def test_create_document_one_byte_over_limit(self, store, sample_session_id): | |
| """Test edge case where document is one byte over the limit""" | |
| with patch("sys.getsizeof", return_value=MAX_FILE_SIZES_PER_SESSION + 1): | |
| result = store.create_document(sample_session_id, "text", "file.txt") | |
| assert result is False | |
| def test_create_document_empty_text(self, store, sample_session_id): | |
| """Test creating a document with empty text""" | |
| result = store.create_document(sample_session_id, "", "empty.txt") | |
| assert result is True | |
| assert "" == store.session_document_map[sample_session_id]["empty.txt"][0] | |
| def test_create_document_special_characters_in_filename( | |
| self, store, sample_session_id | |
| ): | |
| """Test filenames with special characters""" | |
| special_filenames = [ | |
| "file with spaces.txt", | |
| "file_with_underscores.txt", | |
| "file-with-dashes.txt", | |
| "file.multiple.dots.txt", | |
| ] | |
| for filename in special_filenames: | |
| result = store.create_document(sample_session_id, "content", filename) | |
| assert result is True | |
| assert filename in store.session_document_map[sample_session_id] | |
| # ==================== get_document_contents Tests ==================== | |
| def test_get_document_contents_single_document(self, store, sample_session_id): | |
| """Test retrieving content from a session with one document""" | |
| text = "Document content" | |
| store.create_document(sample_session_id, text, "file.txt") | |
| contents = store.get_document_contents(sample_session_id) | |
| assert contents is not None | |
| assert len(contents) == 1 | |
| assert contents[0] == text | |
| def test_get_document_contents_multiple_documents(self, store, sample_session_id): | |
| """Test retrieving content from a session with multiple documents""" | |
| texts = ["First doc", "Second doc", "Third doc"] | |
| filenames = ["file1.txt", "file2.txt", "file3.txt"] | |
| for text, filename in zip(texts, filenames): | |
| store.create_document(sample_session_id, text, filename) | |
| contents = store.get_document_contents(sample_session_id) | |
| assert contents is not None | |
| assert len(contents) == 3 | |
| assert set(contents) == set(texts) | |
| def test_get_document_contents_nonexistent_session(self, store): | |
| """Test retrieving content from a session that doesn't exist""" | |
| contents = store.get_document_contents("nonexistent-session") | |
| assert contents is None | |
| def test_get_document_contents_empty_session(self, store, sample_session_id): | |
| """Test retrieving content from a session with no documents""" | |
| # Create session but then delete all documents | |
| store.create_document(sample_session_id, "text", "file.txt") | |
| store.delete_document(sample_session_id, "file.txt") | |
| contents = store.get_document_contents(sample_session_id) | |
| assert contents is None | |
| def test_get_document_contents_returns_list(self, store, sample_session_id): | |
| """Test that get_document_contents returns a list type""" | |
| store.create_document(sample_session_id, "text", "file.txt") | |
| contents = store.get_document_contents(sample_session_id) | |
| assert isinstance(contents, list) | |
| # ==================== delete_document Tests ==================== | |
| def test_delete_document_existing_file(self, store, sample_session_id): | |
| """Test deleting an existing document""" | |
| store.create_document(sample_session_id, "text1", "file1.txt") | |
| store.create_document(sample_session_id, "text2", "file2.txt") | |
| result = store.delete_document(sample_session_id, "file1.txt") | |
| assert result is False # Session still has documents | |
| assert "file1.txt" not in store.session_document_map[sample_session_id] | |
| assert "file2.txt" in store.session_document_map[sample_session_id] | |
| def test_delete_document_last_file_removes_session(self, store, sample_session_id): | |
| """Test that deleting the last document removes the session""" | |
| store.create_document(sample_session_id, "text", "file.txt") | |
| result = store.delete_document(sample_session_id, "file.txt") | |
| assert result is True # Session was deleted | |
| assert sample_session_id not in store.session_document_map | |
| def test_delete_document_nonexistent_session(self, store): | |
| """Test deleting a document from a nonexistent session""" | |
| result = store.delete_document("nonexistent-session", "file.txt") | |
| assert result is False | |
| def test_delete_document_nonexistent_file(self, store, sample_session_id): | |
| """Test deleting a file that doesn't exist in the session""" | |
| store.create_document(sample_session_id, "text", "file1.txt") | |
| result = store.delete_document(sample_session_id, "nonexistent.txt") | |
| assert result is False | |
| assert "file1.txt" in store.session_document_map[sample_session_id] | |
| def test_delete_document_frees_up_space(self, store, sample_session_id): | |
| """Test that deleting a document frees up space for new uploads""" | |
| # Fill up to near limit | |
| large_text = "x" * (MAX_FILE_SIZES_PER_SESSION - 1000) | |
| store.create_document(sample_session_id, large_text, "large.txt") | |
| # Try to add another file - should fail | |
| result1 = store.create_document(sample_session_id, "y" * 2000, "file2.txt") | |
| assert result1 is False | |
| # Delete the large file | |
| store.delete_document(sample_session_id, "large.txt") | |
| # Now should succeed (assuming session wasn't deleted) | |
| if sample_session_id in store.session_document_map: | |
| result2 = store.create_document(sample_session_id, "y" * 2000, "file2.txt") | |
| assert result2 is True | |
| # ==================== delete_session_documents Tests ==================== | |
| def test_delete_session_documents_existing_session(self, store, sample_session_id): | |
| """Test deleting an entire session""" | |
| store.create_document(sample_session_id, "text1", "file1.txt") | |
| store.create_document(sample_session_id, "text2", "file2.txt") | |
| result = store.delete_session_documents(sample_session_id) | |
| assert result is True | |
| assert sample_session_id not in store.session_document_map | |
| def test_delete_session_documents_nonexistent_session(self, store): | |
| """Test deleting a session that doesn't exist""" | |
| result = store.delete_session_documents("nonexistent-session") | |
| assert result is False | |
| def test_delete_session_documents_multiple_sessions(self, store): | |
| """Test that deleting one session doesn't affect others""" | |
| session1 = "session-1" | |
| session2 = "session-2" | |
| store.create_document(session1, "text1", "file1.txt") | |
| store.create_document(session2, "text2", "file2.txt") | |
| store.delete_session_documents(session1) | |
| assert session1 not in store.session_document_map | |
| assert session2 in store.session_document_map | |
| assert "file2.txt" in store.session_document_map[session2] | |
| # ==================== Integration Tests ==================== | |
| def test_full_workflow(self, store): | |
| """Test complete workflow: create, retrieve, delete""" | |
| session_id = "workflow-session" | |
| # Create documents | |
| store.create_document(session_id, "Doc 1", "file1.txt") | |
| store.create_document(session_id, "Doc 2", "file2.txt") | |
| # Retrieve | |
| contents = store.get_document_contents(session_id) | |
| assert len(contents) == 2 | |
| # Delete one | |
| store.delete_document(session_id, "file1.txt") | |
| contents = store.get_document_contents(session_id) | |
| assert len(contents) == 1 | |
| # Delete last one | |
| store.delete_document(session_id, "file2.txt") | |
| contents = store.get_document_contents(session_id) | |
| assert contents is None | |
| def test_multiple_sessions_isolation(self, store): | |
| """Test that sessions are isolated from each other""" | |
| session1 = "session-1" | |
| session2 = "session-2" | |
| store.create_document(session1, "Session 1 content", "file.txt") | |
| store.create_document(session2, "Session 2 content", "file.txt") | |
| contents1 = store.get_document_contents(session1) | |
| contents2 = store.get_document_contents(session2) | |
| assert contents1[0] == "Session 1 content" | |
| assert contents2[0] == "Session 2 content" | |
| def test_size_calculation_accuracy(self, store, sample_session_id): | |
| """Test that size calculations match sys.getsizeof""" | |
| text = "Test content for size validation" | |
| expected_size = sys.getsizeof(text) | |
| store.create_document(sample_session_id, text, "file.txt") | |
| _, actual_size = store.session_document_map[sample_session_id]["file.txt"] | |
| assert actual_size == expected_size | |
| def test_concurrent_operations_same_session(self, store, sample_session_id): | |
| """Test multiple operations on the same session""" | |
| # Add files | |
| store.create_document(sample_session_id, "text1", "file1.txt") | |
| store.create_document(sample_session_id, "text2", "file2.txt") | |
| # Update one | |
| store.create_document(sample_session_id, "updated text1", "file1.txt") | |
| # Delete one | |
| store.delete_document(sample_session_id, "file2.txt") | |
| # Verify final state | |
| contents = store.get_document_contents(sample_session_id) | |
| assert len(contents) == 1 | |
| assert contents[0] == "updated text1" | |
| def test_edge_case_zero_byte_file(self, store, sample_session_id): | |
| """Test handling of zero-byte files""" | |
| with patch("sys.getsizeof", return_value=0): | |
| result = store.create_document(sample_session_id, "", "empty.txt") | |
| assert result is True | |
| # Should be able to add many zero-byte files | |
| for i in range(100): | |
| result = store.create_document(sample_session_id, "", f"empty{i}.txt") | |
| assert result is True | |