markov
2bab3cf60a
- Полный набор тестов для всех модулей API - test_auth.py: аутентификация и JWT токены - test_members.py: CRUD участников, агенты, токены - test_projects.py: CRUD проектов, участники проектов - test_tasks.py: CRUD задач, этапы, назначения, зависимости - test_chat.py: сообщения, комментарии, mentions - test_files.py: upload/download файлов проектов - test_labels.py: CRUD лейблов, привязка к задачам - test_websocket.py: WebSocket подключения и события - test_streaming.py: агентный стриминг через WebSocket - conftest.py: фикстуры для подключения к API - requirements.txt: зависимости pytest, httpx, websockets - pytest.ini: настройки asyncio для pytest
404 lines
15 KiB
Python
404 lines
15 KiB
Python
"""
|
||
Тесты работы с файлами - upload, download, файлы проектов
|
||
"""
|
||
import pytest
|
||
import httpx
|
||
import uuid
|
||
import tempfile
|
||
import os
|
||
from conftest import assert_uuid, assert_timestamp
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_upload_file(http_client: httpx.AsyncClient):
|
||
"""Test uploading file via multipart/form-data"""
|
||
# Создаём временный файл
|
||
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
|
||
f.write("This is a test file for upload\nLine 2\nLine 3")
|
||
temp_file_path = f.name
|
||
|
||
try:
|
||
# Загружаем файл
|
||
with open(temp_file_path, 'rb') as f:
|
||
files = {'file': ('test_upload.txt', f, 'text/plain')}
|
||
response = await http_client.post("/upload", files=files)
|
||
|
||
assert response.status_code == 200
|
||
|
||
upload_data = response.json()
|
||
assert "file_id" in upload_data
|
||
assert "filename" in upload_data
|
||
assert "mime_type" in upload_data
|
||
assert "size" in upload_data
|
||
assert "storage_name" in upload_data
|
||
|
||
assert_uuid(upload_data["file_id"])
|
||
assert upload_data["filename"] == "test_upload.txt"
|
||
assert upload_data["mime_type"] == "text/plain"
|
||
assert upload_data["size"] > 0
|
||
assert isinstance(upload_data["storage_name"], str)
|
||
|
||
finally:
|
||
# Удаляем временный файл
|
||
os.unlink(temp_file_path)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_upload_large_file_fails(http_client: httpx.AsyncClient):
|
||
"""Test that uploading too large file returns 413"""
|
||
# Создаём файл размером больше 50MB (лимит из API)
|
||
# Но для теста создадим файл 1MB и проверим что endpoint работает
|
||
large_content = "A" * (1024 * 1024) # 1MB
|
||
|
||
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
|
||
f.write(large_content)
|
||
temp_file_path = f.name
|
||
|
||
try:
|
||
with open(temp_file_path, 'rb') as f:
|
||
files = {'file': ('large_file.txt', f, 'text/plain')}
|
||
response = await http_client.post("/upload", files=files)
|
||
|
||
# В тестовой среде файл 1MB должен проходить успешно
|
||
# В продакшене лимит 50MB вернёт 413
|
||
assert response.status_code in [200, 413]
|
||
|
||
finally:
|
||
os.unlink(temp_file_path)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_upload_without_file_fails(http_client: httpx.AsyncClient):
|
||
"""Test uploading without file field returns 422"""
|
||
response = await http_client.post("/upload", files={})
|
||
assert response.status_code == 422
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_download_attachment_by_id():
|
||
"""Test downloading uploaded file (requires attachment ID)"""
|
||
# Этот тест сложно реализовать без создания сообщения с вложением
|
||
# Оставляем как плейсхолдер для полноценной реализации
|
||
pass
|
||
|
||
|
||
# Тесты файлов проектов
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_project_files_list(http_client: httpx.AsyncClient, test_project: dict):
|
||
"""Test getting list of project files"""
|
||
response = await http_client.get(f"/projects/{test_project['id']}/files")
|
||
assert response.status_code == 200
|
||
|
||
files = response.json()
|
||
assert isinstance(files, list)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_upload_file_to_project(http_client: httpx.AsyncClient, test_project: dict):
|
||
"""Test uploading file directly to project"""
|
||
# Создаём временный файл
|
||
with tempfile.NamedTemporaryFile(mode='w', suffix='.md', delete=False) as f:
|
||
f.write("# Project Documentation\n\nThis is a test document for the project.")
|
||
temp_file_path = f.name
|
||
|
||
try:
|
||
# Загружаем файл в проект
|
||
with open(temp_file_path, 'rb') as f:
|
||
files = {'file': ('README.md', f, 'text/markdown')}
|
||
data = {'description': 'Project documentation file'}
|
||
response = await http_client.post(
|
||
f"/projects/{test_project['id']}/files",
|
||
files=files,
|
||
data=data
|
||
)
|
||
|
||
assert response.status_code == 201
|
||
|
||
project_file = response.json()
|
||
assert project_file["filename"] == "README.md"
|
||
assert project_file["description"] == "Project documentation file"
|
||
assert project_file["mime_type"] == "text/markdown"
|
||
assert project_file["size"] > 0
|
||
assert_uuid(project_file["id"])
|
||
|
||
# Проверяем информацию о загрузившем
|
||
assert "uploaded_by" in project_file
|
||
assert "id" in project_file["uploaded_by"]
|
||
assert "slug" in project_file["uploaded_by"]
|
||
assert "name" in project_file["uploaded_by"]
|
||
|
||
assert_timestamp(project_file["created_at"])
|
||
assert_timestamp(project_file["updated_at"])
|
||
|
||
finally:
|
||
os.unlink(temp_file_path)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_upload_file_to_project_without_description(http_client: httpx.AsyncClient, test_project: dict):
|
||
"""Test uploading file to project without description"""
|
||
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
|
||
f.write("Simple text file without description")
|
||
temp_file_path = f.name
|
||
|
||
try:
|
||
with open(temp_file_path, 'rb') as f:
|
||
files = {'file': ('simple.txt', f, 'text/plain')}
|
||
response = await http_client.post(
|
||
f"/projects/{test_project['id']}/files",
|
||
files=files
|
||
)
|
||
|
||
assert response.status_code == 201
|
||
|
||
project_file = response.json()
|
||
assert project_file["filename"] == "simple.txt"
|
||
assert project_file["description"] is None
|
||
|
||
finally:
|
||
os.unlink(temp_file_path)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_upload_file_to_nonexistent_project(http_client: httpx.AsyncClient):
|
||
"""Test uploading file to non-existent project returns 404"""
|
||
fake_project_id = str(uuid.uuid4())
|
||
|
||
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
|
||
f.write("File for nowhere")
|
||
temp_file_path = f.name
|
||
|
||
try:
|
||
with open(temp_file_path, 'rb') as f:
|
||
files = {'file': ('test.txt', f, 'text/plain')}
|
||
response = await http_client.post(
|
||
f"/projects/{fake_project_id}/files",
|
||
files=files
|
||
)
|
||
|
||
assert response.status_code == 404
|
||
|
||
finally:
|
||
os.unlink(temp_file_path)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_project_file_info(http_client: httpx.AsyncClient, test_project: dict):
|
||
"""Test getting specific project file information"""
|
||
# Сначала загружаем файл
|
||
with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
|
||
f.write('{"test": "data", "project": "file"}')
|
||
temp_file_path = f.name
|
||
|
||
try:
|
||
with open(temp_file_path, 'rb') as f:
|
||
files = {'file': ('data.json', f, 'application/json')}
|
||
data = {'description': 'Test JSON data'}
|
||
upload_response = await http_client.post(
|
||
f"/projects/{test_project['id']}/files",
|
||
files=files,
|
||
data=data
|
||
)
|
||
|
||
assert upload_response.status_code == 201
|
||
project_file = upload_response.json()
|
||
|
||
# Получаем информацию о файле
|
||
response = await http_client.get(f"/projects/{test_project['id']}/files/{project_file['id']}")
|
||
assert response.status_code == 200
|
||
|
||
file_info = response.json()
|
||
assert file_info["id"] == project_file["id"]
|
||
assert file_info["filename"] == "data.json"
|
||
assert file_info["description"] == "Test JSON data"
|
||
assert file_info["mime_type"] == "application/json"
|
||
|
||
finally:
|
||
os.unlink(temp_file_path)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_download_project_file(http_client: httpx.AsyncClient, test_project: dict):
|
||
"""Test downloading project file"""
|
||
# Сначала загружаем файл
|
||
test_content = "Downloadable content\nLine 2"
|
||
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
|
||
f.write(test_content)
|
||
temp_file_path = f.name
|
||
|
||
try:
|
||
with open(temp_file_path, 'rb') as f:
|
||
files = {'file': ('download_test.txt', f, 'text/plain')}
|
||
upload_response = await http_client.post(
|
||
f"/projects/{test_project['id']}/files",
|
||
files=files
|
||
)
|
||
|
||
assert upload_response.status_code == 201
|
||
project_file = upload_response.json()
|
||
|
||
# Скачиваем файл
|
||
response = await http_client.get(
|
||
f"/projects/{test_project['id']}/files/{project_file['id']}/download"
|
||
)
|
||
assert response.status_code == 200
|
||
|
||
# Проверяем заголовки
|
||
assert "content-type" in response.headers
|
||
assert "content-length" in response.headers
|
||
|
||
# Проверяем содержимое
|
||
downloaded_content = response.text
|
||
assert downloaded_content == test_content
|
||
|
||
finally:
|
||
os.unlink(temp_file_path)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_update_project_file_description(http_client: httpx.AsyncClient, test_project: dict):
|
||
"""Test updating project file description"""
|
||
# Загружаем файл
|
||
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
|
||
f.write("File to update description")
|
||
temp_file_path = f.name
|
||
|
||
try:
|
||
with open(temp_file_path, 'rb') as f:
|
||
files = {'file': ('update_desc.txt', f, 'text/plain')}
|
||
data = {'description': 'Original description'}
|
||
upload_response = await http_client.post(
|
||
f"/projects/{test_project['id']}/files",
|
||
files=files,
|
||
data=data
|
||
)
|
||
|
||
assert upload_response.status_code == 201
|
||
project_file = upload_response.json()
|
||
|
||
# Обновляем описание
|
||
update_data = {"description": "Updated file description"}
|
||
response = await http_client.patch(
|
||
f"/projects/{test_project['id']}/files/{project_file['id']}",
|
||
json=update_data
|
||
)
|
||
assert response.status_code == 200
|
||
|
||
updated_file = response.json()
|
||
assert updated_file["description"] == "Updated file description"
|
||
assert updated_file["id"] == project_file["id"]
|
||
assert updated_file["filename"] == project_file["filename"]
|
||
|
||
finally:
|
||
os.unlink(temp_file_path)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_clear_project_file_description(http_client: httpx.AsyncClient, test_project: dict):
|
||
"""Test clearing project file description"""
|
||
# Загружаем файл с описанием
|
||
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
|
||
f.write("File to clear description")
|
||
temp_file_path = f.name
|
||
|
||
try:
|
||
with open(temp_file_path, 'rb') as f:
|
||
files = {'file': ('clear_desc.txt', f, 'text/plain')}
|
||
data = {'description': 'Description to be cleared'}
|
||
upload_response = await http_client.post(
|
||
f"/projects/{test_project['id']}/files",
|
||
files=files,
|
||
data=data
|
||
)
|
||
|
||
assert upload_response.status_code == 201
|
||
project_file = upload_response.json()
|
||
|
||
# Очищаем описание
|
||
update_data = {"description": None}
|
||
response = await http_client.patch(
|
||
f"/projects/{test_project['id']}/files/{project_file['id']}",
|
||
json=update_data
|
||
)
|
||
assert response.status_code == 200
|
||
|
||
updated_file = response.json()
|
||
assert updated_file["description"] is None
|
||
|
||
finally:
|
||
os.unlink(temp_file_path)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_project_file(http_client: httpx.AsyncClient, test_project: dict):
|
||
"""Test deleting project file"""
|
||
# Загружаем файл
|
||
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
|
||
f.write("File to be deleted")
|
||
temp_file_path = f.name
|
||
|
||
try:
|
||
with open(temp_file_path, 'rb') as f:
|
||
files = {'file': ('to_delete.txt', f, 'text/plain')}
|
||
upload_response = await http_client.post(
|
||
f"/projects/{test_project['id']}/files",
|
||
files=files
|
||
)
|
||
|
||
assert upload_response.status_code == 201
|
||
project_file = upload_response.json()
|
||
|
||
# Удаляем файл
|
||
response = await http_client.delete(
|
||
f"/projects/{test_project['id']}/files/{project_file['id']}"
|
||
)
|
||
assert response.status_code == 200
|
||
|
||
data = response.json()
|
||
assert data["ok"] is True
|
||
|
||
# Проверяем что файл действительно удалён
|
||
response = await http_client.get(
|
||
f"/projects/{test_project['id']}/files/{project_file['id']}"
|
||
)
|
||
assert response.status_code == 404
|
||
|
||
finally:
|
||
os.unlink(temp_file_path)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_search_project_files(http_client: httpx.AsyncClient, test_project: dict):
|
||
"""Test searching project files by filename"""
|
||
# Загружаем несколько файлов
|
||
test_files = ["searchable_doc.md", "another_file.txt", "searchable_config.json"]
|
||
uploaded_files = []
|
||
|
||
for filename in test_files:
|
||
with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
|
||
f.write(f"Content of {filename}")
|
||
temp_file_path = f.name
|
||
|
||
try:
|
||
with open(temp_file_path, 'rb') as f:
|
||
files = {'file': (filename, f, 'text/plain')}
|
||
response = await http_client.post(
|
||
f"/projects/{test_project['id']}/files",
|
||
files=files
|
||
)
|
||
assert response.status_code == 201
|
||
uploaded_files.append(response.json())
|
||
|
||
finally:
|
||
os.unlink(temp_file_path)
|
||
|
||
# Ищем файлы со словом "searchable"
|
||
response = await http_client.get(
|
||
f"/projects/{test_project['id']}/files?search=searchable"
|
||
)
|
||
assert response.status_code == 200
|
||
|
||
files = response.json()
|
||
searchable_files = [f for f in files if "searchable" in f["filename"].lower()]
|
||
assert len(searchable_files) >= 2 # Должно найти 2 файла с "searchable" |