Spaces:
Sleeping
Sleeping
| # test_feedback_handler.py | |
| """ | |
| Tests for verification feedback handler. | |
| Tests feedback collection, validation, and storage functionality. | |
| """ | |
| import pytest | |
| from datetime import datetime | |
| from src.core.verification_feedback_handler import ( | |
| VerificationFeedbackHandler, | |
| FeedbackValidationError, | |
| ) | |
| from src.core.verification_models import ( | |
| VerificationRecord, | |
| VerificationSession, | |
| TestMessage, | |
| ) | |
| from src.core.verification_store import JSONVerificationStore | |
| from src.core.message_queue_manager import MessageQueueManager | |
| class TestCorrectFeedbackHandling: | |
| """Tests for handling 'Correct' feedback.""" | |
| def test_handle_correct_feedback_saves_record( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify correct feedback saves verification record.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| messages = [ | |
| TestMessage( | |
| message_id="msg_001", | |
| text="I'm feeling anxious", | |
| pre_classified_label="yellow", | |
| ), | |
| TestMessage( | |
| message_id="msg_002", | |
| text="I'm feeling great", | |
| pre_classified_label="green", | |
| ), | |
| ] | |
| queue_manager.initialize_queue(messages) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| # Handle correct feedback | |
| result = handler.handle_correct_feedback( | |
| message=messages[0], | |
| classifier_decision="yellow", | |
| classifier_confidence=0.85, | |
| classifier_indicators=["anxiety"], | |
| ) | |
| assert result is True | |
| # Verify record was saved | |
| loaded_session = store.load_session(sample_verification_session.session_id) | |
| assert len(loaded_session.verifications) == 1 | |
| assert loaded_session.verifications[0].message_id == "msg_001" | |
| assert loaded_session.verifications[0].is_correct is True | |
| def test_handle_correct_feedback_marks_as_correct( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify correct feedback marks record as correct.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| messages = [ | |
| TestMessage( | |
| message_id="msg_001", | |
| text="I'm feeling anxious", | |
| pre_classified_label="yellow", | |
| ), | |
| ] | |
| queue_manager.initialize_queue(messages) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| handler.handle_correct_feedback( | |
| message=messages[0], | |
| classifier_decision="yellow", | |
| classifier_confidence=0.85, | |
| classifier_indicators=["anxiety"], | |
| ) | |
| loaded_session = store.load_session(sample_verification_session.session_id) | |
| record = loaded_session.verifications[0] | |
| assert record.is_correct is True | |
| assert record.ground_truth_label == "yellow" | |
| assert record.classifier_decision == "yellow" | |
| def test_handle_correct_feedback_advances_queue( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify correct feedback advances to next message.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| messages = [ | |
| TestMessage( | |
| message_id="msg_001", | |
| text="First message", | |
| pre_classified_label="yellow", | |
| ), | |
| TestMessage( | |
| message_id="msg_002", | |
| text="Second message", | |
| pre_classified_label="green", | |
| ), | |
| ] | |
| queue_manager.initialize_queue(messages) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| # Initially at first message | |
| assert queue_manager.get_current_message_id() == "msg_001" | |
| # Handle correct feedback | |
| handler.handle_correct_feedback( | |
| message=messages[0], | |
| classifier_decision="yellow", | |
| classifier_confidence=0.85, | |
| classifier_indicators=["anxiety"], | |
| ) | |
| # Should advance to second message | |
| assert queue_manager.get_current_message_id() == "msg_002" | |
| def test_handle_correct_feedback_stores_all_fields( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify correct feedback stores all required fields.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| messages = [ | |
| TestMessage( | |
| message_id="msg_001", | |
| text="Test message", | |
| pre_classified_label="yellow", | |
| ), | |
| ] | |
| queue_manager.initialize_queue(messages) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| handler.handle_correct_feedback( | |
| message=messages[0], | |
| classifier_decision="yellow", | |
| classifier_confidence=0.92, | |
| classifier_indicators=["anxiety", "stress"], | |
| ) | |
| loaded_session = store.load_session(sample_verification_session.session_id) | |
| record = loaded_session.verifications[0] | |
| assert record.message_id == "msg_001" | |
| assert record.original_message == "Test message" | |
| assert record.classifier_decision == "yellow" | |
| assert record.classifier_confidence == 0.92 | |
| assert record.classifier_indicators == ["anxiety", "stress"] | |
| assert record.ground_truth_label == "yellow" | |
| assert record.verifier_notes == "" | |
| assert record.is_correct is True | |
| assert isinstance(record.timestamp, datetime) | |
| class TestIncorrectFeedbackHandling: | |
| """Tests for handling 'Incorrect' feedback.""" | |
| def test_handle_incorrect_feedback_saves_record( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify incorrect feedback saves verification record.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| messages = [ | |
| TestMessage( | |
| message_id="msg_001", | |
| text="I'm feeling anxious", | |
| pre_classified_label="yellow", | |
| ), | |
| ] | |
| queue_manager.initialize_queue(messages) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| # Handle incorrect feedback | |
| result = handler.handle_incorrect_feedback( | |
| message=messages[0], | |
| classifier_decision="yellow", | |
| classifier_confidence=0.85, | |
| classifier_indicators=["anxiety"], | |
| ground_truth_label="red", | |
| verifier_notes="Missed severe indicators", | |
| ) | |
| assert result is True | |
| # Verify record was saved | |
| loaded_session = store.load_session(sample_verification_session.session_id) | |
| assert len(loaded_session.verifications) == 1 | |
| assert loaded_session.verifications[0].message_id == "msg_001" | |
| assert loaded_session.verifications[0].is_correct is False | |
| def test_handle_incorrect_feedback_marks_as_incorrect( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify incorrect feedback marks record as incorrect.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| messages = [ | |
| TestMessage( | |
| message_id="msg_001", | |
| text="I'm feeling anxious", | |
| pre_classified_label="yellow", | |
| ), | |
| ] | |
| queue_manager.initialize_queue(messages) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| handler.handle_incorrect_feedback( | |
| message=messages[0], | |
| classifier_decision="yellow", | |
| classifier_confidence=0.85, | |
| classifier_indicators=["anxiety"], | |
| ground_truth_label="red", | |
| verifier_notes="", | |
| ) | |
| loaded_session = store.load_session(sample_verification_session.session_id) | |
| record = loaded_session.verifications[0] | |
| assert record.is_correct is False | |
| assert record.ground_truth_label == "red" | |
| assert record.classifier_decision == "yellow" | |
| def test_handle_incorrect_feedback_stores_notes( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify incorrect feedback stores optional notes.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| messages = [ | |
| TestMessage( | |
| message_id="msg_001", | |
| text="Test message", | |
| pre_classified_label="yellow", | |
| ), | |
| ] | |
| queue_manager.initialize_queue(messages) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| notes = "Missed severe distress indicators" | |
| handler.handle_incorrect_feedback( | |
| message=messages[0], | |
| classifier_decision="yellow", | |
| classifier_confidence=0.85, | |
| classifier_indicators=["anxiety"], | |
| ground_truth_label="red", | |
| verifier_notes=notes, | |
| ) | |
| loaded_session = store.load_session(sample_verification_session.session_id) | |
| record = loaded_session.verifications[0] | |
| assert record.verifier_notes == notes | |
| def test_handle_incorrect_feedback_advances_queue( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify incorrect feedback advances to next message.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| messages = [ | |
| TestMessage( | |
| message_id="msg_001", | |
| text="First message", | |
| pre_classified_label="yellow", | |
| ), | |
| TestMessage( | |
| message_id="msg_002", | |
| text="Second message", | |
| pre_classified_label="green", | |
| ), | |
| ] | |
| queue_manager.initialize_queue(messages) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| # Initially at first message | |
| assert queue_manager.get_current_message_id() == "msg_001" | |
| # Handle incorrect feedback | |
| handler.handle_incorrect_feedback( | |
| message=messages[0], | |
| classifier_decision="yellow", | |
| classifier_confidence=0.85, | |
| classifier_indicators=["anxiety"], | |
| ground_truth_label="red", | |
| verifier_notes="", | |
| ) | |
| # Should advance to second message | |
| assert queue_manager.get_current_message_id() == "msg_002" | |
| def test_handle_incorrect_feedback_requires_correction( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify incorrect feedback requires correction selection.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| messages = [ | |
| TestMessage( | |
| message_id="msg_001", | |
| text="Test message", | |
| pre_classified_label="yellow", | |
| ), | |
| ] | |
| queue_manager.initialize_queue(messages) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| # Try to handle incorrect feedback without correction | |
| with pytest.raises(FeedbackValidationError) as exc_info: | |
| handler.handle_incorrect_feedback( | |
| message=messages[0], | |
| classifier_decision="yellow", | |
| classifier_confidence=0.85, | |
| classifier_indicators=["anxiety"], | |
| ground_truth_label="", | |
| verifier_notes="", | |
| ) | |
| assert "Please select a correction" in str(exc_info.value) | |
| def test_handle_incorrect_feedback_validates_correction_option( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify incorrect feedback validates correction is valid option.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| messages = [ | |
| TestMessage( | |
| message_id="msg_001", | |
| text="Test message", | |
| pre_classified_label="yellow", | |
| ), | |
| ] | |
| queue_manager.initialize_queue(messages) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| # Try to handle incorrect feedback with invalid correction | |
| with pytest.raises(FeedbackValidationError) as exc_info: | |
| handler.handle_incorrect_feedback( | |
| message=messages[0], | |
| classifier_decision="yellow", | |
| classifier_confidence=0.85, | |
| classifier_indicators=["anxiety"], | |
| ground_truth_label="invalid", | |
| verifier_notes="", | |
| ) | |
| assert "Invalid correction option" in str(exc_info.value) | |
| def test_handle_incorrect_feedback_accepts_all_valid_corrections( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify incorrect feedback accepts all valid correction options.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| # Test each correction option with a different classifier decision | |
| # to ensure the correction is actually different from the classifier's decision | |
| test_cases = [ | |
| ("green", "yellow"), # classifier says yellow, correction is green | |
| ("yellow", "red"), # classifier says red, correction is yellow | |
| ("red", "green"), # classifier says green, correction is red | |
| ] | |
| for correction, classifier_decision in test_cases: | |
| session = VerificationSession( | |
| session_id=f"session_{correction}", | |
| verifier_name="Test Verifier", | |
| dataset_id="dataset_001", | |
| dataset_name="Test Dataset", | |
| ) | |
| store.save_session(session) | |
| queue_manager = MessageQueueManager(session) | |
| messages = [ | |
| TestMessage( | |
| message_id=f"msg_{correction}", | |
| text="Test message", | |
| pre_classified_label=classifier_decision, | |
| ), | |
| ] | |
| queue_manager.initialize_queue(messages) | |
| handler = VerificationFeedbackHandler(session, store, queue_manager) | |
| # Should not raise exception - correction is different from classifier decision | |
| result = handler.handle_incorrect_feedback( | |
| message=messages[0], | |
| classifier_decision=classifier_decision, | |
| classifier_confidence=0.85, | |
| classifier_indicators=["anxiety"], | |
| ground_truth_label=correction, | |
| verifier_notes="", | |
| ) | |
| assert result is True | |
| class TestFeedbackValidation: | |
| """Tests for feedback validation.""" | |
| def test_validate_feedback_input_correct_is_valid( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify validation passes for correct feedback.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| is_valid, error_msg = handler.validate_feedback_input(is_correct=True) | |
| assert is_valid is True | |
| assert error_msg is None | |
| def test_validate_feedback_input_incorrect_requires_correction( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify validation fails for incorrect without correction.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| is_valid, error_msg = handler.validate_feedback_input( | |
| is_correct=False, ground_truth_label="" | |
| ) | |
| assert is_valid is False | |
| assert "Correction Required" in error_msg or "select" in error_msg.lower() | |
| def test_validate_feedback_input_incorrect_with_valid_correction( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify validation passes for incorrect with valid correction.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| is_valid, error_msg = handler.validate_feedback_input( | |
| is_correct=False, ground_truth_label="red" | |
| ) | |
| assert is_valid is True | |
| assert error_msg is None | |
| def test_validate_notes_field_accepts_empty_notes( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify validation accepts empty notes.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| is_valid, error_msg = handler.validate_notes_field("") | |
| assert is_valid is True | |
| assert error_msg is None | |
| def test_validate_notes_field_accepts_valid_notes( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify validation accepts valid notes.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| notes = "This is a valid note explaining the correction" | |
| is_valid, error_msg = handler.validate_notes_field(notes) | |
| assert is_valid is True | |
| assert error_msg is None | |
| def test_validate_notes_field_rejects_excessive_length( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify validation rejects notes exceeding 500 characters.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| notes = "x" * 501 | |
| is_valid, error_msg = handler.validate_notes_field(notes) | |
| assert is_valid is False | |
| assert "500 characters" in error_msg | |
| class TestSessionStatistics: | |
| """Tests for session statistics retrieval.""" | |
| def test_get_session_statistics_after_feedback( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify session statistics are updated after feedback.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| messages = [ | |
| TestMessage( | |
| message_id="msg_001", | |
| text="Message 1", | |
| pre_classified_label="yellow", | |
| ), | |
| TestMessage( | |
| message_id="msg_002", | |
| text="Message 2", | |
| pre_classified_label="green", | |
| ), | |
| ] | |
| queue_manager.initialize_queue(messages) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| # Add correct feedback | |
| handler.handle_correct_feedback( | |
| message=messages[0], | |
| classifier_decision="yellow", | |
| classifier_confidence=0.85, | |
| classifier_indicators=["anxiety"], | |
| ) | |
| stats = handler.get_session_statistics() | |
| assert stats["verified_count"] == 1 | |
| assert stats["correct_count"] == 1 | |
| assert stats["incorrect_count"] == 0 | |
| def test_is_session_complete_false_when_messages_remain( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify session is not complete when messages remain.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| messages = [ | |
| TestMessage( | |
| message_id="msg_001", | |
| text="Message 1", | |
| pre_classified_label="yellow", | |
| ), | |
| TestMessage( | |
| message_id="msg_002", | |
| text="Message 2", | |
| pre_classified_label="green", | |
| ), | |
| ] | |
| queue_manager.initialize_queue(messages) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| assert handler.is_session_complete() is False | |
| def test_is_session_complete_true_when_all_verified( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify session is complete when all messages verified.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| messages = [ | |
| TestMessage( | |
| message_id="msg_001", | |
| text="Message 1", | |
| pre_classified_label="yellow", | |
| ), | |
| ] | |
| queue_manager.initialize_queue(messages) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| # Verify the only message | |
| handler.handle_correct_feedback( | |
| message=messages[0], | |
| classifier_decision="yellow", | |
| classifier_confidence=0.85, | |
| classifier_indicators=["anxiety"], | |
| ) | |
| assert handler.is_session_complete() is True | |
| def test_get_queue_position( | |
| self, sample_verification_session, temp_storage_dir | |
| ): | |
| """Verify queue position is returned correctly.""" | |
| store = JSONVerificationStore(storage_dir=temp_storage_dir) | |
| store.save_session(sample_verification_session) | |
| queue_manager = MessageQueueManager(sample_verification_session) | |
| messages = [ | |
| TestMessage( | |
| message_id="msg_001", | |
| text="Message 1", | |
| pre_classified_label="yellow", | |
| ), | |
| TestMessage( | |
| message_id="msg_002", | |
| text="Message 2", | |
| pre_classified_label="green", | |
| ), | |
| TestMessage( | |
| message_id="msg_003", | |
| text="Message 3", | |
| pre_classified_label="red", | |
| ), | |
| ] | |
| queue_manager.initialize_queue(messages) | |
| handler = VerificationFeedbackHandler( | |
| sample_verification_session, store, queue_manager | |
| ) | |
| current_pos, total = handler.get_queue_position() | |
| assert current_pos == 1 | |
| assert total == 3 | |