350 lines
12 KiB
Python
350 lines
12 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from datetime import datetime
|
||
|
|
from uuid import UUID, uuid4
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from core.http.errors import ApiProblemError
|
||
|
|
from v1.feedback.schemas import FeedbackCreateResponse
|
||
|
|
|
||
|
|
|
||
|
|
class TestFeedbackCreateResponse:
|
||
|
|
def test_valid_response(self):
|
||
|
|
resp = FeedbackCreateResponse(
|
||
|
|
id=str(uuid4()),
|
||
|
|
created_at="2026-04-17T10:30:00Z",
|
||
|
|
)
|
||
|
|
assert isinstance(resp.id, str)
|
||
|
|
assert resp.created_at == "2026-04-17T10:30:00Z"
|
||
|
|
|
||
|
|
def test_extra_fields_forbidden(self):
|
||
|
|
with pytest.raises(Exception):
|
||
|
|
FeedbackCreateResponse(
|
||
|
|
id=str(uuid4()),
|
||
|
|
created_at="2026-04-17T10:30:00Z",
|
||
|
|
unexpected_field="value",
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class _FakeUserFeedback:
|
||
|
|
def __init__(
|
||
|
|
self,
|
||
|
|
*,
|
||
|
|
id: UUID,
|
||
|
|
user_id: UUID | None,
|
||
|
|
feedback_type: str,
|
||
|
|
content: str,
|
||
|
|
images: list[str],
|
||
|
|
device_info: dict,
|
||
|
|
app_version: str,
|
||
|
|
os_version: str,
|
||
|
|
status: str = "pending",
|
||
|
|
) -> None:
|
||
|
|
self.id = id
|
||
|
|
self.user_id = user_id
|
||
|
|
self.feedback_type = feedback_type
|
||
|
|
self.content = content
|
||
|
|
self.images = images
|
||
|
|
self.device_info = device_info
|
||
|
|
self.app_version = app_version
|
||
|
|
self.os_version = os_version
|
||
|
|
self.status = status
|
||
|
|
self.created_at = datetime.now()
|
||
|
|
self.updated_at = datetime.now()
|
||
|
|
|
||
|
|
|
||
|
|
class _FakeFeedbackRepository:
|
||
|
|
def __init__(self) -> None:
|
||
|
|
self._records: list[_FakeUserFeedback] = []
|
||
|
|
self._committed = False
|
||
|
|
|
||
|
|
async def create_feedback(
|
||
|
|
self,
|
||
|
|
*,
|
||
|
|
user_id: UUID | None,
|
||
|
|
feedback_type: str,
|
||
|
|
content: str,
|
||
|
|
images: list[str],
|
||
|
|
device_info: dict,
|
||
|
|
app_version: str,
|
||
|
|
os_version: str,
|
||
|
|
) -> _FakeUserFeedback:
|
||
|
|
record = _FakeUserFeedback(
|
||
|
|
id=uuid4(),
|
||
|
|
user_id=user_id,
|
||
|
|
feedback_type=feedback_type,
|
||
|
|
content=content,
|
||
|
|
images=images,
|
||
|
|
device_info=device_info,
|
||
|
|
app_version=app_version,
|
||
|
|
os_version=os_version,
|
||
|
|
)
|
||
|
|
self._records.append(record)
|
||
|
|
return record
|
||
|
|
|
||
|
|
async def save(self) -> None:
|
||
|
|
self._committed = True
|
||
|
|
|
||
|
|
|
||
|
|
class _FakeUploadFile:
|
||
|
|
def __init__(
|
||
|
|
self,
|
||
|
|
*,
|
||
|
|
filename: str = "test.jpg",
|
||
|
|
content_type: str = "image/jpeg",
|
||
|
|
content: bytes = b"fake-image-data",
|
||
|
|
) -> None:
|
||
|
|
self.filename = filename
|
||
|
|
self.content_type = content_type
|
||
|
|
self._content = content
|
||
|
|
self._read = False
|
||
|
|
|
||
|
|
async def read(self) -> bytes:
|
||
|
|
self._read = True
|
||
|
|
return self._content
|
||
|
|
|
||
|
|
|
||
|
|
class _FakeStorage:
|
||
|
|
def __init__(self) -> None:
|
||
|
|
self.uploaded: list[dict] = []
|
||
|
|
|
||
|
|
async def upload_bytes(
|
||
|
|
self,
|
||
|
|
*,
|
||
|
|
bucket: str,
|
||
|
|
path: str,
|
||
|
|
content: bytes,
|
||
|
|
content_type: str,
|
||
|
|
) -> str:
|
||
|
|
self.uploaded.append(
|
||
|
|
{
|
||
|
|
"bucket": bucket,
|
||
|
|
"path": path,
|
||
|
|
"content_type": content_type,
|
||
|
|
"size": len(content),
|
||
|
|
}
|
||
|
|
)
|
||
|
|
return path
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def fake_repo() -> _FakeFeedbackRepository:
|
||
|
|
return _FakeFeedbackRepository()
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def fake_storage() -> _FakeStorage:
|
||
|
|
return _FakeStorage()
|
||
|
|
|
||
|
|
|
||
|
|
class TestFeedbackServiceValidation:
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_submit_feedback_success_no_images(
|
||
|
|
self, fake_repo: _FakeFeedbackRepository, fake_storage: _FakeStorage
|
||
|
|
):
|
||
|
|
from v1.feedback.service import FeedbackService
|
||
|
|
|
||
|
|
service = FeedbackService(repository=fake_repo, storage=fake_storage)
|
||
|
|
result = await service.submit_feedback(
|
||
|
|
feedback_type="bug",
|
||
|
|
content="App crashes on launch",
|
||
|
|
device_info={"platform": "ios", "model": "iPhone 15"},
|
||
|
|
app_version="1.0.0",
|
||
|
|
os_version="iOS 17.0",
|
||
|
|
images=[],
|
||
|
|
user_id=uuid4(),
|
||
|
|
)
|
||
|
|
assert isinstance(result, FeedbackCreateResponse)
|
||
|
|
assert result.id
|
||
|
|
assert result.created_at
|
||
|
|
assert len(fake_repo._records) == 1
|
||
|
|
assert fake_repo._records[0].feedback_type == "bug"
|
||
|
|
assert fake_repo._records[0].images == []
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_submit_feedback_anonymous(
|
||
|
|
self, fake_repo: _FakeFeedbackRepository, fake_storage: _FakeStorage
|
||
|
|
):
|
||
|
|
from v1.feedback.service import FeedbackService
|
||
|
|
|
||
|
|
service = FeedbackService(repository=fake_repo, storage=fake_storage)
|
||
|
|
await service.submit_feedback(
|
||
|
|
feedback_type="suggestion",
|
||
|
|
content="Add dark mode",
|
||
|
|
device_info={"platform": "android", "model": "Pixel 8"},
|
||
|
|
app_version="1.0.0",
|
||
|
|
os_version="Android 14",
|
||
|
|
images=[],
|
||
|
|
user_id=None,
|
||
|
|
)
|
||
|
|
assert fake_repo._records[0].user_id is None
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_submit_feedback_with_images(
|
||
|
|
self, fake_repo: _FakeFeedbackRepository, fake_storage: _FakeStorage
|
||
|
|
):
|
||
|
|
from v1.feedback.service import FeedbackService
|
||
|
|
|
||
|
|
service = FeedbackService(repository=fake_repo, storage=fake_storage)
|
||
|
|
images = [
|
||
|
|
_FakeUploadFile(filename="screenshot1.jpg", content_type="image/jpeg"),
|
||
|
|
_FakeUploadFile(filename="screenshot2.png", content_type="image/png"),
|
||
|
|
]
|
||
|
|
await service.submit_feedback(
|
||
|
|
feedback_type="bug",
|
||
|
|
content="UI glitch",
|
||
|
|
device_info={"platform": "ios", "model": "iPhone 15"},
|
||
|
|
app_version="1.0.0",
|
||
|
|
os_version="iOS 17.0",
|
||
|
|
images=images, # type: ignore[arg-type]
|
||
|
|
user_id=uuid4(),
|
||
|
|
)
|
||
|
|
assert len(fake_storage.uploaded) == 2
|
||
|
|
assert len(fake_repo._records[0].images) == 2
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_content_empty_raises(
|
||
|
|
self, fake_repo: _FakeFeedbackRepository, fake_storage: _FakeStorage
|
||
|
|
):
|
||
|
|
from v1.feedback.service import FeedbackService
|
||
|
|
|
||
|
|
service = FeedbackService(repository=fake_repo, storage=fake_storage)
|
||
|
|
with pytest.raises(ApiProblemError) as exc_info:
|
||
|
|
await service.submit_feedback(
|
||
|
|
feedback_type="bug",
|
||
|
|
content=" ",
|
||
|
|
device_info={"platform": "ios", "model": "iPhone 15"},
|
||
|
|
app_version="1.0.0",
|
||
|
|
os_version="iOS 17.0",
|
||
|
|
images=[],
|
||
|
|
user_id=None,
|
||
|
|
)
|
||
|
|
assert exc_info.value.code == "FEEDBACK_CONTENT_EMPTY"
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_content_too_long_raises(
|
||
|
|
self, fake_repo: _FakeFeedbackRepository, fake_storage: _FakeStorage
|
||
|
|
):
|
||
|
|
from v1.feedback.service import FeedbackService
|
||
|
|
|
||
|
|
service = FeedbackService(repository=fake_repo, storage=fake_storage)
|
||
|
|
with pytest.raises(ApiProblemError) as exc_info:
|
||
|
|
await service.submit_feedback(
|
||
|
|
feedback_type="bug",
|
||
|
|
content="x" * 501,
|
||
|
|
device_info={"platform": "ios", "model": "iPhone 15"},
|
||
|
|
app_version="1.0.0",
|
||
|
|
os_version="iOS 17.0",
|
||
|
|
images=[],
|
||
|
|
user_id=None,
|
||
|
|
)
|
||
|
|
assert exc_info.value.code == "FEEDBACK_CONTENT_TOO_LONG"
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_too_many_images_raises(
|
||
|
|
self, fake_repo: _FakeFeedbackRepository, fake_storage: _FakeStorage
|
||
|
|
):
|
||
|
|
from v1.feedback.service import FeedbackService
|
||
|
|
|
||
|
|
service = FeedbackService(repository=fake_repo, storage=fake_storage)
|
||
|
|
images = [_FakeUploadFile() for _ in range(4)]
|
||
|
|
with pytest.raises(ApiProblemError) as exc_info:
|
||
|
|
await service.submit_feedback(
|
||
|
|
feedback_type="bug",
|
||
|
|
content="Test",
|
||
|
|
device_info={"platform": "ios", "model": "iPhone 15"},
|
||
|
|
app_version="1.0.0",
|
||
|
|
os_version="iOS 17.0",
|
||
|
|
images=images, # type: ignore[arg-type]
|
||
|
|
user_id=None,
|
||
|
|
)
|
||
|
|
assert exc_info.value.code == "FEEDBACK_TOO_MANY_IMAGES"
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_invalid_image_type_raises(
|
||
|
|
self, fake_repo: _FakeFeedbackRepository, fake_storage: _FakeStorage
|
||
|
|
):
|
||
|
|
from v1.feedback.service import FeedbackService
|
||
|
|
|
||
|
|
service = FeedbackService(repository=fake_repo, storage=fake_storage)
|
||
|
|
images = [_FakeUploadFile(content_type="image/gif")]
|
||
|
|
with pytest.raises(ApiProblemError) as exc_info:
|
||
|
|
await service.submit_feedback(
|
||
|
|
feedback_type="bug",
|
||
|
|
content="Test",
|
||
|
|
device_info={"platform": "ios", "model": "iPhone 15"},
|
||
|
|
app_version="1.0.0",
|
||
|
|
os_version="iOS 17.0",
|
||
|
|
images=images, # type: ignore[arg-type]
|
||
|
|
user_id=None,
|
||
|
|
)
|
||
|
|
assert exc_info.value.code == "FEEDBACK_INVALID_IMAGE_TYPE"
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_image_too_large_raises(
|
||
|
|
self, fake_repo: _FakeFeedbackRepository, fake_storage: _FakeStorage
|
||
|
|
):
|
||
|
|
from v1.feedback.service import FeedbackService
|
||
|
|
|
||
|
|
service = FeedbackService(repository=fake_repo, storage=fake_storage)
|
||
|
|
large_content = b"x" * (6 * 1024 * 1024)
|
||
|
|
images = [_FakeUploadFile(content=large_content, content_type="image/jpeg")]
|
||
|
|
with pytest.raises(ApiProblemError) as exc_info:
|
||
|
|
await service.submit_feedback(
|
||
|
|
feedback_type="bug",
|
||
|
|
content="Test",
|
||
|
|
device_info={"platform": "ios", "model": "iPhone 15"},
|
||
|
|
app_version="1.0.0",
|
||
|
|
os_version="iOS 17.0",
|
||
|
|
images=images, # type: ignore[arg-type]
|
||
|
|
user_id=None,
|
||
|
|
)
|
||
|
|
assert exc_info.value.code == "FEEDBACK_IMAGE_TOO_LARGE"
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_invalid_feedback_type_raises(
|
||
|
|
self, fake_repo: _FakeFeedbackRepository, fake_storage: _FakeStorage
|
||
|
|
):
|
||
|
|
from v1.feedback.service import FeedbackService
|
||
|
|
|
||
|
|
service = FeedbackService(repository=fake_repo, storage=fake_storage)
|
||
|
|
with pytest.raises(ApiProblemError) as exc_info:
|
||
|
|
await service.submit_feedback(
|
||
|
|
feedback_type="invalid_type",
|
||
|
|
content="Test",
|
||
|
|
device_info={"platform": "ios", "model": "iPhone 15"},
|
||
|
|
app_version="1.0.0",
|
||
|
|
os_version="iOS 17.0",
|
||
|
|
images=[],
|
||
|
|
user_id=None,
|
||
|
|
)
|
||
|
|
assert exc_info.value.code == "REQUEST_VALIDATION_ERROR"
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_storage_upload_failure_raises_submit_failed(
|
||
|
|
self, fake_repo: _FakeFeedbackRepository
|
||
|
|
):
|
||
|
|
from v1.feedback.service import FeedbackService
|
||
|
|
|
||
|
|
class _FailingStorage:
|
||
|
|
async def upload_bytes(self, **kwargs: object) -> str:
|
||
|
|
raise RuntimeError("Storage unavailable")
|
||
|
|
|
||
|
|
service = FeedbackService(
|
||
|
|
repository=fake_repo,
|
||
|
|
storage=_FailingStorage(), # type: ignore[arg-type]
|
||
|
|
)
|
||
|
|
images = [_FakeUploadFile()]
|
||
|
|
with pytest.raises(ApiProblemError) as exc_info:
|
||
|
|
await service.submit_feedback(
|
||
|
|
feedback_type="bug",
|
||
|
|
content="Test",
|
||
|
|
device_info={"platform": "ios", "model": "iPhone 15"},
|
||
|
|
app_version="1.0.0",
|
||
|
|
os_version="iOS 17.0",
|
||
|
|
images=images, # type: ignore[arg-type]
|
||
|
|
user_id=None,
|
||
|
|
)
|
||
|
|
assert exc_info.value.code == "FEEDBACK_SUBMIT_FAILED"
|