shinalok
148211e236
- app.py: replace print() with logging, basicConfig with LOG_LEVEL env var - api_client.py: shared AsyncClient instance (connection pooling), URL-encode delete_document path parameter, aclose() for cleanup - services/document.py: validate file exists and extension before ingest - tests/: ChatService (4) + DocumentService (6) unit tests via pytest-asyncio - pyproject.toml: asyncio_mode = auto - requirements-dev.txt: pytest, pytest-asyncio Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
from services.document import DocumentService
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_api():
|
|
api = MagicMock()
|
|
api.ingest = AsyncMock(return_value={"chunks": 5})
|
|
api.list_documents = AsyncMock(return_value=["docs/report.pdf"])
|
|
api.delete_document = AsyncMock()
|
|
return api
|
|
|
|
|
|
@pytest.fixture
|
|
def service(mock_api):
|
|
return DocumentService(mock_api)
|
|
|
|
|
|
async def test_ingest_valid_pdf(service, mock_api, tmp_path):
|
|
pdf = tmp_path / "report.pdf"
|
|
pdf.write_bytes(b"%PDF-1.4")
|
|
result = await service.ingest(str(pdf))
|
|
assert result == {"chunks": 5}
|
|
mock_api.ingest.assert_awaited_once_with(str(pdf))
|
|
|
|
|
|
async def test_ingest_valid_txt(service, mock_api, tmp_path):
|
|
txt = tmp_path / "notes.txt"
|
|
txt.write_text("내용")
|
|
result = await service.ingest(str(txt))
|
|
assert result == {"chunks": 5}
|
|
|
|
|
|
async def test_ingest_nonexistent_file_raises(service):
|
|
with pytest.raises(ValueError, match="파일을 찾을 수 없습니다"):
|
|
await service.ingest("/nonexistent/file.pdf")
|
|
|
|
|
|
async def test_ingest_unsupported_extension_raises(service, tmp_path):
|
|
docx = tmp_path / "doc.docx"
|
|
docx.write_bytes(b"data")
|
|
with pytest.raises(ValueError, match="지원하지 않는 파일 형식"):
|
|
await service.ingest(str(docx))
|
|
|
|
|
|
async def test_list_documents_delegates(service, mock_api):
|
|
result = await service.list_documents()
|
|
assert result == ["docs/report.pdf"]
|
|
mock_api.list_documents.assert_awaited_once()
|
|
|
|
|
|
async def test_delete_document_delegates(service, mock_api):
|
|
await service.delete_document("/path/to/doc.pdf")
|
|
mock_api.delete_document.assert_awaited_once_with("/path/to/doc.pdf")
|