623 lines
24 KiB
Python
623 lines
24 KiB
Python
# backend/tests/api/v1/test_search.py
|
||
"""
|
||
Тесты для глобального поиска через django-watson.
|
||
"""
|
||
|
||
import itertools
|
||
from datetime import timedelta
|
||
|
||
import pytest
|
||
from django.contrib.auth import get_user_model
|
||
from django.contrib.contenttypes.models import ContentType
|
||
from django.core.management import call_command
|
||
from django.utils import timezone
|
||
from rest_framework import status
|
||
from rest_framework.test import APIClient
|
||
from schedule.models import Calendar, CalendarRelation, Event
|
||
|
||
from api.v1.search.views import _can_view_search_result
|
||
from commerce.models import Product
|
||
from communications.models import Chat, Message
|
||
from documents.models import Document
|
||
from employees.models import Department, Employee, Position
|
||
from feed.models import Post
|
||
from notifications.models import Notification
|
||
from requests_app.models import Request
|
||
from tasks.models import TaskBoard, TaskBoardAccessScope
|
||
|
||
_phone_counter = itertools.count(100_000_000)
|
||
_email_counter = itertools.count(1)
|
||
|
||
|
||
@pytest.fixture
|
||
def make_user(db):
|
||
"""Фабрика пользователей."""
|
||
User = get_user_model()
|
||
|
||
def _make_user(email=None, password="pass12345", **extra):
|
||
if email is None:
|
||
email = f"user{next(_email_counter)}@example.com"
|
||
phone = extra.pop("phone_number", None) or f"+79{next(_phone_counter):09d}"
|
||
user = User.objects.create_user(
|
||
email=email, phone_number=phone, send_activation_email=False, **extra
|
||
)
|
||
if password:
|
||
user.set_password(password)
|
||
user.save()
|
||
return user
|
||
|
||
return _make_user
|
||
|
||
|
||
@pytest.fixture
|
||
def regular_user(make_user):
|
||
"""Обычный пользователь без спецправ."""
|
||
return make_user(
|
||
email="regular@example.com",
|
||
first_name="Регуляр",
|
||
last_name="Пользователь",
|
||
is_active=True,
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def api_client(regular_user):
|
||
"""Авторизованный DRF-клиент."""
|
||
client = APIClient()
|
||
client.force_authenticate(user=regular_user)
|
||
return client
|
||
|
||
|
||
@pytest.fixture
|
||
def sample_data(db):
|
||
"""Создаёт тестовые данные для поиска."""
|
||
# Отделы
|
||
dept_it = Department.objects.create(
|
||
name="IT отдел", description="Информационные технологии"
|
||
)
|
||
dept_hr = Department.objects.create(
|
||
name="HR отдел", description="Управление персоналом"
|
||
)
|
||
|
||
# Должности
|
||
pos_dev = Position.objects.create(
|
||
name="Разработчик", description="Python/Django разработчик"
|
||
)
|
||
pos_manager = Position.objects.create(name="Менеджер", description="HR менеджер")
|
||
|
||
# Сотрудники
|
||
emp1 = Employee.objects.create(
|
||
email="ivanov@test.com",
|
||
first_name="Иван",
|
||
last_name="Иванов",
|
||
patronymic="Иванович",
|
||
position=pos_dev,
|
||
phone_number="+79001234567",
|
||
)
|
||
emp1.departments_links.create(department=dept_it)
|
||
|
||
emp2 = Employee.objects.create(
|
||
email="petrov@test.com",
|
||
first_name="Пётр",
|
||
last_name="Петров",
|
||
patronymic="Петрович",
|
||
position=pos_manager,
|
||
phone_number="+79007654321",
|
||
)
|
||
emp2.departments_links.create(department=dept_hr)
|
||
|
||
# Посты
|
||
post1 = Post.objects.create(
|
||
author=emp1,
|
||
title="Новые технологии в разработке",
|
||
body="Сегодня мы внедрили новый фреймворк Django для нашего проекта",
|
||
)
|
||
|
||
post2 = Post.objects.create(
|
||
author=emp2,
|
||
title="Вакансии в HR отделе",
|
||
body="Требуется менеджер по подбору персонала",
|
||
)
|
||
|
||
# Заявления
|
||
req1 = Request.objects.create(
|
||
employee=emp1,
|
||
title="Заявка на отпуск",
|
||
comment="Прошу предоставить отпуск с 1 по 14 февраля",
|
||
type="vacation",
|
||
status="pending",
|
||
)
|
||
|
||
return {
|
||
"departments": [dept_it, dept_hr],
|
||
"employees": [emp1, emp2],
|
||
"posts": [post1, post2],
|
||
"requests": [req1],
|
||
}
|
||
|
||
|
||
@pytest.mark.django_db
|
||
class TestSearchAPI:
|
||
"""Тесты для API /api/v1/search/"""
|
||
|
||
def test_search_requires_authentication(self, client):
|
||
"""Поиск требует аутентификации."""
|
||
response = client.get("/api/v1/search/?q=test")
|
||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||
|
||
def test_search_empty_query(self, api_client):
|
||
"""Пустой запрос возвращает пустые результаты."""
|
||
response = api_client.get("/api/v1/search/?q=")
|
||
assert response.status_code == status.HTTP_200_OK
|
||
assert response.data["query"] == ""
|
||
assert response.data["total"] == 0
|
||
assert len(response.data["results"]) == 0
|
||
|
||
def test_search_no_results(self, api_client, sample_data):
|
||
"""Поиск без результатов."""
|
||
# Пересоздаём индексы watson для тестовых данных
|
||
from django.core.management import call_command
|
||
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
response = api_client.get("/api/v1/search/?q=несуществующийтекст12345")
|
||
assert response.status_code == status.HTTP_200_OK
|
||
assert response.data["total"] == 0
|
||
|
||
def test_search_finds_employee_by_name(self, api_client, sample_data):
|
||
"""Поиск находит сотрудника по имени."""
|
||
from django.core.management import call_command
|
||
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
response = api_client.get("/api/v1/search/?q=Иванов")
|
||
assert response.status_code == status.HTTP_200_OK
|
||
assert response.data["total"] > 0
|
||
assert "employee" in response.data["counts"]
|
||
|
||
# Проверяем, что в результатах есть сотрудник
|
||
employees = [
|
||
r for r in response.data["results"] if r["model_name"] == "employee"
|
||
]
|
||
assert len(employees) > 0
|
||
assert "Иванов" in employees[0]["title"]
|
||
|
||
def test_search_finds_employee_by_email(self, api_client, sample_data):
|
||
"""Поиск находит сотрудника по email."""
|
||
from django.core.management import call_command
|
||
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
response = api_client.get("/api/v1/search/?q=ivanov@test.com")
|
||
assert response.status_code == status.HTTP_200_OK
|
||
|
||
employees = [
|
||
r for r in response.data["results"] if r["model_name"] == "employee"
|
||
]
|
||
assert len(employees) > 0
|
||
assert "ivanov@test.com" in employees[0]["meta"]["email"].lower()
|
||
|
||
def test_search_omits_employee_hidden_from_ui_without_revealing_count(
|
||
self,
|
||
api_client,
|
||
sample_data,
|
||
):
|
||
employee = sample_data["employees"][0]
|
||
employee.is_hidden_in_ui = True
|
||
employee.save(update_fields=["is_hidden_in_ui"])
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
response = api_client.get("/api/v1/search/?q=ivanov@test.com")
|
||
|
||
assert response.status_code == status.HTTP_200_OK
|
||
assert not [
|
||
result
|
||
for result in response.data["results"]
|
||
if result["model_name"] == "employee"
|
||
]
|
||
assert response.data["counts"].get("employee", 0) == 0
|
||
assert response.data["hidden_counts"].get("employee", 0) == 0
|
||
|
||
@pytest.mark.parametrize("query", ["IT", "it", "It"])
|
||
def test_search_finds_department(self, api_client, sample_data, query):
|
||
"""Поиск находит отдел."""
|
||
from django.core.management import call_command
|
||
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
response = api_client.get("/api/v1/search/", {"q": query})
|
||
assert response.status_code == status.HTTP_200_OK
|
||
|
||
departments = [
|
||
r for r in response.data["results"] if r["model_name"] == "department"
|
||
]
|
||
assert len(departments) > 0
|
||
assert "IT" in departments[0]["title"]
|
||
|
||
def test_stop_word_abbreviation_matches_title_tokens_only(self, api_client):
|
||
from django.db import connection
|
||
|
||
if connection.vendor != "postgresql":
|
||
pytest.skip("The narrow fallback is specific to PostgreSQL stop words")
|
||
match = Department.objects.create(name="Отдел (IT)")
|
||
Department.objects.create(name="Security")
|
||
Department.objects.create(name="Сервис", description="IT support")
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
response = api_client.get("/api/v1/search/", {"q": "it", "type": "department"})
|
||
|
||
assert response.status_code == status.HTTP_200_OK
|
||
assert [result["object_id"] for result in response.data["results"]] == [
|
||
match.pk
|
||
]
|
||
assert response.data["counts"]["department"] == 1
|
||
|
||
def test_abbreviation_fallback_preserves_document_permissions(
|
||
self, api_client, regular_user, make_user
|
||
):
|
||
owner = make_user(email="it-document-owner@example.com")
|
||
document = Document.objects.create(
|
||
title="IT",
|
||
uploaded_by=owner,
|
||
sent_to_all=False,
|
||
)
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
hidden_response = api_client.get("/api/v1/search/", {"q": "it"})
|
||
assert hidden_response.status_code == status.HTTP_200_OK
|
||
hidden_documents = [
|
||
result
|
||
for result in hidden_response.data["results"]
|
||
if result["model_name"] == "document"
|
||
]
|
||
assert not hidden_documents
|
||
assert hidden_response.data["counts"].get("document", 0) == 0
|
||
|
||
document.recipients.add(regular_user)
|
||
visible_response = api_client.get("/api/v1/search/", {"q": "it"})
|
||
assert visible_response.status_code == status.HTTP_200_OK
|
||
visible_documents = [
|
||
result
|
||
for result in visible_response.data["results"]
|
||
if result["model_name"] == "document"
|
||
]
|
||
assert [result["object_id"] for result in visible_documents] == [document.pk]
|
||
assert visible_response.data["counts"]["document"] == 1
|
||
|
||
def test_search_finds_post(self, api_client, sample_data):
|
||
"""Поиск находит пост."""
|
||
from django.core.management import call_command
|
||
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
response = api_client.get("/api/v1/search/?q=Django")
|
||
assert response.status_code == status.HTTP_200_OK
|
||
|
||
posts = [r for r in response.data["results"] if r["model_name"] == "post"]
|
||
assert len(posts) > 0
|
||
assert "Django" in posts[0]["description"] or "Django" in posts[0]["title"]
|
||
|
||
def test_search_finds_request(self, api_client, sample_data):
|
||
"""Поиск находит заявление."""
|
||
from django.core.management import call_command
|
||
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
# Авторизуемся как автор заявления
|
||
emp = sample_data["employees"][0]
|
||
api_client.force_authenticate(user=emp)
|
||
|
||
response = api_client.get("/api/v1/search/?q=отпуск")
|
||
assert response.status_code == status.HTTP_200_OK
|
||
|
||
requests = [r for r in response.data["results"] if r["model_name"] == "request"]
|
||
assert len(requests) > 0
|
||
assert "отпуск" in requests[0]["title"].lower()
|
||
|
||
def test_search_limit_parameter(self, api_client, sample_data):
|
||
"""Параметр limit ограничивает количество результатов на тип."""
|
||
from django.core.management import call_command
|
||
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
# Создадим больше 2 постов
|
||
emp = sample_data["employees"][0]
|
||
for i in range(5):
|
||
Post.objects.create(
|
||
author=emp, title=f"Тестовый пост {i}", body="Django разработка"
|
||
)
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
response = api_client.get("/api/v1/search/?q=Django&limit=2")
|
||
assert response.status_code == status.HTTP_200_OK
|
||
|
||
posts = [r for r in response.data["results"] if r["model_name"] == "post"]
|
||
assert len(posts) <= 2
|
||
|
||
def test_search_supports_category_pagination(self, api_client, sample_data):
|
||
"""Следующая страница возвращает новые записи только выбранного типа."""
|
||
from django.core.management import call_command
|
||
|
||
emp = sample_data["employees"][0]
|
||
for i in range(5):
|
||
Post.objects.create(
|
||
author=emp,
|
||
title=f"Пагинация поиска {i}",
|
||
body="Маркер пагинации поиска",
|
||
)
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
first = api_client.get(
|
||
"/api/v1/search/?q=Маркер пагинации поиска&limit=2&type=post"
|
||
)
|
||
second = api_client.get(
|
||
"/api/v1/search/?q=Маркер пагинации поиска&limit=2&offset=2&type=post"
|
||
)
|
||
|
||
assert first.status_code == status.HTTP_200_OK
|
||
assert second.status_code == status.HTTP_200_OK
|
||
assert len(first.data["results"]) == 2
|
||
assert len(second.data["results"]) == 2
|
||
assert all(result["model_name"] == "post" for result in second.data["results"])
|
||
assert {result["object_id"] for result in first.data["results"]}.isdisjoint(
|
||
{result["object_id"] for result in second.data["results"]}
|
||
)
|
||
assert second.data["counts"]["post"] == 5
|
||
|
||
def test_search_response_structure(self, api_client, sample_data):
|
||
"""Проверка структуры ответа API."""
|
||
from django.core.management import call_command
|
||
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
response = api_client.get("/api/v1/search/?q=Иванов")
|
||
assert response.status_code == status.HTTP_200_OK
|
||
|
||
# Проверяем ключи верхнего уровня
|
||
assert "query" in response.data
|
||
assert "results" in response.data
|
||
assert "counts" in response.data
|
||
assert "hidden_counts" in response.data
|
||
assert "total" in response.data
|
||
assert "hidden_total" in response.data
|
||
|
||
# Проверяем структуру результата
|
||
if len(response.data["results"]) > 0:
|
||
result = response.data["results"][0]
|
||
assert "model_name" in result
|
||
assert "object_id" in result
|
||
assert "title" in result
|
||
assert "url" not in result
|
||
|
||
def test_search_request_permissions(self, api_client, sample_data):
|
||
"""Пользователи видят только свои заявления."""
|
||
from django.core.management import call_command
|
||
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
emp1 = sample_data["employees"][0]
|
||
emp2 = sample_data["employees"][1]
|
||
|
||
# Создаём заявление от emp1
|
||
Request.objects.create(
|
||
employee=emp1,
|
||
title="Личная заявка Иванова",
|
||
comment="Конфиденциально",
|
||
type="vacation",
|
||
status="pending",
|
||
)
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
# Авторизуемся как emp2
|
||
api_client.force_authenticate(user=emp2)
|
||
|
||
response = api_client.get("/api/v1/search/?q=Иванова")
|
||
assert response.status_code == status.HTTP_200_OK
|
||
|
||
# emp2 не должен видеть заявку emp1
|
||
requests = [r for r in response.data["results"] if r["model_name"] == "request"]
|
||
for req in requests:
|
||
# Если есть заявления в результатах, они не должны быть от emp1
|
||
if "Личная заявка" in req["title"]:
|
||
pytest.fail("Пользователь видит чужое заявление")
|
||
|
||
def test_search_does_not_bypass_domain_object_permissions(
|
||
self, api_client, regular_user, make_user
|
||
):
|
||
owner = make_user(email="search-owner@example.com")
|
||
marker = "ClassifiedSearchMarker"
|
||
|
||
chat = Chat.objects.create(
|
||
type="private",
|
||
name=marker,
|
||
created_by=owner,
|
||
)
|
||
chat.participants.add(owner)
|
||
Message.objects.create(chat=chat, author=owner, content=marker)
|
||
|
||
document = Document.objects.create(
|
||
title=marker,
|
||
uploaded_by=owner,
|
||
sent_to_all=False,
|
||
)
|
||
|
||
calendar = Calendar.objects.create(
|
||
name="Private search calendar",
|
||
slug=f"private-search-{owner.id}",
|
||
)
|
||
Event.objects.create(
|
||
calendar=calendar,
|
||
creator=owner,
|
||
title=marker,
|
||
start=timezone.now(),
|
||
end=timezone.now() + timedelta(hours=1),
|
||
)
|
||
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
hidden_response = api_client.get(f"/api/v1/search/?q={marker}")
|
||
hidden_models = {item["model_name"] for item in hidden_response.data["results"]}
|
||
assert not _can_view_search_result(regular_user, document, "document")
|
||
assert hidden_models.isdisjoint({"chat", "message", "schedule_event"})
|
||
assert hidden_response.data["hidden_counts"]["chat"] == 1
|
||
assert hidden_response.data["hidden_counts"]["message"] == 1
|
||
assert hidden_response.data["hidden_counts"]["document"] == 1
|
||
assert hidden_response.data["hidden_counts"]["schedule_event"] == 1
|
||
assert hidden_response.data["hidden_total"] >= 4
|
||
|
||
chat.participants.add(regular_user)
|
||
document.recipients.add(regular_user)
|
||
CalendarRelation.objects.create(
|
||
calendar=calendar,
|
||
content_type=ContentType.objects.get_for_model(get_user_model()),
|
||
object_id=regular_user.id,
|
||
distinction="viewer",
|
||
)
|
||
|
||
visible_response = api_client.get(f"/api/v1/search/?q={marker}")
|
||
visible_models = {
|
||
item["model_name"] for item in visible_response.data["results"]
|
||
}
|
||
assert _can_view_search_result(regular_user, document, "document")
|
||
assert {"chat", "message", "schedule_event"} <= visible_models
|
||
assert visible_response.data["hidden_counts"].get("chat", 0) == 0
|
||
assert visible_response.data["hidden_counts"].get("message", 0) == 0
|
||
assert visible_response.data["hidden_counts"].get("document", 0) == 0
|
||
assert visible_response.data["hidden_counts"].get("schedule_event", 0) == 0
|
||
|
||
def test_search_returns_employee_avatar(self, api_client, sample_data):
|
||
"""Результат сотрудника содержит фото для карточки поиска."""
|
||
employee = sample_data["employees"][0]
|
||
employee.avatar = "users/avatars/search-test.jpg"
|
||
employee.save(update_fields=["avatar"])
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
response = api_client.get("/api/v1/search/?q=Иванов")
|
||
|
||
assert response.status_code == status.HTTP_200_OK
|
||
employees = [
|
||
item
|
||
for item in response.data["results"]
|
||
if item["model_name"] == "employee"
|
||
]
|
||
assert employees
|
||
assert employees[0]["image"].endswith("/users/avatars/search-test.jpg")
|
||
|
||
@pytest.mark.parametrize("model_type", ["", "product"])
|
||
def test_search_hides_deferred_products(self, api_client, model_type):
|
||
"""Отложенный модуль не раскрывается через поиск и его счётчики."""
|
||
Product.objects.create(
|
||
sku="SEARCH-SKU-001",
|
||
name="Уникальный поисковый товар",
|
||
barcode="4600000000001",
|
||
)
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
response = api_client.get("/api/v1/search/", {"q": "товар", "type": model_type})
|
||
|
||
assert response.status_code == status.HTTP_200_OK
|
||
products = [
|
||
item for item in response.data["results"] if item["model_name"] == "product"
|
||
]
|
||
assert products == []
|
||
assert "product" not in response.data["counts"]
|
||
assert "product" not in response.data["hidden_counts"]
|
||
assert response.data["total"] == 0
|
||
|
||
def test_search_counts_inaccessible_task_boards(
|
||
self, api_client, regular_user, make_user
|
||
):
|
||
"""Закрытые доски считаются, но не раскрываются в результатах."""
|
||
owner = make_user(email="search-board-owner@example.com")
|
||
board = TaskBoard.objects.create(
|
||
name="Секретная поисковая доска",
|
||
created_by=owner,
|
||
access_scope=TaskBoardAccessScope.PRIVATE,
|
||
)
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
hidden_response = api_client.get("/api/v1/search/?q=поисковая")
|
||
|
||
assert hidden_response.status_code == status.HTTP_200_OK
|
||
assert hidden_response.data["hidden_counts"]["task_board"] == 1
|
||
assert not any(
|
||
item["model_name"] == "task_board"
|
||
for item in hidden_response.data["results"]
|
||
)
|
||
|
||
board.access_scope = TaskBoardAccessScope.ALL
|
||
board.save(update_fields=["access_scope"])
|
||
visible_response = api_client.get("/api/v1/search/?q=поисковая")
|
||
|
||
assert visible_response.status_code == status.HTTP_200_OK
|
||
assert any(
|
||
item["model_name"] == "task_board" and item["object_id"] == board.id
|
||
for item in visible_response.data["results"]
|
||
)
|
||
|
||
def test_search_cyrillic_query(self, api_client, sample_data):
|
||
"""Поиск по кириллице работает корректно."""
|
||
from django.core.management import call_command
|
||
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
response = api_client.get("/api/v1/search/?q=технологии")
|
||
assert response.status_code == status.HTTP_200_OK
|
||
|
||
# Должны найти пост "Новые технологии в разработке"
|
||
posts = [r for r in response.data["results"] if r["model_name"] == "post"]
|
||
assert len(posts) > 0
|
||
|
||
def test_search_formats_notification_v2_without_500(self, api_client, regular_user):
|
||
"""Поиск по уведомлениям v2 не падает на legacy-ожиданиях формата."""
|
||
from django.core.management import call_command
|
||
|
||
Notification.objects.create(
|
||
recipient=regular_user,
|
||
verb="system_notice",
|
||
description="Это обычный текст уведомления для поиска",
|
||
data={"title": "Важное уведомление"},
|
||
action_url="/documents/5/",
|
||
)
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
response = api_client.get("/api/v1/search/?q=обычный")
|
||
|
||
assert response.status_code == status.HTTP_200_OK
|
||
notifications = [
|
||
r for r in response.data["results"] if r["model_name"] == "notification"
|
||
]
|
||
assert len(notifications) > 0
|
||
assert notifications[0]["title"] == "Важное уведомление"
|
||
assert "обычный текст уведомления" in notifications[0]["description"].lower()
|
||
assert "url" not in notifications[0]
|
||
|
||
|
||
@pytest.mark.django_db
|
||
class TestSearchIndexing:
|
||
"""Тесты индексации watson."""
|
||
|
||
def test_watson_index_auto_updates(self, api_client, sample_data):
|
||
"""Индекс watson обновляется автоматически при изменении данных."""
|
||
from django.core.management import call_command
|
||
|
||
call_command("buildwatson", verbosity=0)
|
||
|
||
# Создаём нового сотрудника
|
||
Employee.objects.create(
|
||
email="sidorov@test.com",
|
||
first_name="Сидор",
|
||
last_name="Сидоров",
|
||
patronymic="Сидорович",
|
||
)
|
||
|
||
# watson должен автоматически добавить в индекс
|
||
response = api_client.get("/api/v1/search/?q=Сидоров")
|
||
assert response.status_code == status.HTTP_200_OK
|
||
|
||
employees = [
|
||
r for r in response.data["results"] if r["model_name"] == "employee"
|
||
]
|
||
assert len(employees) > 0
|
||
assert "Сидоров" in employees[0]["title"]
|