Spaces:
Paused
Paused
File size: 8,683 Bytes
8b9e569 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 | 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"""
@pytest.fixture
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"],
}
@pytest.fixture
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 ====================
@pytest.mark.enable_rate_limit
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
|