Spaces:
Paused
Paused
| import pytest | |
| from fastapi.testclient import TestClient | |
| from unittest.mock import patch, Mock | |
| from constants import MAX_COMMENT_LENGTH | |
| from main import app # Adjust import based on your structure | |
| client = TestClient(app) | |
| class TestCommentEndpoint: | |
| """Test the POST /comment 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, "comment": "This is a test comment"} | |
| # ==================== Successful Comment Tests ==================== | |
| def test_comment_success(self, valid_payload): | |
| """Test successful comment submission""" | |
| with patch("main.log_event") as mock_log_event: | |
| response = client.post("/comment", json=valid_payload) | |
| assert response.status_code == 200 | |
| def test_comment_with_long_text(self, base_required_fields): | |
| """Test comment with very long text""" | |
| payload = {**base_required_fields, "comment": "xb" * MAX_COMMENT_LENGTH} | |
| response = client.post("/comment", json=payload) | |
| assert response.status_code == 422 | |
| def test_comment_with_special_characters(self, base_required_fields): | |
| """Test comment with special characters and unicode""" | |
| payload = { | |
| **base_required_fields, | |
| "comment": "Test with special chars: @#$%^&*() 你好 🎉\n\tNew line", | |
| } | |
| response = client.post("/comment", json=payload) | |
| assert response.status_code == 200 | |
| def test_comment_with_multiline_text(self, base_required_fields): | |
| """Test comment with multiple lines""" | |
| payload = {**base_required_fields, "comment": "Line 1\nLine 2\nLine 3"} | |
| response = client.post("/comment", json=payload) | |
| assert response.status_code == 200 | |
| # ==================== Empty Comment Tests ==================== | |
| def test_empty_comment_string(self, base_required_fields): | |
| """Test that empty string comment returns 400""" | |
| payload = {**base_required_fields, "comment": ""} | |
| response = client.post("/comment", json=payload) | |
| assert response.status_code == 422 | |
| def test_whitespace_only_comment(self, base_required_fields): | |
| """Test comment with only whitespace""" | |
| payload = {**base_required_fields, "comment": " "} | |
| # Depends on how backend validates - might be accepted or rejected | |
| response = client.post("/comment", json=payload) | |
| assert response.status_code == 200 | |
| def test_missing_comment_field(self, base_required_fields): | |
| """Test that missing comment field returns validation error""" | |
| payload = {**base_required_fields} | |
| response = client.post("/comment", json=payload) | |
| assert response.status_code == 422 | |
| # ==================== Request Validation Tests ==================== | |
| def test_missing_required_profile_fields(self, base_required_fields): | |
| """Test that missing required fields returns 422""" | |
| payload = {**base_required_fields, "comment": "Test comment"} | |
| del payload["consent"] | |
| response = client.post("/comment", json=payload) | |
| assert response.status_code == 422 | |
| def test_invalid_age_group(self, base_required_fields): | |
| """Test that invalid age group returns 422""" | |
| payload = { | |
| **base_required_fields, | |
| "age_group": "invalid", | |
| "comment": "Test comment", | |
| } | |
| response = client.post("/comment", json=payload) | |
| assert response.status_code == 422 | |
| def test_invalid_gender(self, base_required_fields): | |
| """Test that invalid gender returns 422""" | |
| payload = {**base_required_fields, "gender": "X", "comment": "Test comment"} | |
| response = client.post("/comment", json=payload) | |
| assert response.status_code == 422 | |
| def test_invalid_roles(self, base_required_fields): | |
| """Test that invalid roles return 422""" | |
| payload = { | |
| **base_required_fields, | |
| "roles": ["invalid-role"], | |
| "comment": "Test comment", | |
| } | |
| response = client.post("/comment", json=payload) | |
| assert response.status_code == 422 | |
| def test_empty_roles(self, base_required_fields): | |
| """Test that empty roles set returns 422""" | |
| payload = {**base_required_fields, "roles": [], "comment": "Test comment"} | |
| response = client.post("/comment", json=payload) | |
| assert response.status_code == 422 | |
| # ==================== Background Task Tests ==================== | |
| def test_background_task_receives_correct_data(self, valid_payload): | |
| """Test that background task is called with correct data structure""" | |
| with patch("main.BackgroundTasks.add_task") as mock_add_task: | |
| response = client.post("/comment", json=valid_payload) | |
| assert response.status_code == 200 | |
| # Verify add_task was called | |
| mock_add_task.assert_called_once() | |
| # Check the arguments passed to add_task | |
| call_args = mock_add_task.call_args | |
| # First arg should be log_event function | |
| # Kwargs should contain the data | |
| assert "user_id" in call_args.kwargs | |
| assert "session_id" in call_args.kwargs | |
| assert "data" in call_args.kwargs | |
| data = call_args.kwargs["data"] | |
| assert data["comment"] == "This is a test comment" | |
| assert data["consent"] == True | |
| assert data["age_group"] == "25-34" | |
| def test_different_comment_contents(self, base_required_fields): | |
| """Test various comment contents are accepted""" | |
| comments = [ | |
| "Short", | |
| "A longer comment with multiple words and punctuation!", | |
| "123456789", | |
| "Mixed 123 content with numbers", | |
| ] | |
| for comment_text in comments: | |
| payload = {**base_required_fields, "comment": comment_text} | |
| response = client.post("/comment", json=payload) | |
| assert response.status_code == 200 | |
| # ==================== Rate Limiting Tests ==================== | |
| def test_rate_limiting(self, valid_payload): | |
| """Test that rate limiting works (20 requests per minute)""" | |
| from fastapi.testclient import TestClient | |
| from main import app | |
| rate_limit_client = TestClient(app) | |
| # Make 21 rapid requests | |
| responses = [] | |
| for i in range(21): | |
| response = rate_limit_client.post("/comment", json=valid_payload) | |
| responses.append(response) | |
| # 21st should be rate limited | |
| assert responses[-1].status_code == 429 | |
| # ==================== Integration Tests ==================== | |
| def test_multiple_comments_same_session(self, base_required_fields): | |
| """Test submitting multiple comments from same session""" | |
| comments = ["First comment", "Second comment", "Third comment"] | |
| for comment_text in comments: | |
| payload = {**base_required_fields, "comment": comment_text} | |
| response = client.post("/comment", json=payload) | |
| assert response.status_code == 200 | |
| def test_comments_from_different_sessions(self, base_required_fields): | |
| """Test comments from different sessions""" | |
| sessions = ["session-1", "session-2", "session-3"] | |
| for session_id in sessions: | |
| payload = { | |
| **base_required_fields, | |
| "session_id": session_id, | |
| "comment": f"Comment from {session_id}", | |
| } | |
| response = client.post("/comment", json=payload) | |
| assert response.status_code == 200 | |
| def test_comment_none_value(self, base_required_fields): | |
| """Test that null/None comment is handled""" | |
| payload = {**base_required_fields, "comment": None} | |
| response = client.post("/comment", json=payload) | |
| # Should return 400 or 422 depending on validation | |
| assert response.status_code == 422 | |