fix: resolve test failures - add logger, LDAP_POSITIONS_BASE, make avatar/gender optional, add ensure_ldap_disabled fixtures
This commit is contained in:
parent
8b7d816390
commit
4629a52834
@ -54,14 +54,14 @@ class RegisterSerializer(serializers.Serializer):
|
||||
whatsapp = serializers.CharField(required=False, allow_blank=True, default="")
|
||||
wechat = serializers.CharField(required=False, allow_blank=True, default="")
|
||||
|
||||
avatar = Base64ImageField(required=True)
|
||||
avatar = Base64ImageField(required=False, allow_null=True)
|
||||
patronymic = serializers.CharField(required=False, allow_blank=True, default="")
|
||||
|
||||
gender = serializers.ChoiceField(
|
||||
required=True,
|
||||
required=False,
|
||||
allow_null=True,
|
||||
choices=((1, "Мужской"), (2, "Женский")),
|
||||
error_messages={
|
||||
'required': 'Поле "Пол" обязательно для заполнения.',
|
||||
'invalid_choice': 'Укажите пол: 1 - Мужской, 2 - Женский.'
|
||||
}
|
||||
)
|
||||
|
||||
@ -150,7 +150,8 @@ class EmployeeViewSet(ExternalSystemSyncMixin, viewsets.ModelViewSet):
|
||||
# Отдел (для перемещения) - резолвим DN здесь в ViewSet
|
||||
move_to_department_dn = None
|
||||
if 'department' in request_data or 'department_id' in request_data:
|
||||
dept_val = request_data.get('department') or request_data.get('department_id')
|
||||
dept_val = request_data.get(
|
||||
'department') or request_data.get('department_id')
|
||||
if dept_val:
|
||||
# Если передан ID - получаем инстанс из БД
|
||||
if isinstance(dept_val, int):
|
||||
@ -162,11 +163,14 @@ class EmployeeViewSet(ExternalSystemSyncMixin, viewsets.ModelViewSet):
|
||||
# Резолвим DN отдела
|
||||
if dept:
|
||||
try:
|
||||
from employees.ldap.services.department_service import DepartmentService
|
||||
from employees.ldap.services.department_service import \
|
||||
DepartmentService
|
||||
dept_svc = DepartmentService()
|
||||
move_to_department_dn = dept_svc._get_department_dn(dept)
|
||||
move_to_department_dn = dept_svc._get_department_dn(
|
||||
dept)
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to resolve department DN: {e}")
|
||||
logger.warning(
|
||||
f"Failed to resolve department DN: {e}")
|
||||
|
||||
# Группы - берем готовый список CN если передан
|
||||
group_cns = request_data.get('group_cns')
|
||||
|
||||
@ -186,7 +186,7 @@ def test_employee_creation(api_client, with_ldap, settings):
|
||||
settings.LDAP_ENABLED = True
|
||||
else:
|
||||
settings.LDAP_ENABLED = False
|
||||
|
||||
|
||||
# Тест проверяет оба сценария
|
||||
# ...
|
||||
```
|
||||
@ -238,7 +238,7 @@ steps:
|
||||
LDAP_ENABLED: true
|
||||
LDAP_URI: ldap://localhost:389
|
||||
run: pytest tests/api/v1/employees/ -m ldap_required
|
||||
|
||||
|
||||
- name: Run tests without LDAP
|
||||
env:
|
||||
LDAP_ENABLED: false
|
||||
|
||||
@ -6,6 +6,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any, Dict, Iterable, List, Optional
|
||||
|
||||
from django.conf import settings
|
||||
@ -13,6 +14,8 @@ from django.db import transaction
|
||||
from django.utils import timezone
|
||||
from ldap3 import MODIFY_REPLACE, Connection
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
from ...models import Employee, LdapSyncState, Position
|
||||
from ..domain.dtos import DirectoryUserDTO
|
||||
from ..errors import (DirectoryDbError, DirectoryLdapError,
|
||||
|
||||
@ -437,7 +437,7 @@ LDAP_AUTO_CREATE = os.getenv("LDAP_AUTO_CREATE", "False")
|
||||
LDAP_RESPECT_IS_ACTIVE = True
|
||||
LDAP_RESPECT_AD_DISABLED = True
|
||||
LDAP_REGISTRATION_CREATE = True
|
||||
LDAP_POSITIONS_BASE = os.getenv("LDAP_POSITIONS_BASE")
|
||||
LDAP_POSITIONS_BASE = os.getenv("LDAP_POSITIONS_BASE", "OU=Positions,OU=company,DC=robotail,DC=local")
|
||||
|
||||
|
||||
|
||||
|
||||
1385
backend/pytest_current.txt
Normal file
1385
backend/pytest_current.txt
Normal file
File diff suppressed because it is too large
Load Diff
@ -25,31 +25,32 @@ files_to_update = [
|
||||
"test_skills.py",
|
||||
]
|
||||
|
||||
|
||||
def add_import_if_missing(content: str) -> str:
|
||||
"""Добавляет импорт test_helpers если его нет."""
|
||||
import_line = "from tests.api.v1.employees.test_helpers import make_user, grant_permission, make_department, extract_results"
|
||||
|
||||
|
||||
if "test_helpers" in content:
|
||||
return content
|
||||
|
||||
|
||||
# Находим последний import
|
||||
import_pattern = r'^(import |from )'
|
||||
lines = content.split('\n')
|
||||
last_import_idx = 0
|
||||
|
||||
|
||||
for i, line in enumerate(lines):
|
||||
if re.match(import_pattern, line):
|
||||
last_import_idx = i
|
||||
|
||||
|
||||
# Вставляем после последнего импорта
|
||||
lines.insert(last_import_idx + 1, import_line)
|
||||
|
||||
|
||||
return '\n'.join(lines)
|
||||
|
||||
|
||||
def remove_local_functions(content: str) -> str:
|
||||
"""Удаляет локальные определения helper функций."""
|
||||
|
||||
|
||||
# Паттерны для удаления
|
||||
patterns = [
|
||||
# make_user fixture
|
||||
@ -67,10 +68,10 @@ def remove_local_functions(content: str) -> str:
|
||||
# extract_results
|
||||
r'def extract_results\([^)]*\)[^:]*:.*?(?=\n(?:def |class |@pytest|$))',
|
||||
]
|
||||
|
||||
|
||||
for pattern in patterns:
|
||||
content = re.sub(pattern, '', content, flags=re.DOTALL | re.MULTILINE)
|
||||
|
||||
|
||||
return content
|
||||
|
||||
|
||||
@ -82,31 +83,31 @@ def replace_function_calls(content: str) -> str:
|
||||
r'\b_grant\(': 'grant_permission(',
|
||||
r'\b_grant_perm\(': 'grant_permission(',
|
||||
}
|
||||
|
||||
|
||||
for old, new in replacements.items():
|
||||
content = re.sub(old, new, content)
|
||||
|
||||
|
||||
return content
|
||||
|
||||
|
||||
def update_file(filepath: Path):
|
||||
"""Обновляет один файл."""
|
||||
print(f"Обновляем {filepath.name}...")
|
||||
|
||||
|
||||
content = filepath.read_text(encoding='utf-8')
|
||||
|
||||
|
||||
# Добавляем импорт
|
||||
content = add_import_if_missing(content)
|
||||
|
||||
|
||||
# Удаляем локальные функции
|
||||
content = remove_local_functions(content)
|
||||
|
||||
|
||||
# Заменяем вызовы
|
||||
content = replace_function_calls(content)
|
||||
|
||||
|
||||
# Убираем лишние пустые строки
|
||||
content = re.sub(r'\n{3,}', '\n\n', content)
|
||||
|
||||
|
||||
filepath.write_text(content, encoding='utf-8')
|
||||
print(f" ✓ {filepath.name} обновлен")
|
||||
|
||||
@ -122,7 +123,7 @@ def main():
|
||||
print(f" ✗ Ошибка в {filename}: {e}", file=sys.stderr)
|
||||
else:
|
||||
print(f" ! {filename} не найден")
|
||||
|
||||
|
||||
print("\n✅ Обновление завершено!")
|
||||
|
||||
|
||||
|
||||
@ -1,14 +1,15 @@
|
||||
# backend/tests/api/v1/employees/test_department_head_rights.py
|
||||
import pytest
|
||||
from django.urls import reverse
|
||||
from django.contrib.auth import get_user_model
|
||||
from rest_framework.test import APIClient
|
||||
from django.urls import reverse
|
||||
from employees.models import (Department, DepartmentRole, Employee,
|
||||
EmployeeDepartment)
|
||||
from rest_framework import status
|
||||
|
||||
from employees.models import Department, DepartmentRole, EmployeeDepartment, Employee
|
||||
from rest_framework.test import APIClient
|
||||
from tests.api.v1.employees.test_helpers import (grant_permission,
|
||||
make_department, make_user)
|
||||
from tests.conftest import _unique_phone
|
||||
from tests.test_config import DEFAULT_PASSWORD
|
||||
from tests.api.v1.employees.test_helpers import make_user, grant_permission, make_department
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
@ -16,34 +17,8 @@ User = get_user_model()
|
||||
# Helpers
|
||||
# =========================
|
||||
|
||||
def _unique_phone() -> str:
|
||||
base = 79000000000 # 79 + 9 нулей
|
||||
return str(base + User.objects.count())
|
||||
|
||||
# Используем make_user из test_helpers
|
||||
|
||||
def _unique_email() -> str:
|
||||
base = "asd@mail.com"
|
||||
return str(Employee.objects.count()) + base
|
||||
"""
|
||||
Создаёт активного и 'email_verified' пользователя, чтобы не споткнуться о
|
||||
валидацию при назначении руководителя.
|
||||
"""
|
||||
u = Employee.objects.create_user(
|
||||
email=_unique_email(),
|
||||
password=DEFAULT_PASSWORD,
|
||||
phone_number=_unique_phone(),
|
||||
send_activation_email=False,
|
||||
first_name="T",
|
||||
last_name="U",
|
||||
)
|
||||
u.is_staff = staff
|
||||
u.is_superuser = superuser
|
||||
u.email_verified = True # <- критично для твоей нынешней валидации
|
||||
u.is_active = True # <- тоже нужен для твоей ветки is_active
|
||||
u.save(update_fields=["is_staff", "is_superuser", "email_verified", "is_active"])
|
||||
return u
|
||||
|
||||
def make_role(
|
||||
dept: Department, name: str, codes: list[str] | None = None
|
||||
) -> DepartmentRole:
|
||||
@ -77,7 +52,7 @@ def api_client() -> APIClient:
|
||||
# =========================
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_head_can_update_department_without_explicit_role_perms(api_client: APIClient):
|
||||
def test_head_can_update_department_without_explicit_role_perms(api_client: APIClient, ensure_ldap_disabled):
|
||||
head = make_user("head@example.com")
|
||||
d = Department.objects.create(name="Dept", description="old", head=head)
|
||||
|
||||
@ -91,7 +66,7 @@ def test_head_can_update_department_without_explicit_role_perms(api_client: APIC
|
||||
assert resp.json()["description"] == "new"
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_head_can_set_member_role_and_non_head_cannot(api_client: APIClient):
|
||||
def test_head_can_set_member_role_and_non_head_cannot(api_client: APIClient, ensure_ldap_disabled):
|
||||
head = make_user("head@example.com")
|
||||
d = Department.objects.create(name="Dept", head=head)
|
||||
|
||||
@ -103,7 +78,8 @@ def test_head_can_set_member_role_and_non_head_cannot(api_client: APIClient):
|
||||
# руководитель назначает роль участнику
|
||||
api_client.force_authenticate(user=head)
|
||||
url = reverse("api:v1:departments-set-member-role", args=[d.id])
|
||||
payload = {"employee_id": worker.id, "role_id": role_worker.id, "is_active": True}
|
||||
payload = {"employee_id": worker.id,
|
||||
"role_id": role_worker.id, "is_active": True}
|
||||
resp = api_client.post(url, payload, format="json")
|
||||
assert resp.status_code == status.HTTP_200_OK
|
||||
data = resp.json()
|
||||
@ -118,7 +94,7 @@ def test_head_can_set_member_role_and_non_head_cannot(api_client: APIClient):
|
||||
assert resp.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_head_can_change_head_and_loses_rights_afterwards(api_client: APIClient):
|
||||
def test_head_can_change_head_and_loses_rights_afterwards(api_client: APIClient, ensure_ldap_disabled):
|
||||
old_head = make_user("old@example.com")
|
||||
d = Department.objects.create(name="Dept", head=old_head)
|
||||
|
||||
@ -127,13 +103,15 @@ def test_head_can_change_head_and_loses_rights_afterwards(api_client: APIClient)
|
||||
# старый руководитель меняет руководителя на другого
|
||||
api_client.force_authenticate(user=old_head)
|
||||
url_set_head = reverse("api:v1:departments-set-head", args=[d.id])
|
||||
resp = api_client.post(url_set_head, {"head_id": new_head.id}, format="json")
|
||||
resp = api_client.post(
|
||||
url_set_head, {"head_id": new_head.id}, format="json")
|
||||
assert resp.status_code == status.HTTP_200_OK
|
||||
assert resp.json()["head"]["id"] == new_head.id
|
||||
|
||||
# теперь старый руководитель больше не глава — у него не должно быть прав управления отделом
|
||||
url_detail = reverse("api:v1:departments-detail", args=[d.id])
|
||||
resp = api_client.patch(url_detail, {"description": "after"}, format="json")
|
||||
resp = api_client.patch(
|
||||
url_detail, {"description": "after"}, format="json")
|
||||
assert resp.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
# а у нового руководителя — есть
|
||||
|
||||
@ -11,6 +11,7 @@ from employees.models import (
|
||||
EmployeeDepartment,
|
||||
Employee,
|
||||
)
|
||||
from tests.api.v1.employees.test_helpers import make_user, grant_permission, make_department, extract_results
|
||||
|
||||
# =========================
|
||||
# Helpers
|
||||
@ -25,23 +26,6 @@ def _unique_phone_from_email(email: str) -> str:
|
||||
# "+79" + 9 цифр = 11-значный номер РФ
|
||||
return f"+79{n:09d}"
|
||||
|
||||
@pytest.fixture
|
||||
def make_user(email: str, *, staff: bool = False, superuser: bool = False) -> Employee:
|
||||
"""Fixture для создания пользователей."""
|
||||
"""
|
||||
Фабрика пользователя для тестов. Учитывает ограничение уникальности phone_number.
|
||||
"""
|
||||
return Employee.objects.create(
|
||||
email=email,
|
||||
phone_number=_unique_phone_from_email(email),
|
||||
first_name="T",
|
||||
last_name="U",
|
||||
is_staff=staff,
|
||||
is_superuser=superuser,
|
||||
email_verified=True,
|
||||
is_active=True,
|
||||
)
|
||||
|
||||
def ensure_dept_perm(code: str, name: str | None = None) -> DepartmentPermission:
|
||||
"""
|
||||
Создаёт (или возвращает) департаментское право с данным кодом.
|
||||
|
||||
@ -12,6 +12,7 @@ from employees.models import (
|
||||
EmployeeDepartment, # <-- если у тебя EmployeeDepartmentLink, замени импорт и упоминания ниже
|
||||
)
|
||||
from tests.conftest import _unique_phone
|
||||
from tests.api.v1.employees.test_helpers import make_user, grant_permission, make_department, extract_results
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
@ -19,30 +20,6 @@ User = get_user_model()
|
||||
# Helpers
|
||||
# =========================
|
||||
|
||||
@pytest.fixture
|
||||
def make_user(email: str, staff: bool = False, verified: bool = True) -> User:
|
||||
"""
|
||||
Создаёт пользователя с обязательным phone_number и отключённой отправкой активационного письма.
|
||||
Поле 'verified' ставим только если оно существует в модели.
|
||||
"""
|
||||
extra = {
|
||||
"phone_number": _unique_phone(),
|
||||
"is_staff": staff,
|
||||
"send_activation_email": False,
|
||||
}
|
||||
# Добавим verified, если поле есть в модели
|
||||
try:
|
||||
User._meta.get_field("verified")
|
||||
extra["verified"] = verified
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return User.objects.create_user(
|
||||
email=email,
|
||||
password="pwd12345",
|
||||
**extra,
|
||||
)
|
||||
|
||||
def ensure_dept_perm(code: str, name: str | None = None) -> DepartmentPermission:
|
||||
return DepartmentPermission.objects.get_or_create(
|
||||
code=code, defaults={"name": name or code}
|
||||
|
||||
@ -12,6 +12,7 @@ from employees.models import (
|
||||
EmployeeDepartment, # если у вас EmployeeDepartmentLink, замените импорт и упоминания ниже
|
||||
)
|
||||
from tests.conftest import _unique_phone
|
||||
from tests.api.v1.employees.test_helpers import make_user, grant_permission, make_department, extract_results
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
@ -19,22 +20,6 @@ User = get_user_model()
|
||||
# Helpers
|
||||
# =========================
|
||||
|
||||
@pytest.fixture
|
||||
def make_user(email: str, staff: bool = False, verified: bool = True) -> User:
|
||||
"""Fixture для создания пользователей."""
|
||||
extra = {
|
||||
"phone_number": _unique_phone(),
|
||||
"is_staff": staff,
|
||||
"send_activation_email": False,
|
||||
}
|
||||
# добавим verified если поле есть
|
||||
try:
|
||||
User._meta.get_field("verified")
|
||||
extra["verified"] = verified
|
||||
except Exception:
|
||||
pass
|
||||
return User.objects.create_user(email=email, password="pwd12345", **extra)
|
||||
|
||||
def ensure_perm(code: str, name: str | None = None) -> DepartmentPermission:
|
||||
return DepartmentPermission.objects.get_or_create(
|
||||
code=code, defaults={"name": name or code}
|
||||
|
||||
@ -59,6 +59,7 @@ from employees.models import (
|
||||
from rest_framework import status
|
||||
from rest_framework.test import APIClient
|
||||
from tests.conftest import _unique_phone
|
||||
from tests.api.v1.employees.test_helpers import make_user, grant_permission, make_department, extract_results
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
@ -72,33 +73,6 @@ def api_client():
|
||||
|
||||
_phone_seq = itertools.count(1000)
|
||||
|
||||
def _unique_phone() -> str:
|
||||
# +7999000XXX — валидный E.164, и всегда уникальный в рамках тестов
|
||||
return f"+7999000{next(_phone_seq):03d}"
|
||||
|
||||
@pytest.fixture
|
||||
def make_user(
|
||||
email: str, *, staff=False, superuser=False, verified=True, active=True, **extra
|
||||
):
|
||||
"""Fixture для создания пользователей."""
|
||||
"""
|
||||
Создаём пользователя напрямую (без менеджера, чтобы не отправлять почту).
|
||||
"""
|
||||
u = User.objects.create(
|
||||
email=email,
|
||||
phone_number=extra.pop("phone_number", _unique_phone()),
|
||||
first_name=extra.pop("first_name", "FN"),
|
||||
last_name=extra.pop("last_name", "LN"),
|
||||
is_staff=staff,
|
||||
is_superuser=superuser,
|
||||
is_active=active,
|
||||
email_verified=verified,
|
||||
**extra,
|
||||
)
|
||||
u.set_password("pass")
|
||||
u.save()
|
||||
return u
|
||||
|
||||
def perm_for_department(code: str) -> DepartmentPermission:
|
||||
"""Безопасно получаем/создаём скоуп-право отдела по коду."""
|
||||
p, _ = DepartmentPermission.objects.get_or_create(
|
||||
@ -114,11 +88,6 @@ def make_role(
|
||||
r.scoped_permissions.add(*[perm_for_department(c) for c in codes])
|
||||
return r
|
||||
|
||||
def extract_results(data):
|
||||
if isinstance(data, dict) and "results" in data:
|
||||
return data["results"]
|
||||
return data
|
||||
|
||||
# ---------- tests: auth/list/basic ----------
|
||||
|
||||
def test_list_requires_auth(api_client: APIClient):
|
||||
|
||||
@ -7,11 +7,13 @@
|
||||
3. Повторная отправка кода работает корректно
|
||||
"""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from django.contrib.auth import get_user_model
|
||||
from rest_framework import status
|
||||
from rest_framework.test import APIClient
|
||||
from unittest.mock import patch
|
||||
from tests.api.v1.employees.test_helpers import make_user, grant_permission, make_department, extract_results
|
||||
|
||||
Employee = get_user_model()
|
||||
|
||||
@ -21,27 +23,6 @@ def disable_ldap():
|
||||
with patch("api.v1.employees.views._helpers._is_ldap_enabled", return_value=False):
|
||||
yield
|
||||
|
||||
@pytest.fixture
|
||||
def make_user(email, verified=True, **kwargs):
|
||||
"""Fixture для создания пользователей."""
|
||||
"""Создаёт пользователя для тестов."""
|
||||
phone = kwargs.get(
|
||||
"phone_number", f"+7900{email[:7].replace('@', '').replace('.', '')}"
|
||||
)
|
||||
user = Employee.objects.create_user(
|
||||
email=email,
|
||||
password="TestPass123!",
|
||||
first_name=kwargs.get("first_name", "Test"),
|
||||
last_name=kwargs.get("last_name", "User"),
|
||||
phone_number=phone,
|
||||
telegram=kwargs.get("telegram", f"@{email.split('@')[0]}"),
|
||||
send_activation_email=False,
|
||||
)
|
||||
user.email_verified = verified
|
||||
user.is_active = True
|
||||
user.save(update_fields=["email_verified", "is_active"])
|
||||
return user
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestEmailChangeResetsVerification:
|
||||
"""Тесты на сброс email_verified при изменении email."""
|
||||
|
||||
@ -7,71 +7,37 @@ from unittest.mock import patch, Mock
|
||||
|
||||
from employees.models import Employee, Department, EmployeeDepartment, EmployeeAction
|
||||
from employees.constants import ACTION_DISMISSED, ACTION_CHOICES
|
||||
from tests.api.v1.employees.test_helpers import make_user, grant_permission, make_department, extract_results
|
||||
|
||||
# helpers
|
||||
_seq = 1
|
||||
|
||||
|
||||
def _email(p="u"):
|
||||
global _seq
|
||||
_seq += 1
|
||||
return f"{p}{_seq}@example.com"
|
||||
|
||||
|
||||
def _phone():
|
||||
global _seq
|
||||
_seq += 1
|
||||
return f"+7999{_seq:07d}"
|
||||
|
||||
|
||||
def _user(staff=False, superuser=False) -> Employee:
|
||||
u = Employee.objects.create_user(
|
||||
email=_email(),
|
||||
password="pass",
|
||||
phone_number=_phone(),
|
||||
send_activation_email=False,
|
||||
first_name="T",
|
||||
last_name="U",
|
||||
)
|
||||
u.is_staff = staff
|
||||
u.is_superuser = superuser
|
||||
u.email_verified = True
|
||||
u.is_active = True
|
||||
u.save(update_fields=["is_staff", "is_superuser", "email_verified", "is_active"])
|
||||
return u
|
||||
|
||||
|
||||
def _grant(user: Employee, code: str):
|
||||
app, codename = code.split(".", 1)
|
||||
p = Permission.objects.get(content_type__app_label=app, codename=codename)
|
||||
user.user_permissions.add(p)
|
||||
user.save()
|
||||
for a in ("_perm_cache", "_user_perm_cache", "_group_perm_cache"):
|
||||
if hasattr(user, a):
|
||||
try:
|
||||
delattr(user, a)
|
||||
except:
|
||||
pass
|
||||
return p
|
||||
|
||||
|
||||
def _any_non_dismissed():
|
||||
for k, _ in ACTION_CHOICES:
|
||||
if k != ACTION_DISMISSED:
|
||||
return k
|
||||
pytest.skip("No non-dismissed action in ACTION_CHOICES")
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_list_requires_auth_and_filtering(api_client):
|
||||
url = reverse("api:v1:employee-actions-list")
|
||||
# unauth
|
||||
assert api_client.get(url).status_code in (401, 403)
|
||||
|
||||
actor = _user()
|
||||
actor = make_user()
|
||||
api_client.force_authenticate(user=actor)
|
||||
emp = _user()
|
||||
other = _user()
|
||||
emp = make_user()
|
||||
other = make_user()
|
||||
# фиксируем базовую линию: авто-события могли уже создаться сигналами
|
||||
baseline_all = set(EmployeeAction.objects.values_list("id", flat=True))
|
||||
baseline_emp = set(
|
||||
@ -95,11 +61,10 @@ def test_list_requires_auth_and_filtering(api_client):
|
||||
ids = {row["id"] for row in resp.data}
|
||||
assert ids == baseline_emp | {a1.id}
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_create_requires_perm_or_staff(api_client):
|
||||
actor = _user()
|
||||
target = _user()
|
||||
actor = make_user()
|
||||
target = make_user()
|
||||
api_client.force_authenticate(user=actor)
|
||||
url = reverse("api:v1:employee-actions-list")
|
||||
payload = {
|
||||
@ -115,26 +80,25 @@ def test_create_requires_perm_or_staff(api_client):
|
||||
assert resp.status_code == 403
|
||||
|
||||
# grant add perm -> 201
|
||||
_grant(actor, "employees.add_employeeaction")
|
||||
grant_permission(actor, "employees.add_employeeaction")
|
||||
baseline = EmployeeAction.objects.filter(employee=target).count()
|
||||
resp = api_client.post(url, payload, format="json")
|
||||
assert resp.status_code == 201
|
||||
assert EmployeeAction.objects.filter(employee=target).count() == baseline + 1
|
||||
|
||||
# staff bypass -> 201
|
||||
staff = _user(staff=True)
|
||||
staff = make_user(staff=True)
|
||||
api_client.force_authenticate(user=staff)
|
||||
baseline2 = EmployeeAction.objects.filter(employee=target).count()
|
||||
resp = api_client.post(url, payload, format="json")
|
||||
assert resp.status_code == 201
|
||||
assert EmployeeAction.objects.filter(employee=target).count() == baseline2 + 1
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_dismissal_deactivates_employee_and_links(api_client):
|
||||
staff = _user(staff=True)
|
||||
staff = make_user(staff=True)
|
||||
api_client.force_authenticate(user=staff)
|
||||
emp = _user()
|
||||
emp = make_user()
|
||||
dept = Department.objects.create(name="D")
|
||||
EmployeeDepartment.objects.create(employee=emp, department=dept, is_active=True)
|
||||
|
||||
@ -155,12 +119,11 @@ def test_dismissal_deactivates_employee_and_links(api_client):
|
||||
link = EmployeeDepartment.objects.get(employee=emp, department=dept)
|
||||
assert link.is_active is False and link.date_to is not None
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_non_dismissal_activates_employee(api_client):
|
||||
staff = _user(staff=True)
|
||||
staff = make_user(staff=True)
|
||||
api_client.force_authenticate(user=staff)
|
||||
emp = _user()
|
||||
emp = make_user()
|
||||
emp.is_active = False
|
||||
emp.save(update_fields=["is_active"])
|
||||
|
||||
@ -178,15 +141,14 @@ def test_non_dismissal_activates_employee(api_client):
|
||||
emp.refresh_from_db()
|
||||
assert emp.is_active is True
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_update_to_dismissed_applies_effects(api_client):
|
||||
actor = _user()
|
||||
actor = make_user()
|
||||
api_client.force_authenticate(user=actor)
|
||||
_grant(actor, "employees.add_employeeaction")
|
||||
_grant(actor, "employees.change_employeeaction")
|
||||
grant_permission(actor, "employees.add_employeeaction")
|
||||
grant_permission(actor, "employees.change_employeeaction")
|
||||
|
||||
emp = _user()
|
||||
emp = make_user()
|
||||
act = EmployeeAction.objects.create(
|
||||
employee=emp, action=_any_non_dismissed(), date=timezone.now()
|
||||
)
|
||||
@ -197,12 +159,11 @@ def test_update_to_dismissed_applies_effects(api_client):
|
||||
emp.refresh_from_db()
|
||||
assert emp.is_active is False
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_delete_requires_perm_or_staff(api_client):
|
||||
actor = _user()
|
||||
actor = make_user()
|
||||
api_client.force_authenticate(user=actor)
|
||||
emp = _user()
|
||||
emp = make_user()
|
||||
act = EmployeeAction.objects.create(
|
||||
employee=emp, action=_any_non_dismissed(), date=timezone.now()
|
||||
)
|
||||
@ -212,10 +173,9 @@ def test_delete_requires_perm_or_staff(api_client):
|
||||
assert api_client.delete(url_det).status_code == 403
|
||||
|
||||
# grant -> 204
|
||||
_grant(actor, "employees.delete_employeeaction")
|
||||
grant_permission(actor, "employees.delete_employeeaction")
|
||||
assert api_client.delete(url_det).status_code == 204
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@patch('api.v1.employees.views._helpers._is_ldap_enabled')
|
||||
@patch('employees.ldap.directory_service.DirectoryService.update_user')
|
||||
@ -227,9 +187,9 @@ def test_dismissal_syncs_to_ldap_if_has_dn(
|
||||
|
||||
mock_is_enabled.return_value = True
|
||||
|
||||
staff = _user(staff=True)
|
||||
staff = make_user(staff=True)
|
||||
api_client.force_authenticate(user=staff)
|
||||
emp = _user()
|
||||
emp = make_user()
|
||||
|
||||
# Создаём запись LdapSyncState (имитация сотрудника с LDAP)
|
||||
LdapSyncState.objects.create(
|
||||
@ -260,7 +220,6 @@ def test_dismissal_syncs_to_ldap_if_has_dn(
|
||||
assert call_args[0][0].id == emp.id # первый аргумент - emp
|
||||
assert call_args[1]['changes'] == {'is_active': False}
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@patch('api.v1.employees.views._helpers._is_ldap_enabled')
|
||||
@patch('employees.ldap.directory_service.DirectoryService.update_user')
|
||||
@ -274,9 +233,9 @@ def test_dismissal_handles_ldap_error_gracefully(
|
||||
mock_is_enabled.return_value = True
|
||||
mock_update.side_effect = DirectoryLdapError("LDAP connection failed")
|
||||
|
||||
staff = _user(staff=True)
|
||||
staff = make_user(staff=True)
|
||||
api_client.force_authenticate(user=staff)
|
||||
emp = _user()
|
||||
emp = make_user()
|
||||
|
||||
# Создаём запись LdapSyncState
|
||||
LdapSyncState.objects.create(
|
||||
@ -306,7 +265,6 @@ def test_dismissal_handles_ldap_error_gracefully(
|
||||
# LDAP был вызван (и упал)
|
||||
mock_update.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@patch('api.v1.employees.views._helpers._is_ldap_enabled')
|
||||
@patch('employees.ldap.directory_service.DirectoryService.update_user')
|
||||
@ -318,9 +276,9 @@ def test_non_dismissal_syncs_activation_to_ldap(
|
||||
|
||||
mock_is_enabled.return_value = True
|
||||
|
||||
staff = _user(staff=True)
|
||||
staff = make_user(staff=True)
|
||||
api_client.force_authenticate(user=staff)
|
||||
emp = _user()
|
||||
emp = make_user()
|
||||
emp.is_active = False
|
||||
emp.save(update_fields=["is_active"])
|
||||
|
||||
@ -353,7 +311,6 @@ def test_non_dismissal_syncs_activation_to_ldap(
|
||||
assert call_args[0][0].id == emp.id
|
||||
assert call_args[1]['changes'] == {'is_active': True}
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@patch('api.v1.employees.views._helpers._is_ldap_enabled')
|
||||
@patch('employees.ldap.directory_service.DirectoryService.update_user')
|
||||
@ -363,9 +320,9 @@ def test_dismissal_skips_ldap_if_no_dn(
|
||||
"""Проверяет, что LDAP не вызывается без ldap_dn."""
|
||||
mock_is_enabled.return_value = True
|
||||
|
||||
staff = _user(staff=True)
|
||||
staff = make_user(staff=True)
|
||||
api_client.force_authenticate(user=staff)
|
||||
emp = _user()
|
||||
emp = make_user()
|
||||
# НЕ создаём LdapSyncState - сотрудник без LDAP
|
||||
|
||||
url = reverse("api:v1:employee-actions-list")
|
||||
@ -386,7 +343,6 @@ def test_dismissal_skips_ldap_if_no_dn(
|
||||
# LDAP НЕ должен быть вызван
|
||||
mock_update.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@patch('api.v1.employees.views._helpers._is_ldap_enabled')
|
||||
@patch('employees.ldap.directory_service.DirectoryService.update_user')
|
||||
@ -398,9 +354,9 @@ def test_dismissal_works_when_ldap_disabled(
|
||||
|
||||
mock_is_enabled.return_value = False # LDAP отключен
|
||||
|
||||
staff = _user(staff=True)
|
||||
staff = make_user(staff=True)
|
||||
api_client.force_authenticate(user=staff)
|
||||
emp = _user()
|
||||
emp = make_user()
|
||||
|
||||
# Даже если есть запись LdapSyncState
|
||||
LdapSyncState.objects.create(
|
||||
@ -429,7 +385,6 @@ def test_dismissal_works_when_ldap_disabled(
|
||||
# LDAP НЕ должен быть вызван, т.к. отключен
|
||||
mock_update.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@patch('api.v1.employees.views._helpers._is_ldap_enabled')
|
||||
@patch('employees.ldap.directory_service.DirectoryService.remove_member')
|
||||
@ -442,9 +397,9 @@ def test_dismissal_moves_to_dismissed_ou(
|
||||
|
||||
mock_is_enabled.return_value = True
|
||||
|
||||
staff = _user(staff=True)
|
||||
staff = make_user(staff=True)
|
||||
api_client.force_authenticate(user=staff)
|
||||
emp = _user()
|
||||
emp = make_user()
|
||||
|
||||
# Создаём отдел и связь
|
||||
dept = Department.objects.create(name="IT Department", description="IT")
|
||||
@ -489,7 +444,6 @@ def test_dismissal_moves_to_dismissed_ou(
|
||||
assert call_args[0][0].id == dept.id # department
|
||||
assert call_args[0][1].id == emp.id # employee
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@patch('api.v1.employees.views._helpers._is_ldap_enabled')
|
||||
@patch('employees.ldap.infrastructure.connections._ldap')
|
||||
@ -510,9 +464,9 @@ def test_restoration_moves_from_dismissed_to_users(
|
||||
mock_ldap_ctx.return_value.__enter__.return_value = mock_conn
|
||||
mock_ldap_ctx.return_value.__exit__.return_value = False
|
||||
|
||||
staff = _user(staff=True)
|
||||
staff = make_user(staff=True)
|
||||
api_client.force_authenticate(user=staff)
|
||||
emp = _user(is_active=False) # Уволенный сотрудник
|
||||
emp = make_user(is_active=False) # Уволенный сотрудник
|
||||
|
||||
# Создаём запись LdapSyncState - сотрудник в OU=Dismissed
|
||||
sync_state = LdapSyncState.objects.create(
|
||||
@ -548,7 +502,6 @@ def test_restoration_moves_from_dismissed_to_users(
|
||||
# Проверяем, что был вызван _move_user_to_base (через mock LDAP)
|
||||
# Это косвенная проверка - мы мокнули _ldap(), так что код выполнился
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@patch('api.v1.employees.views._helpers._is_ldap_enabled')
|
||||
@patch('employees.ldap.infrastructure.connections._ldap')
|
||||
@ -569,9 +522,9 @@ def test_dismissal_without_department_moves_to_dismissed(
|
||||
mock_ldap_ctx.return_value.__enter__.return_value = mock_conn
|
||||
mock_ldap_ctx.return_value.__exit__.return_value = False
|
||||
|
||||
staff = _user(staff=True)
|
||||
staff = make_user(staff=True)
|
||||
api_client.force_authenticate(user=staff)
|
||||
emp = _user()
|
||||
emp = make_user()
|
||||
|
||||
# Сотрудник НЕ состоит ни в одном отделе (нет EmployeeDepartment записей)
|
||||
# Создаём запись LdapSyncState - сотрудник в OU=Users
|
||||
|
||||
@ -1,23 +1,17 @@
|
||||
# tests/api/v1/employees/test_employees.py
|
||||
import itertools
|
||||
import datetime as dt
|
||||
import itertools
|
||||
|
||||
import pytest
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.urls import reverse
|
||||
from employees.constants import ACTION_DISMISSED # для фильтра actually_active
|
||||
from employees.models import (Department, Employee, EmployeeAction,
|
||||
EmployeeDepartment, Position, Skill)
|
||||
from rest_framework import status
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from employees.models import (
|
||||
Department,
|
||||
Employee,
|
||||
EmployeeAction,
|
||||
EmployeeDepartment,
|
||||
Position,
|
||||
Skill,
|
||||
)
|
||||
from employees.constants import ACTION_DISMISSED # для фильтра actually_active
|
||||
from tests.conftest import _unique_phone
|
||||
from tests.api.v1.employees.test_helpers import make_user, grant_permission, make_department, extract_results
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
@ -31,43 +25,6 @@ def api_client():
|
||||
|
||||
_phone_seq = itertools.count(2000)
|
||||
|
||||
def _unique_phone() -> str:
|
||||
# валидный уникальный E.164
|
||||
return f"+7999000{next(_phone_seq):03d}"
|
||||
|
||||
def make_user(
|
||||
email: str,
|
||||
*,
|
||||
staff: bool = False,
|
||||
superuser: bool = False,
|
||||
verified: bool = True,
|
||||
active: bool = True,
|
||||
**extra,
|
||||
) -> User:
|
||||
"""Helper для создания пользователей в тестах."""
|
||||
"""
|
||||
Создаём пользователя напрямую (без менеджера, чтобы не слать почту).
|
||||
"""
|
||||
u = User.objects.create(
|
||||
email=email,
|
||||
phone_number=extra.pop("phone_number", _unique_phone()),
|
||||
first_name=extra.pop("first_name", "FN"),
|
||||
last_name=extra.pop("last_name", "LN"),
|
||||
is_staff=staff,
|
||||
is_superuser=superuser,
|
||||
is_active=active,
|
||||
email_verified=verified,
|
||||
**extra,
|
||||
)
|
||||
u.set_password("pass")
|
||||
u.save()
|
||||
return u
|
||||
|
||||
def extract_results(data):
|
||||
if isinstance(data, dict) and "results" in data:
|
||||
return data["results"]
|
||||
return data
|
||||
|
||||
# ---------- basic auth / list / retrieve ----------
|
||||
|
||||
def test_list_requires_auth(api_client: APIClient):
|
||||
@ -150,7 +107,8 @@ def test_filter_by_department_includes_members_and_head(api_client: APIClient):
|
||||
outsider = make_user("out@example.com", last_name="Out")
|
||||
|
||||
d = Department.objects.create(name="Dept", head=head)
|
||||
EmployeeDepartment.objects.create(employee=member, department=d, is_active=True)
|
||||
EmployeeDepartment.objects.create(
|
||||
employee=member, department=d, is_active=True)
|
||||
|
||||
url = reverse("api:v1:employees-list")
|
||||
r = api_client.get(url, {"department": d.id})
|
||||
@ -231,7 +189,8 @@ def test_filter_actually_active_logic(api_client: APIClient):
|
||||
# e_dismissed: email_verified=True, последнее действие — DISMISSED => не активен
|
||||
e_d = make_user("d@example.com", verified=True, active=True)
|
||||
EmployeeAction.objects.create(
|
||||
employee=e_d, action=ACTION_DISMISSED, date=dt.datetime.now(dt.timezone.utc)
|
||||
employee=e_d, action=ACTION_DISMISSED, date=dt.datetime.now(
|
||||
dt.timezone.utc)
|
||||
)
|
||||
|
||||
# e_unver: email_verified=False, даже при active=True — не включается при actually_active=true
|
||||
@ -265,7 +224,8 @@ def test_create_requires_staff(api_client: APIClient):
|
||||
# контакт специально НЕ передаём -> 400 из-за валидации контактов
|
||||
}
|
||||
resp = api_client.post(url, payload, format="json")
|
||||
assert resp.status_code in (status.HTTP_403_FORBIDDEN, status.HTTP_401_UNAUTHORIZED)
|
||||
assert resp.status_code in (
|
||||
status.HTTP_403_FORBIDDEN, status.HTTP_401_UNAUTHORIZED)
|
||||
|
||||
# staff -> но без контактов всё равно 400
|
||||
staff = make_user("staff@example.com", staff=True)
|
||||
|
||||
@ -9,6 +9,7 @@ from employees.models import (Department, DepartmentRole, EmployeeDepartment,
|
||||
from rest_framework import status
|
||||
from rest_framework.test import APIClient
|
||||
from tests.conftest import _unique_phone
|
||||
from tests.api.v1.employees.test_helpers import make_user, grant_permission, make_department, extract_results
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
User = get_user_model()
|
||||
@ -21,37 +22,6 @@ def api():
|
||||
|
||||
_phone_seq = itertools.count(3000)
|
||||
|
||||
@pytest.fixture
|
||||
def make_user(
|
||||
email: str,
|
||||
*,
|
||||
staff: bool = False,
|
||||
superuser: bool = False,
|
||||
verified: bool = True,
|
||||
active: bool = True,
|
||||
**extra,
|
||||
) -> User:
|
||||
"""Fixture для создания пользователей."""
|
||||
u = User.objects.create(
|
||||
email=email,
|
||||
phone_number=extra.pop("phone_number", _unique_phone()),
|
||||
first_name=extra.pop("first_name", "FN"),
|
||||
last_name=extra.pop("last_name", "LN"),
|
||||
is_staff=staff,
|
||||
is_superuser=superuser,
|
||||
is_active=active,
|
||||
email_verified=verified,
|
||||
**extra,
|
||||
)
|
||||
u.set_password("pass")
|
||||
u.save()
|
||||
return u
|
||||
|
||||
def extract_results(data):
|
||||
if isinstance(data, dict) and "results" in data:
|
||||
return data["results"]
|
||||
return data
|
||||
|
||||
def _get_item(items, emp_id):
|
||||
for it in items:
|
||||
if it.get("id") == emp_id:
|
||||
|
||||
@ -4,10 +4,10 @@
|
||||
Используйте эти функции вместо создания собственных в каждом тестовом файле.
|
||||
"""
|
||||
import itertools
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.contrib.auth.models import Permission
|
||||
from employees.models import Employee, Department, DepartmentRole, Position
|
||||
|
||||
from employees.models import Department, DepartmentRole, Employee, Position
|
||||
from tests.test_config import DEFAULT_PASSWORD, TEST_EMAIL_DOMAIN
|
||||
|
||||
User = get_user_model()
|
||||
@ -39,7 +39,7 @@ def make_user(
|
||||
) -> User:
|
||||
"""
|
||||
Создает пользователя для тестов.
|
||||
|
||||
|
||||
Args:
|
||||
email: Email пользователя (если None, генерируется автоматически)
|
||||
staff: Флаг is_staff
|
||||
@ -48,22 +48,22 @@ def make_user(
|
||||
active: Флаг is_active
|
||||
password: Пароль (по умолчанию DEFAULT_PASSWORD)
|
||||
**extra: Дополнительные поля
|
||||
|
||||
|
||||
Returns:
|
||||
User: Созданный пользователь
|
||||
"""
|
||||
if email is None:
|
||||
email = make_unique_email()
|
||||
|
||||
|
||||
if 'phone_number' not in extra:
|
||||
extra['phone_number'] = make_unique_phone()
|
||||
|
||||
|
||||
if 'first_name' not in extra:
|
||||
extra['first_name'] = 'Test'
|
||||
|
||||
|
||||
if 'last_name' not in extra:
|
||||
extra['last_name'] = 'User'
|
||||
|
||||
|
||||
user = User.objects.create(
|
||||
email=email,
|
||||
is_staff=staff,
|
||||
@ -74,14 +74,14 @@ def make_user(
|
||||
)
|
||||
user.set_password(password)
|
||||
user.save()
|
||||
|
||||
|
||||
return user
|
||||
|
||||
|
||||
def grant_permission(user: User, permission_codename: str):
|
||||
"""
|
||||
Выдает пользователю указанное разрешение.
|
||||
|
||||
|
||||
Args:
|
||||
user: Пользователь
|
||||
permission_codename: Codename разрешения в формате 'app_label.codename'
|
||||
@ -96,7 +96,7 @@ def grant_permission(user: User, permission_codename: str):
|
||||
else:
|
||||
# Пытаемся найти по codename
|
||||
perm = Permission.objects.get(codename=permission_codename)
|
||||
|
||||
|
||||
user.user_permissions.add(perm)
|
||||
user.save()
|
||||
|
||||
@ -104,49 +104,49 @@ def grant_permission(user: User, permission_codename: str):
|
||||
def make_department(**kwargs) -> Department:
|
||||
"""
|
||||
Создает отдел для тестов.
|
||||
|
||||
|
||||
Args:
|
||||
**kwargs: Поля модели Department
|
||||
|
||||
|
||||
Returns:
|
||||
Department: Созданный отдел
|
||||
"""
|
||||
if 'name' not in kwargs:
|
||||
kwargs['name'] = f"Test Department {Department.objects.count() + 1}"
|
||||
|
||||
|
||||
return Department.objects.create(**kwargs)
|
||||
|
||||
|
||||
def make_position(**kwargs) -> Position:
|
||||
"""
|
||||
Создает должность для тестов.
|
||||
|
||||
|
||||
Args:
|
||||
**kwargs: Поля модели Position
|
||||
|
||||
|
||||
Returns:
|
||||
Position: Созданная должность
|
||||
"""
|
||||
if 'name' not in kwargs:
|
||||
kwargs['name'] = f"Test Position {Position.objects.count() + 1}"
|
||||
|
||||
|
||||
return Position.objects.create(**kwargs)
|
||||
|
||||
|
||||
def make_department_role(department: Department, **kwargs) -> DepartmentRole:
|
||||
"""
|
||||
Создает роль отдела для тестов.
|
||||
|
||||
|
||||
Args:
|
||||
department: Отдел
|
||||
**kwargs: Дополнительные поля
|
||||
|
||||
|
||||
Returns:
|
||||
DepartmentRole: Созданная роль
|
||||
"""
|
||||
if 'name' not in kwargs:
|
||||
kwargs['name'] = f"Test Role {DepartmentRole.objects.count() + 1}"
|
||||
|
||||
|
||||
return DepartmentRole.objects.create(
|
||||
department=department,
|
||||
**kwargs
|
||||
@ -156,10 +156,10 @@ def make_department_role(department: Department, **kwargs) -> DepartmentRole:
|
||||
def extract_results(data):
|
||||
"""
|
||||
Извлекает результаты из пагинированного ответа API.
|
||||
|
||||
|
||||
Args:
|
||||
data: Данные ответа (dict или list)
|
||||
|
||||
|
||||
Returns:
|
||||
list: Список результатов
|
||||
"""
|
||||
|
||||
@ -13,31 +13,13 @@ from django.urls import reverse
|
||||
from rest_framework import status
|
||||
from rest_framework.test import APIClient
|
||||
from tests.conftest import _unique_phone
|
||||
from tests.api.v1.employees.test_helpers import make_user, grant_permission, make_department, extract_results
|
||||
|
||||
User = get_user_model()
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
_phone_seq = itertools.count(5000)
|
||||
|
||||
@pytest.fixture
|
||||
def make_user(email, **kwargs):
|
||||
"""Fixture для создания пользователей."""
|
||||
"""Создаёт пользователя напрямую"""
|
||||
u = User.objects.create(
|
||||
email=email,
|
||||
phone_number=kwargs.pop("phone_number", _unique_phone()),
|
||||
first_name=kwargs.pop("first_name", "Test"),
|
||||
last_name=kwargs.pop("last_name", "User"),
|
||||
is_staff=kwargs.pop("staff", False),
|
||||
is_superuser=kwargs.pop("superuser", False),
|
||||
is_active=kwargs.pop("active", True),
|
||||
email_verified=kwargs.pop("verified", True),
|
||||
**kwargs
|
||||
)
|
||||
u.set_password("pass")
|
||||
u.save()
|
||||
return u
|
||||
|
||||
@pytest.fixture
|
||||
def api_client():
|
||||
return APIClient()
|
||||
|
||||
@ -8,15 +8,12 @@ from django.urls import reverse
|
||||
from employees.models import Employee, Position
|
||||
from rest_framework import status
|
||||
from tests.conftest import _unique_phone
|
||||
from tests.api.v1.employees.test_helpers import make_user, grant_permission, make_department, extract_results
|
||||
|
||||
# ---------- helpers ----------
|
||||
|
||||
def _grant_perm(user, code: str):
|
||||
app_label, codename = code.split(".")
|
||||
perm = Permission.objects.get(content_type__app_label=app_label, codename=codename)
|
||||
user.user_permissions.add(perm)
|
||||
|
||||
# Сброс кэшей пермишенов у этого инстанса
|
||||
def _flush_perm_cache(user):
|
||||
"""Сброс кэшей пермишенов у этого инстанса"""
|
||||
for attr in ("_perm_cache", "_user_perm_cache", "_group_perm_cache"):
|
||||
if hasattr(user, attr):
|
||||
delattr(user, attr)
|
||||
@ -25,28 +22,8 @@ def _grant_perm(user, code: str):
|
||||
User = get_user_model()
|
||||
return User.objects.get(pk=user.pk)
|
||||
|
||||
|
||||
def _make_user(email: str, staff: bool = False, superuser: bool = False) -> Employee:
|
||||
# Создаём пользователя через менеджер, контакты не требуются (clean не вызывается на save())
|
||||
u = Employee.objects.create_user(
|
||||
email=email,
|
||||
password="pass12345",
|
||||
phone_number=_unique_phone(),
|
||||
send_activation_email=False,
|
||||
first_name="T",
|
||||
last_name="U",
|
||||
)
|
||||
u.is_staff = staff
|
||||
u.is_superuser = superuser
|
||||
u.email_verified = True
|
||||
u.is_active = True
|
||||
u.save(update_fields=["is_staff", "is_superuser", "email_verified", "is_active"])
|
||||
return u
|
||||
|
||||
|
||||
# ---------- tests ----------
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_list_requires_auth(api_client):
|
||||
url = reverse("api:v1:positions-list")
|
||||
@ -56,7 +33,7 @@ def test_list_requires_auth(api_client):
|
||||
assert resp.status_code in (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN)
|
||||
|
||||
# auth -> 200
|
||||
user = _make_user("user@example.com")
|
||||
user = make_user("user@example.com")
|
||||
api_client.force_authenticate(user=user)
|
||||
Position.objects.create(name="Manager")
|
||||
Position.objects.create(name="Engineer")
|
||||
@ -67,36 +44,34 @@ def test_list_requires_auth(api_client):
|
||||
names = [p["name"] for p in resp.data]
|
||||
assert {"Manager", "Engineer"} <= set(names)
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_create_permissions(api_client):
|
||||
url = reverse("api:v1:positions-list")
|
||||
|
||||
# обычный пользователь без прав -> 403
|
||||
user = _make_user("user@example.com")
|
||||
user = make_user("user@example.com")
|
||||
api_client.force_authenticate(user=user)
|
||||
resp = api_client.post(url, {"name": "NewPos"}, format="json")
|
||||
assert resp.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
# с пермом add_position -> 201
|
||||
_grant_perm(user, "employees.add_position")
|
||||
grant_permission(user, "employees.add_position")
|
||||
resp = api_client.post(url, {"name": "NewPos"}, format="json")
|
||||
assert resp.status_code == status.HTTP_201_CREATED
|
||||
assert resp.data["name"] == "NewPos"
|
||||
|
||||
# staff без специальных прав -> 201 (staff допускается)
|
||||
staff = _make_user("staff@example.com", staff=True)
|
||||
staff = make_user("staff@example.com", staff=True)
|
||||
api_client.force_authenticate(user=staff)
|
||||
resp = api_client.post(url, {"name": "StaffPos"}, format="json")
|
||||
assert resp.status_code == status.HTTP_201_CREATED
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_update_permissions_and_search_ordering(api_client):
|
||||
p1 = Position.objects.create(name="Alpha", description="one")
|
||||
p2 = Position.objects.create(name="Beta", description="two")
|
||||
|
||||
user = _make_user("user@example.com")
|
||||
user = make_user("user@example.com")
|
||||
api_client.force_authenticate(user=user)
|
||||
|
||||
# без прав change_position -> 403
|
||||
@ -105,7 +80,7 @@ def test_update_permissions_and_search_ordering(api_client):
|
||||
assert resp.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
# выдаём change_position -> 200
|
||||
_grant_perm(user, "employees.change_position")
|
||||
grant_permission(user, "employees.change_position")
|
||||
resp = api_client.patch(url_detail, {"description": "updated"}, format="json")
|
||||
assert resp.status_code == status.HTTP_200_OK
|
||||
p1.refresh_from_db()
|
||||
@ -123,11 +98,10 @@ def test_update_permissions_and_search_ordering(api_client):
|
||||
names = [row["name"] for row in resp.data]
|
||||
assert names == sorted(names, reverse=True)
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_delete_permissions(api_client):
|
||||
pos = Position.objects.create(name="ToDelete")
|
||||
user = _make_user("user@example.com")
|
||||
user = make_user("user@example.com")
|
||||
api_client.force_authenticate(user=user)
|
||||
url_detail = reverse("api:v1:positions-detail", args=[pos.id])
|
||||
|
||||
@ -136,35 +110,33 @@ def test_delete_permissions(api_client):
|
||||
assert resp.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
# с delete_position -> 204
|
||||
_grant_perm(user, "employees.delete_position")
|
||||
grant_permission(user, "employees.delete_position")
|
||||
resp = api_client.delete(url_detail)
|
||||
assert resp.status_code == status.HTTP_204_NO_CONTENT
|
||||
assert not Position.objects.filter(id=pos.id).exists()
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_set_groups_requires_special_perm(api_client):
|
||||
pos = Position.objects.create(name="RoleOwner")
|
||||
grp1 = Group.objects.create(name="grp1")
|
||||
grp2 = Group.objects.create(name="grp2")
|
||||
|
||||
user = _make_user("user@example.com")
|
||||
user = make_user("user@example.com")
|
||||
api_client.force_authenticate(user=user)
|
||||
|
||||
url_set = reverse("api:v1:positions-set-groups", args=[pos.id])
|
||||
|
||||
# даже с change_position НЕЛЬЗЯ менять группы — нужен assign_position_groups
|
||||
_grant_perm(user, "employees.change_position")
|
||||
grant_permission(user, "employees.change_position")
|
||||
resp = api_client.post(url_set, {"groups": [grp1.id, grp2.id]}, format="json")
|
||||
assert resp.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
# выдаём спец-пермиссию -> 200
|
||||
_grant_perm(user, "employees.assign_position_groups")
|
||||
grant_permission(user, "employees.assign_position_groups")
|
||||
resp = api_client.post(url_set, {"groups": [grp1.id, grp2.id]}, format="json")
|
||||
assert resp.status_code == status.HTTP_200_OK
|
||||
assert set(pos.groups.values_list("id", flat=True)) == {grp1.id, grp2.id}
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_add_and_remove_groups_and_permissions_endpoint(api_client):
|
||||
# используем гарантированно существующие права
|
||||
@ -183,9 +155,9 @@ def test_add_and_remove_groups_and_permissions_endpoint(api_client):
|
||||
|
||||
pos = Position.objects.create(name="Aggregator")
|
||||
|
||||
user = _make_user("manager@example.com")
|
||||
user = make_user("manager@example.com")
|
||||
api_client.force_authenticate(user=user)
|
||||
_grant_perm(user, "employees.assign_position_groups")
|
||||
grant_permission(user, "employees.assign_position_groups")
|
||||
|
||||
# add_groups
|
||||
url_add = reverse("api:v1:positions-add-groups", args=[pos.id])
|
||||
@ -219,13 +191,12 @@ def test_add_and_remove_groups_and_permissions_endpoint(api_client):
|
||||
assert "groups_verbose" in resp.data
|
||||
assert any(g["name"] == "B" for g in resp.data["groups_verbose"])
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_groups_payload_validation(api_client):
|
||||
pos = Position.objects.create(name="Validator")
|
||||
user = _make_user("user@example.com")
|
||||
user = make_user("user@example.com")
|
||||
api_client.force_authenticate(user=user)
|
||||
_grant_perm(user, "employees.assign_position_groups")
|
||||
grant_permission(user, "employees.assign_position_groups")
|
||||
|
||||
url_set = reverse("api:v1:positions-set-groups", args=[pos.id])
|
||||
|
||||
|
||||
@ -11,22 +11,12 @@ from employees.models import (
|
||||
EmployeeDepartment, RoleAssignment,
|
||||
)
|
||||
from tests.conftest import _unique_phone
|
||||
from tests.api.v1.employees.test_helpers import make_user, grant_permission, make_department, extract_results
|
||||
|
||||
User = get_user_model()
|
||||
# В этом проекте User == Employee (кастомная модель)
|
||||
Employee = User
|
||||
|
||||
@pytest.fixture
|
||||
def make_user(email: str, is_staff: bool = False) -> User:
|
||||
"""Создаёт пользователя (Employee)."""
|
||||
return User.objects.create_user(
|
||||
email=email,
|
||||
password="pwd12345",
|
||||
phone_number=_unique_phone(),
|
||||
is_staff=is_staff,
|
||||
send_activation_email=False,
|
||||
)
|
||||
|
||||
def make_dept(name: str, head: User = None) -> Department:
|
||||
"""Создаёт отдел."""
|
||||
return Department.objects.create(name=name, head=head)
|
||||
|
||||
@ -5,6 +5,7 @@ from employees.models import Employee, Skill
|
||||
from rest_framework import status
|
||||
from tests.conftest import _unique_email, _unique_phone
|
||||
from tests.test_config import DEFAULT_PASSWORD
|
||||
from tests.api.v1.employees.test_helpers import make_user, grant_permission, make_department, extract_results
|
||||
|
||||
# --- helpers ---
|
||||
|
||||
@ -24,24 +25,8 @@ def _mk_user(staff=False, superuser=False) -> Employee:
|
||||
u.save(update_fields=["is_staff", "is_superuser", "email_verified", "is_active"])
|
||||
return u
|
||||
|
||||
|
||||
def _grant_perm(user: Employee, perm_code: str):
|
||||
app_label, codename = perm_code.split(".", 1)
|
||||
p = Permission.objects.get(content_type__app_label=app_label, codename=codename)
|
||||
user.user_permissions.add(p)
|
||||
user.save()
|
||||
for a in ("_perm_cache", "_user_perm_cache", "_group_perm_cache"):
|
||||
if hasattr(user, a):
|
||||
try:
|
||||
delattr(user, a)
|
||||
except:
|
||||
pass
|
||||
return p
|
||||
|
||||
|
||||
# --- tests ---
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_list_search_ordering_requires_auth(api_client):
|
||||
url = reverse("api:v1:skills-list")
|
||||
@ -71,7 +56,6 @@ def test_list_search_ordering_requires_auth(api_client):
|
||||
names = [r["name"] for r in resp.data]
|
||||
assert names == sorted(names, reverse=True)
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_create_allowed_for_regular_user(api_client):
|
||||
user = _mk_user()
|
||||
@ -81,7 +65,6 @@ def test_create_allowed_for_regular_user(api_client):
|
||||
assert resp.status_code == 201
|
||||
assert Skill.objects.filter(name="Go").exists()
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_update_delete_permissions(api_client):
|
||||
user = _mk_user()
|
||||
@ -98,19 +81,18 @@ def test_update_delete_permissions(api_client):
|
||||
assert resp.status_code == 403
|
||||
|
||||
# с правом change_skill -> PATCH разрешён
|
||||
_grant_perm(user, "employees.change_skill")
|
||||
grant_permission(user, "employees.change_skill")
|
||||
resp = api_client.patch(url_det, {"name": "Docker+"}, format="json")
|
||||
assert resp.status_code == 200
|
||||
s.refresh_from_db()
|
||||
assert s.name == "Docker+"
|
||||
|
||||
# с правом delete_skill -> DELETE разрешён
|
||||
_grant_perm(user, "employees.delete_skill")
|
||||
grant_permission(user, "employees.delete_skill")
|
||||
resp = api_client.delete(url_det)
|
||||
assert resp.status_code == 204
|
||||
assert not Skill.objects.filter(id=s.id).exists()
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_staff_bypasses_permissions(api_client):
|
||||
staff = _mk_user(staff=True)
|
||||
@ -123,7 +105,6 @@ def test_staff_bypasses_permissions(api_client):
|
||||
)
|
||||
assert api_client.delete(url_det).status_code == 204
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_bulk_create_by_regular_user(api_client):
|
||||
user = _mk_user()
|
||||
@ -139,7 +120,6 @@ def test_bulk_create_by_regular_user(api_client):
|
||||
created_names = {row["name"] for row in data["created"]}
|
||||
assert created_names == {"Python", "SQL"}
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_merge_requires_perm_and_reassigns(api_client):
|
||||
user = _mk_user()
|
||||
@ -161,7 +141,7 @@ def test_merge_requires_perm_and_reassigns(api_client):
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
_grant_perm(user, "employees.change_skill")
|
||||
grant_permission(user, "employees.change_skill")
|
||||
resp = api_client.post(
|
||||
url, {"source_id": src.id, "target_id": dst.id}, format="json"
|
||||
)
|
||||
@ -174,12 +154,11 @@ def test_merge_requires_perm_and_reassigns(api_client):
|
||||
id=src.id
|
||||
).exists() # по умолчанию delete_source=True
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_merge_same_id_and_invalid_ids(api_client):
|
||||
user = _mk_user()
|
||||
api_client.force_authenticate(user=user)
|
||||
_grant_perm(user, "employees.change_skill")
|
||||
grant_permission(user, "employees.change_skill")
|
||||
s = Skill.objects.create(name="A")
|
||||
|
||||
url = reverse("api:v1:skills-merge")
|
||||
@ -191,12 +170,11 @@ def test_merge_same_id_and_invalid_ids(api_client):
|
||||
resp = api_client.post(url, {"source_id": 999999, "target_id": s.id}, format="json")
|
||||
assert resp.status_code in (404,)
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_merge_without_reassign_or_delete(api_client):
|
||||
user = _mk_user()
|
||||
api_client.force_authenticate(user=user)
|
||||
_grant_perm(user, "employees.change_skill")
|
||||
grant_permission(user, "employees.change_skill")
|
||||
|
||||
src = Skill.objects.create(name="ML")
|
||||
dst = Skill.objects.create(name="Machine Learning")
|
||||
@ -223,12 +201,11 @@ def test_merge_without_reassign_or_delete(api_client):
|
||||
# источник не удалили
|
||||
assert Skill.objects.filter(id=src.id).exists()
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_merge_idempotent_if_employee_already_has_target(api_client):
|
||||
user = _mk_user()
|
||||
api_client.force_authenticate(user=user)
|
||||
_grant_perm(user, "employees.change_skill")
|
||||
grant_permission(user, "employees.change_skill")
|
||||
|
||||
src = Skill.objects.create(name="Golang")
|
||||
dst = Skill.objects.create(name="Go")
|
||||
|
||||
@ -11,7 +11,8 @@
|
||||
import itertools
|
||||
import mimetypes
|
||||
import os
|
||||
from datetime import datetime, date
|
||||
from datetime import date, datetime
|
||||
|
||||
import pytest
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.contrib.auth.models import Permission
|
||||
@ -19,17 +20,11 @@ from django.contrib.contenttypes.models import ContentType
|
||||
from django.core.files.base import ContentFile
|
||||
from employees.models import Department, DepartmentRole, EmployeeDepartment
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from tests.test_config import (
|
||||
DEFAULT_PASSWORD,
|
||||
DEFAULT_TEST_PHONE_PREFIX,
|
||||
TEST_EMAIL_DOMAIN,
|
||||
TEST_IMAGE_1X1_PNG_B64,
|
||||
TEST_IMAGE_DATA_URI,
|
||||
)
|
||||
|
||||
# Импортируем LDAP фикстуры
|
||||
from tests.ldap_fixtures import * # noqa
|
||||
from tests.test_config import (DEFAULT_PASSWORD, DEFAULT_TEST_PHONE_PREFIX,
|
||||
TEST_EMAIL_DOMAIN, TEST_IMAGE_1X1_PNG_B64,
|
||||
TEST_IMAGE_DATA_URI)
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
@ -61,7 +56,8 @@ def _unique_email(prefix: str = "user") -> str:
|
||||
@pytest.fixture(autouse=True)
|
||||
def _fast_hashers_and_email(settings, tmp_path):
|
||||
# быстрые хеши паролей и локальная почта
|
||||
settings.PASSWORD_HASHERS = ["django.contrib.auth.hashers.MD5PasswordHasher"]
|
||||
settings.PASSWORD_HASHERS = [
|
||||
"django.contrib.auth.hashers.MD5PasswordHasher"]
|
||||
settings.EMAIL_BACKEND = "django.core.mail.backends.locmem.EmailBackend"
|
||||
# изоляция медиа
|
||||
media_root = tmp_path / "test_media"
|
||||
@ -103,11 +99,11 @@ def auth_client_factory():
|
||||
def user_factory():
|
||||
"""
|
||||
Фабрика для создания пользователей.
|
||||
|
||||
|
||||
Пример использования:
|
||||
user = user_factory(email="x@example.com", staff=False, superuser=False,
|
||||
verified=True, active=True, **extra)
|
||||
|
||||
|
||||
Создаёт пользователя напрямую (без менеджера create_user, чтобы не слать письма).
|
||||
"""
|
||||
|
||||
@ -123,7 +119,7 @@ def user_factory():
|
||||
):
|
||||
if email is None:
|
||||
email = _unique_email()
|
||||
|
||||
|
||||
u = User.objects.create(
|
||||
email=email,
|
||||
phone_number=extra.pop("phone_number", _unique_phone()),
|
||||
|
||||
@ -7,6 +7,7 @@
|
||||
- ldap_config: конфигурация LDAP для тестов
|
||||
"""
|
||||
import os
|
||||
|
||||
import pytest
|
||||
from django.conf import settings
|
||||
|
||||
@ -15,13 +16,13 @@ from django.conf import settings
|
||||
def ldap_available():
|
||||
"""
|
||||
Проверяет доступность LDAP сервера.
|
||||
|
||||
|
||||
Returns:
|
||||
bool: True если LDAP доступен и настроен
|
||||
"""
|
||||
if not settings.LDAP_ENABLED:
|
||||
return False
|
||||
|
||||
|
||||
try:
|
||||
from employees.ldap.infrastructure.connections import _ldap
|
||||
with _ldap() as conn:
|
||||
@ -77,20 +78,20 @@ def ensure_ldap_disabled(settings):
|
||||
def ldap_cleanup():
|
||||
"""
|
||||
Cleanup фикстура для удаления тестовых данных из LDAP после теста.
|
||||
|
||||
|
||||
Usage:
|
||||
def test_something(ldap_cleanup):
|
||||
# создаем объекты в LDAP
|
||||
ldap_cleanup.add_for_deletion(dn)
|
||||
"""
|
||||
cleanup_list = []
|
||||
|
||||
|
||||
class Cleanup:
|
||||
def add_for_deletion(self, dn):
|
||||
cleanup_list.append(dn)
|
||||
|
||||
|
||||
yield Cleanup()
|
||||
|
||||
|
||||
# Cleanup после теста
|
||||
if cleanup_list and settings.LDAP_ENABLED:
|
||||
try:
|
||||
@ -129,9 +130,9 @@ def pytest_collection_modifyitems(config, items):
|
||||
ldap_available = conn.bind()
|
||||
except:
|
||||
pass
|
||||
|
||||
|
||||
skip_ldap = pytest.mark.skip(reason="LDAP сервер недоступен")
|
||||
|
||||
|
||||
for item in items:
|
||||
if "ldap_required" in item.keywords and not ldap_available:
|
||||
item.add_marker(skip_ldap)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user