Robotail/Corp
Robotail/Corp
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Add procurement refactoring plan and merge migrations

Browse Source
This commit is contained in:
igor_nadein 2026-02-27 17:55:42 +03:00
parent ce69498db2
commit 8ff3ffcd67
14 changed files with 2974 additions and 440 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
backend/api/v1/urls.py
View File

@ -81,6 +81,8 @@ urlpatterns = [
),
# Notifications API
path("notifications/", include("api.v1.notifications.urls")),
# Procurement API
path("procurement/", include("api.v1.procurement.urls")),
# Router URLs (включая все ViewSets)
path("", include(router.urls)),
]

14
backend/employees/migrations/0040_merge_20260227_1639.py Normal file
View File

@ -0,0 +1,14 @@
# Generated by Django 5.2.4 on 2026-02-27 13:39
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('employees', '0035_add_manage_equipment_permission'),
('employees', '0039_employee_employees_e_usernam_4d3ddc_idx'),
]
operations = [
]

14
backend/notifications/migrations/0007_merge_20260227_1639.py Normal file
View File

@ -0,0 +1,14 @@
# Generated by Django 5.2.4 on 2026-02-27 13:39
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('notifications', '0005_procurement_notification_types'),
('notifications', '0006_add_device_name'),
]
operations = [
]

52
backend/test_corporate_ip.py
View File

@ -1,52 +0,0 @@
#!/usr/bin/env python
"""Проверка разрешения IP для корпоративной сети 172.11.0.0/16"""
import os
import sys
sys.path.insert(0, os.path.dirname(__file__))
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'eusrr_backend.settings')
import django
django.setup()
from common.ip_restrictions import is_ip_allowed
print("\n" + "=" * 60)
print("Проверка IP-адресов для корпоративной сети 172.11.0.0/16")
print("=" * 60)
test_ips = [
('172.11.0.117', True, 'Ваш текущий IP'),
('172.11.0.1', True, 'Корпоративная сеть 172.11.x.x'),
('172.11.255.255', True, 'Край диапазона 172.11.x.x'),
('172.12.0.1', False, 'За пределами 172.11.x.x'),
('172.16.0.1', True, 'Стандартная приватная сеть'),
('192.168.1.1', True, 'Локальная сеть'),
('10.0.0.1', True, 'Приватная сеть класса A'),
('8.8.8.8', False, 'Публичный IP (Google DNS)'),
]
passed = 0
failed = 0
for ip, expected, description in test_ips:
result = is_ip_allowed(ip)
status = "✅" if result == expected else "❌"
if result == expected:
passed += 1
else:
failed += 1
result_text = "разрешен" if result else "заблокирован"
expected_text = "разрешен" if expected else "заблокирован"
print(f"{status} {ip:20} -> {result_text:12} "
f"(ожидалось: {expected_text:12}) # {description}")
print("\n" + "=" * 60)
if failed == 0:
print(f"✅ Все тесты пройдены: {passed}/{passed + failed}")
else:
print(f"⚠️ Тестов пройдено: {passed}/{passed + failed}, провалено: {failed}")
print("=" * 60 + "\n")

171
backend/test_ip_restrictions.py
View File

@ -1,171 +0,0 @@
#!/usr/bin/env python
"""
Тестирование IP-ограничений для регистрации.
Запуск:
python backend/test_ip_restrictions.py
"""
import os
import sys
import django
# Настройка Django
sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
os.environ.setdefault("DJANGO_SECRET_KEY", "test-key")
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "eusrr_backend.settings")
django.setup()
from common.ip_restrictions import (
get_client_ip,
is_local_ip,
is_ip_allowed,
)
def test_is_local_ip():
"""Тест проверки локальных IP."""
print("\n=== Тест: is_local_ip ===")
test_cases = [
('127.0.0.1', True, 'localhost IPv4'),
('127.0.0.100', True, 'localhost range'),
('192.168.1.1', True, 'private class C'),
('192.168.100.50', True, 'private class C'),
('10.0.0.1', True, 'private class A'),
('172.16.0.1', True, 'private class B'),
('172.31.255.255', True, 'private class B edge'),
('8.8.8.8', False, 'public Google DNS'),
('203.0.113.50', False, 'public TEST-NET'),
('::1', True, 'localhost IPv6'),
('invalid-ip', False, 'invalid IP'),
]
passed = 0
failed = 0
for ip, expected, description in test_cases:
result = is_local_ip(ip)
status = "✅" if result == expected else "❌"
if result == expected:
passed += 1
else:
failed += 1
print(f"{status} {ip:20} -> {result:5} (expected: {expected:5}) # {description}")
print(f"\nПройдено: {passed}/{passed + failed}")
return failed == 0
def test_is_ip_allowed():
"""Тест проверки разрешенных IP с кастомными настройками."""
print("\n=== Тест: is_ip_allowed ===")
# Тест 1: Разрешить все
print("\n1. Разрешить все IP ['*']:")
result = is_ip_allowed('203.0.113.50', ['*'])
print(f" {'✅' if result else '❌'} 203.0.113.50 -> {result} (expected: True)")
# Тест 2: Разрешить конкретную сеть
print("\n2. Разрешить сеть 192.168.1.0/24:")
allowed_net = ['192.168.1.0/24']
tests = [
('192.168.1.1', True),
('192.168.1.100', True),
('192.168.1.255', True),
('192.168.2.1', False),
('10.0.0.1', False),
]
for ip, expected in tests:
result = is_ip_allowed(ip, allowed_net)
status = "✅" if result == expected else "❌"
print(f" {status} {ip:20} -> {result:5} (expected: {expected:5})")
# Тест 3: Разрешить конкретные IP
print("\n3. Разрешить конкретные IP ['192.168.1.100', '10.0.0.5']:")
allowed_ips = ['192.168.1.100', '10.0.0.5']
tests = [
('192.168.1.100', True),
('10.0.0.5', True),
('192.168.1.101', False),
('10.0.0.6', False),
]
for ip, expected in tests:
result = is_ip_allowed(ip, allowed_ips)
status = "✅" if result == expected else "❌"
print(f" {status} {ip:20} -> {result:5} (expected: {expected:5})")
# Тест 4: По умолчанию (только локальные)
print("\n4. По умолчанию (None = только локальные IP):")
tests = [
('127.0.0.1', True),
('192.168.1.1', True),
('10.0.0.1', True),
('8.8.8.8', False),
('203.0.113.50', False),
]
for ip, expected in tests:
result = is_ip_allowed(ip, None)
status = "✅" if result == expected else "❌"
print(f" {status} {ip:20} -> {result:5} (expected: {expected:5})")
def test_get_client_ip():
"""Тест получения IP из request."""
print("\n=== Тест: get_client_ip ===")
from django.test import RequestFactory
factory = RequestFactory()
# Тест 1: Обычный запрос
print("\n1. Обычный запрос (без прокси):")
request = factory.get('/')
ip = get_client_ip(request)
print(f" IP: {ip}")
# Тест 2: С заголовком X-Forwarded-For
print("\n2. Запрос через прокси (X-Forwarded-For):")
request = factory.get('/', HTTP_X_FORWARDED_FOR='203.0.113.50, 192.168.1.1')
ip = get_client_ip(request)
expected = '203.0.113.50'
status = "✅" if ip == expected else "❌"
print(f" {status} IP: {ip} (expected: {expected})")
# Тест 3: С одним IP в X-Forwarded-For
print("\n3. Запрос через прокси (один IP):")
request = factory.get('/', HTTP_X_FORWARDED_FOR='10.0.0.5')
ip = get_client_ip(request)
expected = '10.0.0.5'
status = "✅" if ip == expected else "❌"
print(f" {status} IP: {ip} (expected: {expected})")
def main():
"""Запуск всех тестов."""
print("=" * 60)
print("Тестирование модуля IP-ограничений")
print("=" * 60)
try:
test_is_local_ip()
test_is_ip_allowed()
test_get_client_ip()
print("\n" + "=" * 60)
print("✅ Все тесты выполнены")
print("=" * 60)
except Exception as e:
print(f"\n❌ Ошибка: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
main()

167
backend/test_ip_simple.py
View File

@ -1,167 +0,0 @@
#!/usr/bin/env python
"""
Простой тест IP-ограничений (без Django).
"""
import ipaddress
def is_local_ip(ip_address: str) -> bool:
"""Проверяет, является ли IP-адрес локальным."""
try:
ip = ipaddress.ip_address(ip_address)
return ip.is_private or ip.is_loopback or ip.is_link_local
except ValueError:
return False
def is_ip_allowed(ip_address: str, allowed_networks=None):
"""Проверяет, разрешен ли доступ с данного IP-адреса."""
if allowed_networks is None:
return is_local_ip(ip_address)
if not allowed_networks:
return False
if allowed_networks == ['*'] or '*' in allowed_networks:
return True
try:
ip = ipaddress.ip_address(ip_address)
for network_str in allowed_networks:
if '/' in network_str:
network = ipaddress.ip_network(network_str, strict=False)
if ip in network:
return True
else:
allowed_ip = ipaddress.ip_address(network_str)
if ip == allowed_ip:
return True
return False
except ValueError:
return False
def test_is_local_ip():
"""Тест проверки локальных IP."""
print("\n=== Тест: is_local_ip ===")
test_cases = [
('127.0.0.1', True, 'localhost IPv4'),
('127.0.0.100', True, 'localhost range'),
('192.168.1.1', True, 'private class C'),
('192.168.100.50', True, 'private class C'),
('10.0.0.1', True, 'private class A'),
('172.16.0.1', True, 'private class B'),
('172.31.255.255', True, 'private class B edge'),
('8.8.8.8', False, 'public Google DNS'),
('203.0.113.50', False, 'public TEST-NET'),
('::1', True, 'localhost IPv6'),
('invalid-ip', False, 'invalid IP'),
]
passed = 0
failed = 0
for ip, expected, description in test_cases:
result = is_local_ip(ip)
status = "✅" if result == expected else "❌"
if result == expected:
passed += 1
else:
failed += 1
print(f"{status} {ip:20} -> {str(result):5} "
f"(expected: {str(expected):5}) # {description}")
print(f"\nПройдено: {passed}/{passed + failed}")
return failed == 0
def test_is_ip_allowed():
"""Тест проверки разрешенных IP с кастомными настройками."""
print("\n=== Тест: is_ip_allowed ===")
# Тест 1: Разрешить все
print("\n1. Разрешить все IP ['*']:")
result = is_ip_allowed('203.0.113.50', ['*'])
status = '✅' if result else '❌'
print(f" {status} 203.0.113.50 -> {result} (expected: True)")
# Тест 2: Разрешить конкретную сеть
print("\n2. Разрешить сеть 192.168.1.0/24:")
allowed_net = ['192.168.1.0/24']
tests = [
('192.168.1.1', True),
('192.168.1.100', True),
('192.168.1.255', True),
('192.168.2.1', False),
('10.0.0.1', False),
]
for ip, expected in tests:
result = is_ip_allowed(ip, allowed_net)
status = "✅" if result == expected else "❌"
print(f" {status} {ip:20} -> {str(result):5} "
f"(expected: {str(expected):5})")
# Тест 3: Разрешить конкретные IP
print("\n3. Разрешить IP ['192.168.1.100', '10.0.0.5']:")
allowed_ips = ['192.168.1.100', '10.0.0.5']
tests = [
('192.168.1.100', True),
('10.0.0.5', True),
('192.168.1.101', False),
('10.0.0.6', False),
]
for ip, expected in tests:
result = is_ip_allowed(ip, allowed_ips)
status = "✅" if result == expected else "❌"
print(f" {status} {ip:20} -> {str(result):5} "
f"(expected: {str(expected):5})")
# Тест 4: По умолчанию (только локальные)
print("\n4. По умолчанию (None = только локальные IP):")
tests = [
('127.0.0.1', True),
('192.168.1.1', True),
('10.0.0.1', True),
('8.8.8.8', False),
('203.0.113.50', False),
]
for ip, expected in tests:
result = is_ip_allowed(ip, None)
status = "✅" if result == expected else "❌"
print(f" {status} {ip:20} -> {str(result):5} "
f"(expected: {str(expected):5})")
def main():
"""Запуск всех тестов."""
print("=" * 60)
print("Тестирование модуля IP-ограничений")
print("=" * 60)
try:
test_is_local_ip()
test_is_ip_allowed()
print("\n" + "=" * 60)
print("✅ Все тесты выполнены успешно")
print("=" * 60)
except Exception as e:
print(f"\n❌ Ошибка: {e}")
import traceback
traceback.print_exc()
return 1
return 0
if __name__ == "__main__":
exit(main())

50
backend/test_patch_import.py
View File

@ -1,50 +0,0 @@
#!/usr/bin/env python
"""Проверка импорта и применения патча."""
import os
import sys
import django
# Setup Django
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "eusrr_backend.settings")
print("\n" + "=" * 80)
print("1. ЗАПУСК DJANGO")
print("=" * 80)
django.setup()
print("\n" + "=" * 80)
print("2. ИМПОРТ Event БЕЗ ПАТЧА")
print("=" * 80)
from schedule.models import Event
print(f"Event._event_params: {Event._event_params}")
print(f"Имя метода: {Event._event_params.__name__}")
print("\n" + "=" * 80)
print("3. ИМПОРТ ПАТЧА")
print("=" * 80)
import schedule_patch # noqa
print("\n" + "=" * 80)
print("4. ПРОВЕРКА Event ПОСЛЕ ПАТЧА")
print("=" * 80)
print(f"Event._event_params: {Event._event_params}")
print(f"Имя метода: {Event._event_params.__name__}")
# Проверим на реальном событии
print("\n" + "=" * 80)
print("5. ТЕСТ НА РЕАЛЬНОМ СОБЫТИИ")
print("=" * 80)
event = Event.objects.filter(title__icontains='ыфвыфыф').first()
if event:
print(f"Event ID: {event.id}")
result = event._event_params()
print(f"_event_params() result: {result}")
if 'byweekday' in result:
print(f"byweekday = {result['byweekday']}")
if len(result['byweekday']) > 1:
print("✅ ПАТЧ РАБОТАЕТ - несколько дней сохранены!")
else:
print("❌ ПАТЧ НЕ РАБОТАЕТ - только один день!")
else:
print("Событие не найдено")

1
backend/tests/api/v1/procurement/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Тесты API v1 для модуля закупок."""

203
backend/tests/api/v1/procurement/conftest.py Normal file
View File

@ -0,0 +1,203 @@
"""
Общие fixtures для тестов API модуля procurement.
"""
import pytest
from decimal import Decimal
from datetime import date
from employees.models import Department, Employee, EmployeeDepartment
from procurement.models import (
Budget,
Equipment,
EquipmentCategory,
ProcurementItem,
ProcurementRequest,
Supplier,
)
from procurement.constants import (
ProcurementStatus,
UrgencyLevel,
EquipmentStatus,
)
# ==============================================================================
# ПОЛЬЗОВАТЕЛИ
# ==============================================================================
@pytest.fixture
def user_factory(db):
"""Фабрика для создания пользователей."""
def _create_user(email, **kwargs):
defaults = {
"password": "testpass123",
"phone_number": f"+7999{email[:7]}",
"first_name": "Тест",
"last_name": "Пользователь",
"send_activation_email": False,
}
defaults.update(kwargs)
return Employee.objects.create_user(email=email, **defaults)
return _create_user
@pytest.fixture
def link_factory(db):
"""Фабрика для связи пользователя с отделом."""
def _link(employee, department, **kwargs):
defaults = {"is_active": True}
defaults.update(kwargs)
return EmployeeDepartment.objects.create(
employee=employee,
department=department,
**defaults
)
return _link
# ==============================================================================
# ОТДЕЛЫ И БЮДЖЕТЫ
# ==============================================================================
@pytest.fixture
def department_factory(db):
"""Фабрика для создания отделов."""
def _create_department(name="Тестовый отдел", **kwargs):
defaults = {"description": f"Описание {name}"}
defaults.update(kwargs)
return Department.objects.create(name=name, **defaults)
return _create_department
@pytest.fixture
def budget_factory(db):
"""Фабрика для создания бюджетов."""
def _create_budget(department, year=2026, quarter=1, **kwargs):
defaults = {
"allocated_amount": Decimal("1000000.00"),
"spent_amount": Decimal("0.00"),
}
defaults.update(kwargs)
return Budget.objects.create(
department=department,
year=year,
quarter=quarter,
**defaults
)
return _create_budget
# ==============================================================================
# ЗАЯВКИ НА ЗАКУПКУ
# ==============================================================================
@pytest.fixture
def procurement_request_factory(db):
"""Фабрика для создания заявок на закупку."""
def _create_request(department, requestor, **kwargs):
defaults = {
"title": "Тестовая заявка",
"description": "Описание заявки",
"status": ProcurementStatus.DRAFT,
"urgency": UrgencyLevel.MEDIUM,
}
defaults.update(kwargs)
return ProcurementRequest.objects.create(
department=department,
requestor=requestor,
**defaults
)
return _create_request
@pytest.fixture
def procurement_item_factory(db):
"""Фабрика для создания позиций в заявке."""
def _create_item(request, **kwargs):
defaults = {
"name": "Тестовый товар",
"description": "Описание товара",
"quantity": 1,
"unit": "шт",
"estimated_unit_price": Decimal("10000.00"),
}
defaults.update(kwargs)
return ProcurementItem.objects.create(
request=request,
**defaults
)
return _create_item
# ==============================================================================
# ОБОРУДОВАНИЕ
# ==============================================================================
@pytest.fixture
def category_factory(db):
"""Фабрика для создания категорий оборудования."""
def _create_category(name="Тестовая категория", **kwargs):
defaults = {
"description": f"Описание {name}",
"icon": "bi-box",
}
defaults.update(kwargs)
return EquipmentCategory.objects.create(name=name, **defaults)
return _create_category
@pytest.fixture
def equipment_factory(db):
"""Фабрика для создания оборудования."""
counter = {"value": 0}
def _create_equipment(category, department, **kwargs):
counter["value"] += 1
defaults = {
"name": f"Тестовое оборудование {counter['value']}",
"inventory_number": f"INV-TEST-{counter['value']:04d}",
"status": EquipmentStatus.AVAILABLE,
"purchase_date": date.today(),
"purchase_cost": Decimal("50000.00"),
}
defaults.update(kwargs)
return Equipment.objects.create(
category=category,
department=department,
**defaults
)
return _create_equipment
# ==============================================================================
# ПОСТАВЩИКИ
# ==============================================================================
@pytest.fixture
def supplier_factory(db):
"""Фабрика для создания поставщиков."""
def _create_supplier(name="Тестовый поставщик", **kwargs):
defaults = {
"contact_person": "Контактное лицо",
"phone": "+79991112233",
"email": "supplier@example.com",
"is_active": True,
}
defaults.update(kwargs)
return Supplier.objects.create(name=name, **defaults)
return _create_supplier

480
backend/tests/api/v1/procurement/test_budgets.py Normal file
View File

@ -0,0 +1,480 @@
"""
Тесты API для бюджетов (Budget).
"""
from decimal import Decimal
import pytest
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APIClient
from employees.models import Department, Employee, EmployeeDepartment
from procurement.models import Budget, ProcurementRequest
from procurement.constants import ProcurementStatus
pytestmark = pytest.mark.django_db
@pytest.fixture
def api_client():
"""API клиент."""
return APIClient()
@pytest.fixture
def department(db):
"""Тестовый отдел."""
return Department.objects.create(
name="IT отдел",
description="Отдел IT"
)
@pytest.fixture
def user(db, department):
"""Обычный пользователь."""
user = Employee.objects.create_user(
email="user@example.com",
password="testpass123",
phone_number="+79991111111",
first_name="Иван",
last_name="Иванов",
send_activation_email=False,
)
EmployeeDepartment.objects.create(
employee=user,
department=department,
is_active=True
)
return user
@pytest.fixture
def finance_user(db):
"""Пользователь с правами на управление бюджетами."""
user = Employee.objects.create_user(
email="finance@example.com",
password="testpass123",
phone_number="+79992222222",
first_name="Финансист",
last_name="Финансов",
is_staff=True,
send_activation_email=False,
)
# Даем право на управление бюджетами
from django.contrib.auth.models import Permission
perm = Permission.objects.get(
codename='change_budget',
content_type__app_label='procurement'
)
user.user_permissions.add(perm)
return user
@pytest.fixture
def budget(db, department):
"""Тестовый бюджет."""
return Budget.objects.create(
department=department,
year=2026,
quarter=1,
allocated_amount=Decimal("1000000.00"),
spent_amount=Decimal("250000.00"),
)
# ==============================================================================
# ТЕСТЫ СПИСКА БЮДЖЕТОВ
# ==============================================================================
class TestBudgetList:
"""Тесты получения списка бюджетов."""
def test_list_unauthorized(self, api_client):
"""Неавторизованный доступ запрещен."""
url = reverse('api:v1:procurement:budget-list')
response = api_client.get(url)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_list_budgets_as_staff(
self, api_client, finance_user, budget
):
"""Финансист видит все бюджеты."""
api_client.force_authenticate(user=finance_user)
url = reverse('api:v1:procurement:budget-list')
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
results = response.data['results']
assert len(results) >= 1
assert 'remaining_amount' in results[0]
assert 'utilization_percentage' in results[0]
def test_filter_by_year(
self, api_client, finance_user, department, budget
):
"""Фильтр по году."""
# Создаем бюджет на другой год
Budget.objects.create(
department=department,
year=2025,
quarter=4,
allocated_amount=Decimal("800000.00"),
spent_amount=Decimal("600000.00"),
)
api_client.force_authenticate(user=finance_user)
url = reverse('api:v1:procurement:budget-list')
response = api_client.get(url, {'year': 2026})
assert response.status_code == status.HTTP_200_OK
results = response.data['results']
assert all(r['year'] == 2026 for r in results)
def test_filter_by_department(
self, api_client, finance_user, db, budget
):
"""Фильтр по отделу."""
other_dept = Department.objects.create(
name="HR отдел",
description="Другой отдел"
)
Budget.objects.create(
department=other_dept,
year=2026,
quarter=1,
allocated_amount=Decimal("500000.00"),
spent_amount=Decimal("0.00"),
)
api_client.force_authenticate(user=finance_user)
url = reverse('api:v1:procurement:budget-list')
response = api_client.get(url, {'department': budget.department.id})
assert response.status_code == status.HTTP_200_OK
results = response.data['results']
assert all(r['department'] == budget.department.id for r in results)
# ==============================================================================
# ТЕСТЫ СОЗДАНИЯ БЮДЖЕТА
# ==============================================================================
class TestBudgetCreate:
"""Тесты создания бюджета."""
def test_create_budget_as_finance(
self, api_client, finance_user, department
):
"""Создание бюджета финансистом."""
api_client.force_authenticate(user=finance_user)
url = reverse('api:v1:procurement:budget-list')
data = {
'department': department.id,
'year': 2026,
'quarter': 2,
'allocated_amount': '1200000.00',
}
response = api_client.post(url, data, format='json')
assert response.status_code == status.HTTP_201_CREATED
assert response.data['year'] == 2026
assert response.data['quarter'] == 2
assert Decimal(response.data['allocated_amount']) == Decimal('1200000.00')
assert Decimal(response.data['spent_amount']) == Decimal('0.00')
def test_create_budget_regular_user_forbidden(
self, api_client, user, department
):
"""Обычный пользователь не может создавать бюджеты."""
api_client.force_authenticate(user=user)
url = reverse('api:v1:procurement:budget-list')
data = {
'department': department.id,
'year': 2026,
'quarter': 2,
'allocated_amount': '1000000.00',
}
response = api_client.post(url, data, format='json')
assert response.status_code == status.HTTP_403_FORBIDDEN
def test_create_duplicate_budget_fails(
self, api_client, finance_user, budget
):
"""Нельзя создать дублирующий бюджет (отдел+год+квартал)."""
api_client.force_authenticate(user=finance_user)
url = reverse('api:v1:procurement:budget-list')
data = {
'department': budget.department.id,
'year': budget.year,
'quarter': budget.quarter,
'allocated_amount': '2000000.00',
}
response = api_client.post(url, data, format='json')
assert response.status_code == status.HTTP_400_BAD_REQUEST
# ==============================================================================
# ТЕСТЫ ДЕТАЛЬНОЙ ИНФОРМАЦИИ
# ==============================================================================
class TestBudgetDetail:
"""Тесты получения детальной информации о бюджете."""
def test_retrieve_budget(
self, api_client, finance_user, budget
):
"""Получение детальной информации."""
api_client.force_authenticate(user=finance_user)
url = reverse(
'api:v1:procurement:budget-detail',
kwargs={'pk': budget.id}
)
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
assert response.data['id'] == budget.id
assert Decimal(response.data['allocated_amount']) == budget.allocated_amount
assert Decimal(response.data['spent_amount']) == budget.spent_amount
# Проверяем вычисляемые поля
assert 'remaining_amount' in response.data
assert 'utilization_percentage' in response.data
assert Decimal(response.data['remaining_amount']) == Decimal('750000.00')
def test_retrieve_shows_reserved_amount(
self, api_client, finance_user, budget, user
):
"""Детальная информация показывает зарезервированные средства."""
# Создаем pending заявку
pending_request = ProcurementRequest.objects.create(
title="Заявка",
description="Описание",
department=budget.department,
requestor=user,
status=ProcurementStatus.PENDING,
)
from procurement.models import ProcurementItem
ProcurementItem.objects.create(
request=pending_request,
name="Товар",
quantity=1,
estimated_unit_price=Decimal("50000.00"),
)
api_client.force_authenticate(user=finance_user)
url = reverse(
'api:v1:procurement:budget-detail',
kwargs={'pk': budget.id}
)
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
# Должно быть поле reserved_amount
assert 'reserved_amount' in response.data
# ==============================================================================
# ТЕСТЫ ОБНОВЛЕНИЯ БЮДЖЕТА
# ==============================================================================
class TestBudgetUpdate:
"""Тесты обновления бюджета."""
def test_update_budget_as_finance(
self, api_client, finance_user, budget
):
"""Обновление бюджета финансистом."""
api_client.force_authenticate(user=finance_user)
url = reverse(
'api:v1:procurement:budget-detail',
kwargs={'pk': budget.id}
)
data = {
'allocated_amount': '1500000.00',
'spent_amount': '300000.00',
}
response = api_client.patch(url, data, format='json')
assert response.status_code == status.HTTP_200_OK
assert Decimal(response.data['allocated_amount']) == Decimal('1500000.00')
assert Decimal(response.data['spent_amount']) == Decimal('300000.00')
def test_update_budget_regular_user_forbidden(
self, api_client, user, budget
):
"""Обычный пользователь не может обновлять бюджеты."""
api_client.force_authenticate(user=user)
url = reverse(
'api:v1:procurement:budget-detail',
kwargs={'pk': budget.id}
)
data = {'allocated_amount': '2000000.00'}
response = api_client.patch(url, data, format='json')
assert response.status_code == status.HTTP_403_FORBIDDEN
# ==============================================================================
# ТЕСТЫ СПЕЦИАЛЬНЫХ ACTIONS
# ==============================================================================
class TestBudgetActions:
"""Тесты специальных действий с бюджетами."""
def test_current_budget(
self, api_client, finance_user, budget
):
"""Получение бюджета текущего квартала."""
api_client.force_authenticate(user=finance_user)
url = reverse('api:v1:procurement:budget-current')
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
# Должен вернуть список бюджетов на текущий квартал
assert isinstance(response.data, list) or 'results' in response.data
def test_my_department_budget(
self, api_client, user, budget
):
"""Получение бюджета своего отдела."""
api_client.force_authenticate(user=user)
url = reverse('api:v1:procurement:budget-my-department')
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
# Пользователь видит бюджет своего отдела
if response.data: # Может быть пустой если нет бюджета
assert response.data[0]['department'] == budget.department.id
# ==============================================================================
# ТЕСТЫ БИЗНЕС-ЛОГИКИ
# ==============================================================================
class TestBudgetBusinessLogic:
"""Тесты бизнес-логики бюджетов."""
def test_budget_utilization_calculation(
self, api_client, finance_user, budget
):
"""Проверка расчета процента использования."""
api_client.force_authenticate(user=finance_user)
url = reverse(
'api:v1:procurement:budget-detail',
kwargs={'pk': budget.id}
)
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
# allocated = 1000000, spent = 250000
# utilization = 250000 / 1000000 * 100 = 25%
utilization = Decimal(response.data['utilization_percentage'])
assert utilization == Decimal('25.00')
def test_budget_remaining_amount(
self, api_client, finance_user, budget
):
"""Проверка расчета остатка."""
api_client.force_authenticate(user=finance_user)
url = reverse(
'api:v1:procurement:budget-detail',
kwargs={'pk': budget.id}
)
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
# allocated = 1000000, spent = 250000
# remaining = 750000
remaining = Decimal(response.data['remaining_amount'])
assert remaining == Decimal('750000.00')
def test_negative_budget_not_allowed(
self, api_client, finance_user, budget
):
"""Нельзя установить отрицательный бюджет."""
api_client.force_authenticate(user=finance_user)
url = reverse(
'api:v1:procurement:budget-detail',
kwargs={'pk': budget.id}
)
data = {'allocated_amount': '-100000.00'}
response = api_client.patch(url, data, format='json')
assert response.status_code == status.HTTP_400_BAD_REQUEST
def test_spent_more_than_allocated_warning(
self, api_client, finance_user, budget
):
"""Можно потратить больше выделенного (с предупреждением)."""
# Обновляем потраченную сумму больше выделенной
budget.spent_amount = Decimal("1200000.00")
budget.save()
api_client.force_authenticate(user=finance_user)
url = reverse(
'api:v1:procurement:budget-detail',
kwargs={'pk': budget.id}
)
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
# remaining должен быть отрицательным
remaining = Decimal(response.data['remaining_amount'])
assert remaining < 0
# utilization > 100%
utilization = Decimal(response.data['utilization_percentage'])
assert utilization > 100
# ==============================================================================
# ТЕСТЫ УДАЛЕНИЯ
# ==============================================================================
class TestBudgetDelete:
"""Тесты удаления бюджета."""
def test_delete_budget_as_finance(
self, api_client, finance_user, budget
):
"""Удаление бюджета финансистом."""
api_client.force_authenticate(user=finance_user)
url = reverse(
'api:v1:procurement:budget-detail',
kwargs={'pk': budget.id}
)
response = api_client.delete(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert not Budget.objects.filter(id=budget.id).exists()
def test_delete_budget_regular_user_forbidden(
self, api_client, user, budget
):
"""Обычный пользователь не может удалять бюджеты."""
api_client.force_authenticate(user=user)
url = reverse(
'api:v1:procurement:budget-detail',
kwargs={'pk': budget.id}
)
response = api_client.delete(url)
assert response.status_code == status.HTTP_403_FORBIDDEN
assert Budget.objects.filter(id=budget.id).exists()

473
backend/tests/api/v1/procurement/test_equipment.py Normal file
View File

@ -0,0 +1,473 @@
"""
Тесты API для оборудования (Equipment).
"""
from decimal import Decimal
from datetime import date, timedelta
import pytest
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APIClient
from employees.models import Department, Employee, EmployeeDepartment
from procurement.constants import EquipmentStatus
from procurement.models import Equipment, EquipmentCategory, MaintenanceRecord
from procurement.constants import MaintenanceType
pytestmark = pytest.mark.django_db
@pytest.fixture
def api_client():
"""API клиент"""
return APIClient()
@pytest.fixture
def department(db):
"""Тестовый отдел."""
return Department.objects.create(
name="IT отдел",
description="Отдел IT"
)
@pytest.fixture
def user(db, department):
"""Обычный пользователь."""
user = Employee.objects.create_user(
email="user@example.com",
password="testpass123",
phone_number="+79991111111",
first_name="Иван",
last_name="Иванов",
send_activation_email=False,
)
EmployeeDepartment.objects.create(
employee=user,
department=department,
is_active=True
)
return user
@pytest.fixture
def staff_user(db):
"""Staff пользователь."""
return Employee.objects.create_user(
email="staff@example.com",
password="testpass123",
phone_number="+79992222222",
first_name="Админ",
last_name="Админов",
is_staff=True,
send_activation_email=False,
)
@pytest.fixture
def category(db):
"""Категория оборудования."""
return EquipmentCategory.objects.create(
name="Ноутбуки",
description="Портативные компьютеры",
icon="bi-laptop"
)
@pytest.fixture
def equipment(db, category, department, user):
"""Тестовое оборудование."""
return Equipment.objects.create(
name="Ноутбук Dell Latitude 5530",
inventory_number="INV-2026-001",
serial_number="SN123456789",
category=category,
status=EquipmentStatus.AVAILABLE,
department=department,
responsible_person=user,
location="Офис 3.14",
purchase_date=date(2026, 1, 15),
warranty_until=date(2029, 1, 15),
purchase_cost=Decimal("95000.00"),
notes="Для разработки",
)
# ==============================================================================
# ТЕСТЫ СПИСКА ОБОРУДОВАНИЯ
# ==============================================================================
class TestEquipmentList:
"""Тесты получения списка оборудования."""
def test_list_unauthorized(self, api_client):
"""Неавторизованный доступ запрещен."""
url = reverse('api:v1:procurement:equipment-list')
response = api_client.get(url)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_list_equipment(
self, api_client, user, equipment
):
"""Получение списка оборудования."""
api_client.force_authenticate(user=user)
url = reverse('api:v1:procurement:equipment-list')
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
results = response.data['results']
assert len(results) >= 1
# Проверяем структуру
item = results[0]
assert 'inventory_number' in item
assert 'name' in item
assert 'status' in item
assert 'category_name' in item
assert 'department_name' in item
def test_filter_by_status(
self, api_client, user, category, department, equipment
):
"""Фильтр по статусу."""
# Создаем оборудование с разными статусами
Equipment.objects.create(
name="Монитор",
inventory_number="INV-2026-002",
category=category,
status=EquipmentStatus.IN_USE,
department=department,
purchase_date=date.today(),
purchase_cost=Decimal("20000.00"),
)
api_client.force_authenticate(user=user)
url = reverse('api:v1:procurement:equipment-list')
# Фильтр AVAILABLE
response = api_client.get(url, {'status': EquipmentStatus.AVAILABLE})
assert response.status_code == status.HTTP_200_OK
results = response.data['results']
assert all(r['status'] == EquipmentStatus.AVAILABLE for r in results)
def test_filter_by_department(
self, api_client, user, db, category, department, equipment
):
"""Фильтр по отделу."""
other_dept = Department.objects.create(
name="HR отдел",
description="Другой отдел"
)
Equipment.objects.create(
name="Принтер",
inventory_number="INV-2026-003",
category=category,
status=EquipmentStatus.AVAILABLE,
department=other_dept,
purchase_date=date.today(),
purchase_cost=Decimal("15000.00"),
)
api_client.force_authenticate(user=user)
url = reverse('api:v1:procurement:equipment-list')
response = api_client.get(url, {'department': department.id})
assert response.status_code == status.HTTP_200_OK
results = response.data['results']
# Все должны быть из нужного отдела
assert all(r['department'] == department.id for r in results)
def test_search_equipment(
self, api_client, user, equipment
):
"""Поиск оборудования."""
api_client.force_authenticate(user=user)
url = reverse('api:v1:procurement:equipment-list')
response = api_client.get(url, {'search': 'latitude'})
assert response.status_code == status.HTTP_200_OK
assert len(response.data['results']) >= 1
# ==============================================================================
# ТЕСТЫ СОЗДАНИЯ ОБОРУДОВАНИЯ
# ==============================================================================
class TestEquipmentCreate:
"""Тесты создания оборудования."""
def test_create_equipment_as_staff(
self, api_client, staff_user, category, department
):
"""Создание оборудования staff пользователем."""
api_client.force_authenticate(user=staff_user)
url = reverse('api:v1:procurement:equipment-list')
data = {
'name': 'Монитор LG 27"',
'inventory_number': 'INV-2026-100',
'serial_number': 'LG987654321',
'category': category.id,
'status': EquipmentStatus.AVAILABLE,
'department': department.id,
'location': 'Склад',
'purchase_date': '2026-02-01',
'warranty_until': '2029-02-01',
'purchase_cost': '25000.00',
'notes': 'Для офиса',
}
response = api_client.post(url, data, format='json')
assert response.status_code == status.HTTP_201_CREATED
assert response.data['name'] == 'Монитор LG 27"'
assert response.data['inventory_number'] == 'INV-2026-100'
def test_create_equipment_regular_user_forbidden(
self, api_client, user, category, department
):
"""Обычный пользователь не может создавать оборудование."""
api_client.force_authenticate(user=user)
url = reverse('api:v1:procurement:equipment-list')
data = {
'name': 'Тестовое оборудование',
'inventory_number': 'INV-2026-999',
'category': category.id,
'department': department.id,
'purchase_date': '2026-02-01',
'purchase_cost': '10000.00',
}
response = api_client.post(url, data, format='json')
assert response.status_code == status.HTTP_403_FORBIDDEN
def test_create_equipment_duplicate_inventory_number(
self, api_client, staff_user, category, department, equipment
):
"""Нельзя создать с дублирующим инвентарным номером."""
api_client.force_authenticate(user=staff_user)
url = reverse('api:v1:procurement:equipment-list')
data = {
'name': 'Другое оборудование',
'inventory_number': equipment.inventory_number, # Дубликат!
'category': category.id,
'department': department.id,
'purchase_date': '2026-02-01',
'purchase_cost': '10000.00',
}
response = api_client.post(url, data, format='json')
assert response.status_code == status.HTTP_400_BAD_REQUEST
# ==============================================================================
# ТЕСТЫ ДЕТАЛЬНОЙ ИНФОРМАЦИИ
# ==============================================================================
class TestEquipmentDetail:
"""Тесты получения детальной информации об оборудовании."""
def test_retrieve_equipment(
self, api_client, user, equipment
):
"""Получение детальной информации."""
api_client.force_authenticate(user=user)
url = reverse(
'api:v1:procurement:equipment-detail',
kwargs={'pk': equipment.id}
)
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
assert response.data['id'] == equipment.id
assert response.data['name'] == equipment.name
assert response.data['inventory_number'] == equipment.inventory_number
assert 'is_under_warranty' in response.data
assert response.data['is_under_warranty'] is True
def test_retrieve_with_maintenance_history(
self, api_client, user, equipment
):
"""Детальная информация включает историю ТО."""
# Создаем записи об обслуживании
MaintenanceRecord.objects.create(
equipment=equipment,
date=date.today() - timedelta(days=30),
type=MaintenanceType.PLANNED,
description="Плановое ТО",
cost=Decimal("5000.00"),
performed_by=user,
)
api_client.force_authenticate(user=user)
url = reverse(
'api:v1:procurement:equipment-detail',
kwargs={'pk': equipment.id}
)
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
assert 'maintenance_history' in response.data
assert len(response.data['maintenance_history']) >= 1
# ==============================================================================
# ТЕСТЫ ОБНОВЛЕНИЯ ОБОРУДОВАНИЯ
# ==============================================================================
class TestEquipmentUpdate:
"""Тесты обновления оборудования."""
def test_update_equipment_as_staff(
self, api_client, staff_user, equipment
):
"""Обновление оборудования staff пользователем."""
api_client.force_authenticate(user=staff_user)
url = reverse(
'api:v1:procurement:equipment-detail',
kwargs={'pk': equipment.id}
)
data = {
'location': 'Офис 4.01',
'notes': 'Перемещено в новый офис',
}
response = api_client.patch(url, data, format='json')
assert response.status_code == status.HTTP_200_OK
assert response.data['location'] == 'Офис 4.01'
assert 'новый офис' in response.data['notes'].lower()
def test_update_equipment_regular_user_forbidden(
self, api_client, user, equipment
):
"""Обычный пользователь не может обновлять."""
api_client.force_authenticate(user=user)
url = reverse(
'api:v1:procurement:equipment-detail',
kwargs={'pk': equipment.id}
)
data = {'location': 'Новое место'}
response = api_client.patch(url, data, format='json')
assert response.status_code == status.HTTP_403_FORBIDDEN
# ==============================================================================
# ТЕСТЫ СПЕЦИАЛЬНЫХ ДЕЙСТВИЙ
# ==============================================================================
class TestEquipmentActions:
"""Тесты специальных действий с оборудованием."""
def test_transfer_equipment(
self, api_client, staff_user, equipment, db
):
"""Передача оборудования."""
new_user = Employee.objects.create_user(
email="newuser@example.com",
password="testpass123",
phone_number="+79995555555",
first_name="Новый",
last_name="Пользователь",
send_activation_email=False,
)
api_client.force_authenticate(user=staff_user)
url = reverse(
'api:v1:procurement:equipment-transfer',
kwargs={'pk': equipment.id}
)
data = {
'to_person': new_user.id,
'to_location': 'Офис 2.10',
'reason': 'Перевод в другой отдел',
}
response = api_client.post(url, data, format='json')
assert response.status_code == status.HTTP_200_OK
# Проверяем что ответственный изменился
equipment.refresh_from_db()
assert equipment.responsible_person == new_user
assert equipment.location == 'Офис 2.10'
def test_generate_qr_code(
self, api_client, user, equipment
):
"""Генерация QR-кода."""
api_client.force_authenticate(user=user)
url = reverse(
'api:v1:procurement:equipment-qr-code',
kwargs={'pk': equipment.id}
)
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
assert 'qr_code' in response.data
# QR код должен быть в base64
assert response.data['qr_code'].startswith('data:image/')
# ==============================================================================
# ТЕСТЫ КАТЕГОРИЙ
# ==============================================================================
class TestEquipmentCategories:
"""Тесты API категорий оборудования."""
def test_list_categories(
self, api_client, user, category
):
"""Список категорий."""
api_client.force_authenticate(user=user)
url = reverse('api:v1:procurement:equipmentcategory-list')
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
assert len(response.data) >= 1
def test_create_category_as_staff(
self, api_client, staff_user
):
"""Создание категории staff пользователем."""
api_client.force_authenticate(user=staff_user)
url = reverse('api:v1:procurement:equipmentcategory-list')
data = {
'name': 'Мониторы',
'description': 'Все виды мониторов',
'icon': 'bi-display',
}
response = api_client.post(url, data, format='json')
assert response.status_code == status.HTTP_201_CREATED
assert response.data['name'] == 'Мониторы'
def test_hierarchical_categories(
self, api_client, staff_user, category
):
"""Создание подкатегории."""
api_client.force_authenticate(user=staff_user)
url = reverse('api:v1:procurement:equipmentcategory-list')
data = {
'name': 'Игровые ноутбуки',
'parent': category.id,
'description': 'Высокопроизводительные ноутбуки',
}
response = api_client.post(url, data, format='json')
assert response.status_code == status.HTTP_201_CREATED
assert response.data['parent'] == category.id

532
backend/tests/api/v1/procurement/test_procurement_requests.py Normal file
View File

@ -0,0 +1,532 @@
"""
Тесты API для заявок на закупку (ProcurementRequest).
"""
from decimal import Decimal
import pytest
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APIClient
from employees.models import Department, Employee, EmployeeDepartment
from procurement.constants import (
ApprovalRole,
ApprovalStatus,
ProcurementStatus,
UrgencyLevel,
)
from procurement.models import (
Approval,
Budget,
ProcurementItem,
ProcurementRequest,
)
pytestmark = pytest.mark.django_db
@pytest.fixture
def api_client():
"""API клиент."""
return APIClient()
@pytest.fixture
def department(db):
"""Тестовый отдел."""
return Department.objects.create(
name="IT отдел",
description="Отдел информационных технологий"
)
@pytest.fixture
def user(db, department):
"""Обычный пользователь."""
user = Employee.objects.create_user(
email="user@example.com",
password="testpass123",
phone_number="+79991111111",
first_name="Иван",
last_name="Иванов",
send_activation_email=False,
)
# Связываем с отделом
EmployeeDepartment.objects.create(
employee=user,
department=department,
is_active=True
)
return user
@pytest.fixture
def department_head(db, department):
"""Руководитель отдела."""
head = Employee.objects.create_user(
email="head@example.com",
password="testpass123",
phone_number="+79992222222",
first_name="Петр",
last_name="Петров",
send_activation_email=False,
)
# Назначаем руководителем отдела (автоматически создаст EmployeeDepartment)
department.head = head
department.save()
return head
@pytest.fixture
def staff_user(db):
"""Staff пользователь."""
return Employee.objects.create_user(
email="staff@example.com",
password="testpass123",
phone_number="+79993333333",
first_name="Админ",
last_name="Админов",
is_staff=True,
send_activation_email=False,
)
@pytest.fixture
def budget(db, department):
"""Бюджет отдела."""
return Budget.objects.create(
department=department,
year=2026,
quarter=1,
allocated_amount=Decimal("500000.00"),
spent_amount=Decimal("0.00"),
)
@pytest.fixture
def procurement_request(db, department, user):
"""Тестовая заявка."""
return ProcurementRequest.objects.create(
title="Закупка ноутбуков",
description="Нужны новые ноутбуки для разработчиков",
department=department,
requestor=user,
status=ProcurementStatus.DRAFT,
urgency=UrgencyLevel.MEDIUM,
)
@pytest.fixture
def procurement_item(db, procurement_request):
"""Позиция в заявке."""
return ProcurementItem.objects.create(
request=procurement_request,
name="Ноутбук Dell XPS 15",
description="i7, 32GB RAM, 1TB SSD",
quantity=3,
unit="шт",
estimated_unit_price=Decimal("120000.00"),
supplier_info="https://example.com/dell-xps-15",
)
# ==============================================================================
# ТЕСТЫ СПИСКА ЗАЯВОК
# ==============================================================================
class TestProcurementRequestList:
"""Тесты получения списка заявок."""
def test_list_unauthorized(self, api_client):
"""Неавторизованный доступ запрещен."""
url = reverse('api:v1:procurement:procurementrequest-list')
response = api_client.get(url)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_list_own_requests(
self, api_client, user, procurement_request
):
"""Пользователь видит свои заявки."""
api_client.force_authenticate(user=user)
url = reverse('api:v1:procurement:procurementrequest-list')
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
results = response.data['results']
assert len(results) == 1
assert results[0]['id'] == procurement_request.id
assert results[0]['title'] == "Закупка ноутбуков"
def test_list_department_requests(
self, api_client, department_head, procurement_request
):
"""Руководитель видит заявки своего отдела."""
api_client.force_authenticate(user=department_head)
url = reverse('api:v1:procurement:procurementrequest-list')
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
# Руководитель может видеть заявки отдела через scope=department
def test_filter_by_status(
self, api_client, user, department, procurement_request
):
"""Фильтрация по статусу."""
# Создаем заявку с другим статусом
ProcurementRequest.objects.create(
title="Другая заявка",
description="Описание",
department=department,
requestor=user,
status=ProcurementStatus.PENDING,
)
api_client.force_authenticate(user=user)
url = reverse('api:v1:procurement:procurementrequest-list')
# Фильтр DRAFT
response = api_client.get(url, {'status': ProcurementStatus.DRAFT})
assert response.status_code == status.HTTP_200_OK
assert len(response.data['results']) == 1
assert response.data['results'][0]['status'] == ProcurementStatus.DRAFT
def test_search_by_title(
self, api_client, user, procurement_request
):
"""Поиск по названию."""
api_client.force_authenticate(user=user)
url = reverse('api:v1:procurement:procurementrequest-list')
response = api_client.get(url, {'search': 'ноутбук'})
assert response.status_code == status.HTTP_200_OK
assert len(response.data['results']) == 1
# ==============================================================================
# ТЕСТЫ СОЗДАНИЯ ЗАЯВОК
# ==============================================================================
class TestProcurementRequestCreate:
"""Тесты создания заявок."""
def test_create_request_with_items(
self, api_client, user, department
):
"""Создание заявки с позициями."""
api_client.force_authenticate(user=user)
url = reverse('api:v1:procurement:procurementrequest-list')
data = {
'title': 'Закупка мониторов',
'description': 'Нужны мониторы для офиса',
'department': department.id,
'urgency': UrgencyLevel.LOW,
'items': [
{
'name': 'Монитор 27"',
'quantity': 5,
'unit': 'шт',
'estimated_unit_price': '25000.00',
'description': '4K, IPS',
},
{
'name': 'Монитор 24"',
'quantity': 10,
'unit': 'шт',
'estimated_unit_price': '18000.00',
},
]
}
response = api_client.post(url, data, format='json')
assert response.status_code == status.HTTP_201_CREATED
assert response.data['title'] == 'Закупка мониторов'
assert response.data['requestor'] == user.id
assert response.data['status'] == ProcurementStatus.DRAFT
# Проверяем что позиции созданы
request_id = response.data['id']
request = ProcurementRequest.objects.get(id=request_id)
assert request.items.count() == 2
assert request.total_cost == Decimal('305000.00') # 5*25k + 10*18k
def test_create_request_wrong_department(
self, api_client, user, db
):
"""Нельзя создать заявку для чужого отдела."""
other_dept = Department.objects.create(
name="HR отдел", description="Другой отдел"
)
api_client.force_authenticate(user=user)
url = reverse('api:v1:procurement:procurementrequest-list')
data = {
'title': 'Заявка',
'description': 'Описание',
'department': other_dept.id,
'urgency': UrgencyLevel.MEDIUM,
'items': []
}
response = api_client.post(url, data, format='json')
# Должна быть ошибка валидации
assert response.status_code in [
status.HTTP_400_BAD_REQUEST,
status.HTTP_403_FORBIDDEN
]
# ==============================================================================
# ТЕСТЫ ДЕТАЛЬНОЙ ИНФОРМАЦИИ
# ==============================================================================
class TestProcurementRequestDetail:
"""Тесты получения детальной информации."""
def test_retrieve_own_request(
self, api_client, user, procurement_request, procurement_item
):
"""Получение своей заявки."""
api_client.force_authenticate(user=user)
url = reverse(
'api:v1:procurement:procurementrequest-detail',
kwargs={'pk': procurement_request.id}
)
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
assert response.data['id'] == procurement_request.id
assert response.data['title'] == "Закупка ноутбуков"
assert 'items' in response.data
assert len(response.data['items']) == 1
assert 'total_cost' in response.data
assert 'is_editable' in response.data
assert response.data['is_editable'] is True
def test_retrieve_other_request_forbidden(
self, api_client, db, procurement_request
):
"""Нельзя просмотреть чужую заявку (обычному пользователю)."""
other_user = Employee.objects.create_user(
email="other@example.com",
password="testpass123",
phone_number="+79994444444",
first_name="Другой",
last_name="Пользователь",
send_activation_email=False,
)
api_client.force_authenticate(user=other_user)
url = reverse(
'api:v1:procurement:procurementrequest-detail',
kwargs={'pk': procurement_request.id}
)
response = api_client.get(url)
assert response.status_code == status.HTTP_404_NOT_FOUND
# ==============================================================================
# ТЕСТЫ ОБНОВЛЕНИЯ ЗАЯВОК
# ==============================================================================
class TestProcurementRequestUpdate:
"""Тесты обновления заявок."""
def test_update_draft_request(
self, api_client, user, procurement_request
):
"""Обновление заявки в статусе DRAFT."""
api_client.force_authenticate(user=user)
url = reverse(
'api:v1:procurement:procurementrequest-detail',
kwargs={'pk': procurement_request.id}
)
data = {
'title': 'Обновленное название',
'urgency': UrgencyLevel.HIGH,
}
response = api_client.patch(url, data, format='json')
assert response.status_code == status.HTTP_200_OK
assert response.data['title'] == 'Обновленное название'
assert response.data['urgency'] == UrgencyLevel.HIGH
def test_cannot_update_submitted_request(
self, api_client, user, procurement_request
):
"""Нельзя обновить отправленную заявку."""
procurement_request.status = ProcurementStatus.PENDING
procurement_request.save()
api_client.force_authenticate(user=user)
url = reverse(
'api:v1:procurement:procurementrequest-detail',
kwargs={'pk': procurement_request.id}
)
data = {'title': 'Новое название'}
response = api_client.patch(url, data, format='json')
assert response.status_code == status.HTTP_403_FORBIDDEN
# ==============================================================================
# ТЕСТЫ WORKFLOW ДЕЙСТВИЙ
# ==============================================================================
class TestProcurementRequestWorkflow:
"""Тесты workflow действий с заявками."""
def test_submit_request(
self, api_client, user, procurement_request, procurement_item, budget
):
"""Отправка заявки на согласование."""
api_client.force_authenticate(user=user)
url = reverse(
'api:v1:procurement:procurementrequest-submit',
kwargs={'pk': procurement_request.id}
)
response = api_client.post(url)
assert response.status_code == status.HTTP_200_OK
# Проверяем статус изменен
procurement_request.refresh_from_db()
assert procurement_request.status == ProcurementStatus.PENDING
assert procurement_request.submitted_at is not None
def test_submit_without_items_fails(
self, api_client, user, procurement_request
):
"""Нельзя отправить заявку без позиций."""
api_client.force_authenticate(user=user)
url = reverse(
'api:v1:procurement:procurementrequest-submit',
kwargs={'pk': procurement_request.id}
)
response = api_client.post(url)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert 'позицию' in response.data['error'].lower()
def test_approve_request_by_head(
self, api_client, department_head, procurement_request,
procurement_item, budget
):
"""Согласование заявки руководителем."""
# Отправляем заявку
procurement_request.status = ProcurementStatus.PENDING
procurement_request.save()
# Создаем запись согласования
Approval.objects.create(
request=procurement_request,
approver=department_head,
role=ApprovalRole.DEPARTMENT_HEAD,
status=ApprovalStatus.PENDING,
)
api_client.force_authenticate(user=department_head)
url = reverse(
'api:v1:procurement:procurementrequest-approve',
kwargs={'pk': procurement_request.id}
)
response = api_client.post(url, {'comment': 'Одобрено'})
assert response.status_code == status.HTTP_200_OK
def test_reject_request(
self, api_client, department_head, procurement_request,
procurement_item
):
"""Отклонение заявки."""
procurement_request.status = ProcurementStatus.PENDING
procurement_request.save()
Approval.objects.create(
request=procurement_request,
approver=department_head,
role=ApprovalRole.DEPARTMENT_HEAD,
status=ApprovalStatus.PENDING,
)
api_client.force_authenticate(user=department_head)
url = reverse(
'api:v1:procurement:procurementrequest-reject',
kwargs={'pk': procurement_request.id}
)
data = {'comment': 'Недостаточно обоснования'}
response = api_client.post(url, data)
assert response.status_code == status.HTTP_200_OK
procurement_request.refresh_from_db()
assert procurement_request.status == ProcurementStatus.REJECTED
def test_cancel_own_request(
self, api_client, user, procurement_request
):
"""Отмена своей заявки."""
api_client.force_authenticate(user=user)
url = reverse(
'api:v1:procurement:procurementrequest-cancel',
kwargs={'pk': procurement_request.id}
)
response = api_client.post(url)
assert response.status_code == status.HTTP_200_OK
procurement_request.refresh_from_db()
assert procurement_request.status == ProcurementStatus.CANCELLED
# ==============================================================================
# ТЕСТЫ УДАЛЕНИЯ
# ==============================================================================
class TestProcurementRequestDelete:
"""Тесты удаления заявок."""
def test_delete_draft_request(
self, api_client, user, procurement_request
):
"""Удаление черновика."""
api_client.force_authenticate(user=user)
url = reverse(
'api:v1:procurement:procurementrequest-detail',
kwargs={'pk': procurement_request.id}
)
response = api_client.delete(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert not ProcurementRequest.objects.filter(
id=procurement_request.id
).exists()
def test_cannot_delete_submitted_request(
self, api_client, user, procurement_request
):
"""Нельзя удалить отправленную заявку."""
procurement_request.status = ProcurementStatus.PENDING
procurement_request.save()
api_client.force_authenticate(user=user)
url = reverse(
'api:v1:procurement:procurementrequest-detail',
kwargs={'pk': procurement_request.id}
)
response = api_client.delete(url)
assert response.status_code == status.HTTP_403_FORBIDDEN

365
backend/tests/api/v1/procurement/test_suppliers.py Normal file
View File

@ -0,0 +1,365 @@
"""
Тесты API для поставщиков (Supplier).
"""
from decimal import Decimal
import pytest
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APIClient
from employees.models import Employee
from procurement.models import Supplier
pytestmark = pytest.mark.django_db
@pytest.fixture
def api_client():
"""API клиент."""
return APIClient()
@pytest.fixture
def user(db):
"""Обычный пользователь."""
return Employee.objects.create_user(
email="user@example.com",
password="testpass123",
phone_number="+79991111111",
first_name="Иван",
last_name="Иванов",
send_activation_email=False,
)
@pytest.fixture
def staff_user(db):
"""Staff пользователь."""
return Employee.objects.create_user(
email="staff@example.com",
password="testpass123",
phone_number="+79992222222",
first_name="Админ",
last_name="Админов",
is_staff=True,
send_activation_email=False,
)
@pytest.fixture
def supplier(db):
"""Тестовый поставщик."""
return Supplier.objects.create(
name="ООО \"Компьютеры\"",
contact_person="Иванов Иван",
phone="+79991234567",
email="info@computers.ru",
website="https://computers.ru",
inn="7701234567",
address="Москва, ул. Ленина, д. 1",
rating=Decimal("4.50"),
is_active=True,
notes="Основной поставщик IT-оборудования",
)
# ==============================================================================
# ТЕСТЫ СПИСКА ПОСТАВЩИКОВ
# ==============================================================================
class TestSupplierList:
"""Тесты получения списка поставщиков."""
def test_list_unauthorized(self, api_client):
"""Неавторизованный доступ запрещен."""
url = reverse('api:v1:procurement:supplier-list')
response = api_client.get(url)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_list_suppliers(
self, api_client, user, supplier
):
"""Получение списка поставщиков."""
api_client.force_authenticate(user=user)
url = reverse('api:v1:procurement:supplier-list')
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
results = response.data['results']
assert len(results) >= 1
# Проверяем структуру
item = results[0]
assert 'name' in item
assert 'contact_person' in item
assert 'phone' in item
assert 'email' in item
assert 'rating' in item
assert 'is_active' in item
def test_search_suppliers(
self, api_client, user, supplier
):
"""Поиск поставщиков."""
api_client.force_authenticate(user=user)
url = reverse('api:v1:procurement:supplier-list')
response = api_client.get(url, {'search': 'компьютер'})
assert response.status_code == status.HTTP_200_OK
assert len(response.data['results']) >= 1
def test_filter_active_suppliers(
self, api_client, user, db
):
"""Фильтр только активных поставщиков."""
# Создаем активного и неактивного
Supplier.objects.create(
name="Активный поставщик",
is_active=True,
)
Supplier.objects.create(
name="Неактивный поставщик",
is_active=False,
)
api_client.force_authenticate(user=user)
url = reverse('api:v1:procurement:supplier-list')
response = api_client.get(url, {'is_active': 'true'})
assert response.status_code == status.HTTP_200_OK
results = response.data['results']
assert all(r['is_active'] for r in results)
# ==============================================================================
# ТЕСТЫ СОЗДАНИЯ ПОСТАВЩИКА
# ==============================================================================
class TestSupplierCreate:
"""Тесты создания поставщика."""
def test_create_supplier_as_staff(
self, api_client, staff_user
):
"""Создание поставщика staff пользователем."""
api_client.force_authenticate(user=staff_user)
url = reverse('api:v1:procurement:supplier-list')
data = {
'name': 'ООО "Новый поставщик"',
'contact_person': 'Петров Петр',
'phone': '+79997654321',
'email': 'new@supplier.ru',
'website': 'https://newsupplier.ru',
'inn': '7707654321',
'address': 'Санкт-Петербург, Невский пр., д. 10',
'rating': '5.00',
'notes': 'Проверенный поставщик',
}
response = api_client.post(url, data, format='json')
assert response.status_code == status.HTTP_201_CREATED
assert response.data['name'] == 'ООО "Новый поставщик"'
assert response.data['inn'] == '7707654321'
def test_create_supplier_regular_user_forbidden(
self, api_client, user
):
"""Обычный пользователь не может создавать поставщиков."""
api_client.force_authenticate(user=user)
url = reverse('api:v1:procurement:supplier-list')
data = {
'name': 'Тестовый поставщик',
'email': 'test@supplier.ru',
}
response = api_client.post(url, data, format='json')
assert response.status_code == status.HTTP_403_FORBIDDEN
def test_create_supplier_minimum_data(
self, api_client, staff_user
):
"""Создание поставщика с минимальными данными."""
api_client.force_authenticate(user=staff_user)
url = reverse('api:v1:procurement:supplier-list')
data = {
'name': 'Минимальный поставщик',
}
response = api_client.post(url, data, format='json')
assert response.status_code == status.HTTP_201_CREATED
assert response.data['name'] == 'Минимальный поставщик'
assert response.data['is_active'] is True # По умолчанию
# ==============================================================================
# ТЕСТЫ ДЕТАЛЬНОЙ ИНФОРМАЦИИ
# ==============================================================================
class TestSupplierDetail:
"""Тесты получения детальной информации о поставщике."""
def test_retrieve_supplier(
self, api_client, user, supplier
):
"""Получение детальной информации."""
api_client.force_authenticate(user=user)
url = reverse(
'api:v1:procurement:supplier-detail',
kwargs={'pk': supplier.id}
)
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
assert response.data['id'] == supplier.id
assert response.data['name'] == supplier.name
assert response.data['inn'] == supplier.inn
assert Decimal(response.data['rating']) == supplier.rating
# ==============================================================================
# ТЕСТЫ ОБНОВЛЕНИЯ ПОСТАВЩИКА
# ==============================================================================
class TestSupplierUpdate:
"""Тесты обновления поставщика."""
def test_update_supplier_as_staff(
self, api_client, staff_user, supplier
):
"""Обновление поставщика staff пользователем."""
api_client.force_authenticate(user=staff_user)
url = reverse(
'api:v1:procurement:supplier-detail',
kwargs={'pk': supplier.id}
)
data = {
'rating': '4.80',
'notes': 'Обновленная информация о поставщике',
}
response = api_client.patch(url, data, format='json')
assert response.status_code == status.HTTP_200_OK
assert Decimal(response.data['rating']) == Decimal('4.80')
assert 'Обновленная' in response.data['notes']
def test_update_supplier_regular_user_forbidden(
self, api_client, user, supplier
):
"""Обычный пользователь не может обновлять поставщиков."""
api_client.force_authenticate(user=user)
url = reverse(
'api:v1:procurement:supplier-detail',
kwargs={'pk': supplier.id}
)
data = {'rating': '5.00'}
response = api_client.patch(url, data, format='json')
assert response.status_code == status.HTTP_403_FORBIDDEN
def test_deactivate_supplier(
self, api_client, staff_user, supplier
):
"""Деактивация поставщика."""
api_client.force_authenticate(user=staff_user)
url = reverse(
'api:v1:procurement:supplier-detail',
kwargs={'pk': supplier.id}
)
data = {'is_active': False}
response = api_client.patch(url, data, format='json')
assert response.status_code == status.HTTP_200_OK
assert response.data['is_active'] is False
# ==============================================================================
# ТЕСТЫ УДАЛЕНИЯ
# ==============================================================================
class TestSupplierDelete:
"""Тесты удаления поставщика."""
def test_delete_supplier_as_staff(
self, api_client, staff_user, supplier
):
"""Удаление поставщика staff пользователем."""
api_client.force_authenticate(user=staff_user)
url = reverse(
'api:v1:procurement:supplier-detail',
kwargs={'pk': supplier.id}
)
response = api_client.delete(url)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert not Supplier.objects.filter(id=supplier.id).exists()
def test_delete_supplier_regular_user_forbidden(
self, api_client, user, supplier
):
"""Обычный пользователь не может удалять поставщиков."""
api_client.force_authenticate(user=user)
url = reverse(
'api:v1:procurement:supplier-detail',
kwargs={'pk': supplier.id}
)
response = api_client.delete(url)
assert response.status_code == status.HTTP_403_FORBIDDEN
assert Supplier.objects.filter(id=supplier.id).exists()
# ==============================================================================
# ТЕСТЫ БИЗНЕС-ЛОГИКИ
# ==============================================================================
class TestSupplierBusinessLogic:
"""Тесты бизнес-логики поставщиков."""
def test_rating_validation(
self, api_client, staff_user
):
"""Валидация рейтинга (не может быть отрицательным)."""
api_client.force_authenticate(user=staff_user)
url = reverse('api:v1:procurement:supplier-list')
data = {
'name': 'Поставщик с плохим рейтингом',
'rating': '-1.00',
}
response = api_client.post(url, data, format='json')
assert response.status_code == status.HTTP_400_BAD_REQUEST
def test_inn_format(
self, api_client, staff_user, supplier
):
"""ИНН должен быть корректным (10 или 12 цифр)."""
api_client.force_authenticate(user=staff_user)
url = reverse(
'api:v1:procurement:supplier-detail',
kwargs={'pk': supplier.id}
)
# ИНН из 10 цифр - валиден
data = {'inn': '1234567890'}
response = api_client.patch(url, data, format='json')
assert response.status_code == status.HTTP_200_OK
# ИНН из 12 цифр - валиден
data = {'inn': '123456789012'}
response = api_client.patch(url, data, format='json')
assert response.status_code == status.HTTP_200_OK

890
docs/in-progress/PROCUREMENT_REFACTORING_PLAN.md Normal file
View File

@ -0,0 +1,890 @@
# План рефакторинга модуля Procurement для pip-публикации
**Цель:** Сделать модуль `procurement` независимым и готовым к публикации на PyPI.
**Статус:** 🚧 В разработке
**Дата начала:** 27 февраля 2026
**Оценка времени:** 2-3 недели
**Риск:** ⚠️ Средний (модуль не в проде, но есть тесты)
---
## 📊 Анализ текущего состояния
### Проблемы tight coupling:
**19 жестких связей:**
```python
# ❌ Текущее состояние
from employees.models import Department # Жесткая зависимость
User = get_user_model() # Предполагает employees.Employee
ProcurementRequest:
- department: ForeignKey(Department) # 1
- requestor: ForeignKey(User) # 2
- executor: ForeignKey(User) # 3
ProcurementItem:
- request: ForeignKey(ProcurementRequest) # OK (внутренняя)
Approval:
- request: ForeignKey(ProcurementRequest) # OK
- approver: ForeignKey(User) # 4
EquipmentCategory:
- parent: ForeignKey('self') # OK
Equipment:
- category: ForeignKey(EquipmentCategory) # OK
- department: ForeignKey(Department) # 5
- responsible_person: ForeignKey(User) # 6
- procurement_request: ForeignKey(...) # OK
MaintenanceRecord:
- equipment: ForeignKey(Equipment) # OK
- performed_by: ForeignKey(User) # 7
EquipmentTransfer:
- equipment: ForeignKey(Equipment) # OK
- from_department: ForeignKey(Department) # 8
- to_department: ForeignKey(Department) # 9
- from_person: ForeignKey(User) # 10
- to_person: ForeignKey(User) # 11
- created_by: ForeignKey(User) # 12
Budget:
- department: ForeignKey(Department) # 13
```
**Итого:**
- 13 зависимостей от внешних моделей
- 9 ссылок на User (Employee)
- 4 ссылки на Department
---
## 🎯 Целевая архитектура
### Принципы pip-ready модуля:
1. **Swappable models** - модели настраиваются через settings
2. **Abstract базы** - переиспользуемая логика
3. **Минимальные зависимости** - только Django core
4. **Type hints** - полная типизация
5. **Документация** - docstrings everywhere
### Целевая структура:
```python
# procurement/settings.py (новый файл)
from django.conf import settings
PROCUREMENT_USER_MODEL = getattr(
settings,
'PROCUREMENT_USER_MODEL',
settings.AUTH_USER_MODEL
)
PROCUREMENT_DEPARTMENT_MODEL = getattr(
settings,
'PROCUREMENT_DEPARTMENT_MODEL',
'employees.Department' # default для EUSRR
)
# procurement/models.py
from django.contrib.auth import get_user_model
from .conf import PROCUREMENT_USER_MODEL, PROCUREMENT_DEPARTMENT_MODEL
class ProcurementRequest(models.Model):
# ✅ Настраиваемые модели
department = models.ForeignKey(
PROCUREMENT_DEPARTMENT_MODEL,
on_delete=models.PROTECT,
...
)
requestor = models.ForeignKey(
PROCUREMENT_USER_MODEL,
on_delete=models.PROTECT,
...
)
```
---
## 📋 План рефакторинга (10 шагов)
### **Этап 1: Подготовка (1 день)**
#### ✅ Шаг 1.1: Создать бэкап
```bash
# Бэкап БД
python manage.py dumpdata procurement > backup_procurement_$(date +%Y%m%d).json
# Бэкап кода
git checkout -b refactor/procurement-pip-ready
git commit -am "Checkpoint: Before procurement refactoring"
```
#### ✅ Шаг 1.2: Запустить существующие тесты
```bash
pytest backend/tests/api/v1/procurement/ -v --tb=short
# Записать результаты как baseline
```
#### ✅ Шаг 1.3: Создать файлы конфигурации
```
procurement/
conf.py # Настройки модуля (новый)
setup.py # pip package metadata (новый)
pyproject.toml # Modern Python packaging (новый)
MANIFEST.in # Файлы для включения в пакет (новый)
```
---
### **Этап 2: Создание конфигурации (2 дня)**
#### ✅ Шаг 2.1: Создать procurement/conf.py
```python
"""
Настройки модуля procurement.
Можно переопределить в settings.py проекта:
PROCUREMENT_USER_MODEL = 'auth.User'
PROCUREMENT_DEPARTMENT_MODEL = 'myapp.Department'
"""
from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
# User model (usually Employee or auth.User)
PROCUREMENT_USER_MODEL = getattr(
settings,
'PROCUREMENT_USER_MODEL',
settings.AUTH_USER_MODEL
)
# Department model (must have: name, head, active_employees)
PROCUREMENT_DEPARTMENT_MODEL = getattr(
settings,
'PROCUREMENT_DEPARTMENT_MODEL',
'employees.Department'
)
# Approval thresholds (₽)
APPROVAL_THRESHOLD_LOW = getattr(
settings,
'PROCUREMENT_APPROVAL_THRESHOLD_LOW',
10000
)
APPROVAL_THRESHOLD_HIGH = getattr(
settings,
'PROCUREMENT_APPROVAL_THRESHOLD_HIGH',
100000
)
# Validation функция
def validate_department_model():
"""Проверяет что DEPARTMENT_MODEL имеет нужные поля"""
from django.apps import apps
try:
model = apps.get_model(PROCUREMENT_DEPARTMENT_MODEL)
except LookupError:
raise ImproperlyConfigured(
f"PROCUREMENT_DEPARTMENT_MODEL '{PROCUREMENT_DEPARTMENT_MODEL}' "
f"not found. Make sure the app is in INSTALLED_APPS."
)
# Проверяем обязательные поля
required_fields = ['name']
for field in required_fields:
if not hasattr(model, field):
raise ImproperlyConfigured(
f"PROCUREMENT_DEPARTMENT_MODEL must have '{field}' field"
)
return model
```
#### ✅ Шаг 2.2: Обновить procurement/apps.py
```python
from django.apps import AppConfig
class ProcurementConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'procurement'
verbose_name = 'Закупки и инвентарь'
def ready(self):
# Импорт signals
from . import signals
# Валидация настроек
from .conf import validate_department_model
validate_department_model()
```
---
### **Этап 3: Рефакторинг моделей (3 дня)**
#### ✅ Шаг 3.1: Обновить imports в models.py
```python
# ❌ Старое
from employees.models import Department
User = get_user_model()
# ✅ Новое
from django.apps import apps
from .conf import PROCUREMENT_USER_MODEL, PROCUREMENT_DEPARTMENT_MODEL
def get_user_model():
return apps.get_model(PROCUREMENT_USER_MODEL)
def get_department_model():
return apps.get_model(PROCUREMENT_DEPARTMENT_MODEL)
```
#### ✅ Шаг 3.2: Изменить ForeignKey на строковые ссылки
```python
# ❌ Старое
department = models.ForeignKey(
Department,
on_delete=models.PROTECT,
...
)
# ✅ Новое
department = models.ForeignKey(
PROCUREMENT_DEPARTMENT_MODEL, # Строка!
on_delete=models.PROTECT,
...
)
# ❌ Старое
requestor = models.ForeignKey(
User,
on_delete=models.PROTECT,
...
)
# ✅ Новое
requestor = models.ForeignKey(
PROCUREMENT_USER_MODEL, # Строка!
on_delete=models.PROTECT,
...
)
```
#### ✅ Шаг 3.3: Добавить type hints
```python
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from django.contrib.auth.models import AbstractUser
from django.db.models import Model
UserModel = AbstractUser
DepartmentModel = Model
else:
UserModel = None
DepartmentModel = None
class ProcurementRequest(models.Model):
# Type hints для IDE
requestor: UserModel
department: DepartmentModel
def approve(self, by_user: UserModel) -> None:
"""Одобрить заявку"""
...
```
---
### **Этап 4: Миграции (1 день)**
#### ✅ Шаг 4.1: Создать новую миграцию
```bash
python manage.py makemigrations procurement
```
**Важно:** Миграция должна использовать `swappable_dependency`:
```python
# migrations/XXXX_refactor_swappable.py
from django.conf import settings
from django.db import migrations, models
import procurement.conf
class Migration(migrations.Migration):
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
migrations.swappable_dependency(procurement.conf.PROCUREMENT_DEPARTMENT_MODEL),
('procurement', '0001_initial'),
]
operations = [
# Django автоматически создаст AlterField для ForeignKey
...
]
```
#### ✅ Шаг 4.2: Применить миграции
```bash
python manage.py migrate procurement
```
#### ✅ Шаг 4.3: Проверить БД
```bash
python manage.py dbshell
\d+ procurement_procurementrequest
# Проверить что FK ссылаются на правильные таблицы
```
---
### **Этап 5: Обновление кода (2 дня)**
#### ✅ Шаг 5.1: Обновить методы моделей
```python
# Везде где используется User/Department, заменить на get_*_model()
class ProcurementRequest(models.Model):
def check_budget_available(self):
Department = get_department_model() # ← Добавить
try:
budget = Budget.objects.get(
department=self.department,
...
)
```
#### ✅ Шаг 5.2: Обновить permissions.py
```python
# api/v1/procurement/permissions.py
from procurement.conf import get_department_model
class IsDepartmentHead(permissions.BasePermission):
def has_permission(self, request, view):
Department = get_department_model() # ← Добавить
# Проверяем через свойство head
# Предполагается что у Department есть поле head
return Department.objects.filter(head=request.user).exists()
```
#### ✅ Шаг 5.3: Обновить serializers.py
```python
# api/v1/procurement/serializers.py
from procurement.conf import PROCUREMENT_USER_MODEL
class ProcurementRequestSerializer(serializers.ModelSerializer):
requestor = serializers.PrimaryKeyRelatedField(
queryset=get_user_model().objects.all() # ← Использовать get_user_model()
)
```
---
### **Этап 6: Обновление тестов (2 дня)**
#### ✅ Шаг 6.1: Обновить conftest.py
```python
# tests/api/v1/procurement/conftest.py
from django.apps import apps
from procurement.conf import PROCUREMENT_USER_MODEL, PROCUREMENT_DEPARTMENT_MODEL
@pytest.fixture
def user_factory(db):
"""Фабрика пользователей - работает с любой моделью"""
User = apps.get_model(PROCUREMENT_USER_MODEL)
def _create(**kwargs):
return User.objects.create_user(**kwargs)
return _create
@pytest.fixture
def department_factory(db):
"""Фабрика отделов - работает с любой моделью"""
Department = apps.get_model(PROCUREMENT_DEPARTMENT_MODEL)
def _create(**kwargs):
defaults = {'name': 'Test Department'}
defaults.update(kwargs)
return Department.objects.create(**defaults)
return _create
```
#### ✅ Шаг 6.2: Запустить тесты
```bash
pytest backend/tests/api/v1/procurement/ -v
# Все 69 тестов должны пройти!
```
---
### **Этап 7: Создание pip пакета (1 день)**
#### ✅ Шаг 7.1: Создать setup.py
```python
# procurement/setup.py
from setuptools import setup, find_packages
setup(
name='django-procurement',
version='0.1.0',
description='Django app for procurement and inventory management',
long_description=open('README.md').read(),
long_description_content_type='text/markdown',
author='Your Name',
author_email='your@email.com',
url='https://github.com/yourorg/django-procurement',
packages=find_packages(),
include_package_data=True,
install_requires=[
'Django>=4.2',
'djangorestframework>=3.14',
'Pillow>=10.0', # Для QR кодов
],
extras_require={
'dev': [
'pytest>=7.0',
'pytest-django>=4.5',
'black>=23.0',
'isort>=5.12',
]
},
classifiers=[
'Development Status :: 3 - Alpha',
'Framework :: Django',
'Framework :: Django :: 4.2',
'Framework :: Django :: 5.0',
'Intended Audience :: Developers',
'License :: OSI Approved :: MIT License',
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.10',
'Programming Language :: Python :: 3.11',
'Programming Language :: Python :: 3.12',
],
python_requires='>=3.10',
)
```
#### ✅ Шаг 7.2: Создать pyproject.toml
```toml
# procurement/pyproject.toml
[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "django-procurement"
version = "0.1.0"
description = "Django app for procurement and inventory management"
readme = "README.md"
requires-python = ">=3.10"
license = {text = "MIT"}
authors = [
{name = "Your Name", email = "your@email.com"}
]
keywords = ["django", "procurement", "inventory", "equipment"]
classifiers = [
"Development Status :: 3 - Alpha",
"Framework :: Django",
"Programming Language :: Python :: 3",
]
dependencies = [
"Django>=4.2",
"djangorestframework>=3.14",
"Pillow>=10.0",
]
[project.optional-dependencies]
dev = [
"pytest>=7.0",
"pytest-django>=4.5",
"black>=23.0",
"isort>=5.12",
]
[project.urls]
Homepage = "https://github.com/yourorg/django-procurement"
Documentation = "https://django-procurement.readthedocs.io"
Repository = "https://github.com/yourorg/django-procurement"
```
#### ✅ Шаг 7.3: Создать MANIFEST.in
```
# procurement/MANIFEST.in
include LICENSE
include README.md
include pyproject.toml
recursive-include procurement/templates *
recursive-include procurement/static *
recursive-include procurement/migrations *.py
recursive-exclude * __pycache__
recursive-exclude * *.pyc
recursive-exclude * *.pyo
```
---
### **Этап 8: Документация (2 дня)**
#### ✅ Шаг 8.1: Создать README.md
```markdown
# Django Procurement
Django приложение для управления закупками и инвентарем.
## Особенности
- ✅ Swappable User/Department models
- ✅ Workflow для заявок на закупку
- ✅ Управление оборудованием
- ✅ Система согласований
- ✅ REST API (DRF)
- ✅ Полные type hints
- ✅ 80%+ test coverage
## Установка
```bash
pip install django-procurement
```
## Быстрый старт
1. Добавьте в INSTALLED_APPS:
```python
INSTALLED_APPS = [
...
'rest_framework',
'procurement',
]
```
2. Настройте модели (опционально):
```python
# settings.py
PROCUREMENT_USER_MODEL = 'auth.User' # default: AUTH_USER_MODEL
PROCUREMENT_DEPARTMENT_MODEL = 'myapp.Department' # default: 'employees.Department'
# Пороги согласования (₽)
PROCUREMENT_APPROVAL_THRESHOLD_LOW = 10000
PROCUREMENT_APPROVAL_THRESHOLD_HIGH = 100000
```
3. Примените миграции:
```bash
python manage.py migrate procurement
```
4. Подключите URLs:
```python
urlpatterns = [
path('api/v1/procurement/', include('procurement.api.urls')),
]
```
## Требования к Department model
Если используется кастомная модель отдела, она должна иметь:
- `name` (CharField) - название отдела
- `head` (ForeignKey to User, optional) - руководитель
Пример:
```python
class Department(models.Model):
name = models.CharField(max_length=255)
head = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.SET_NULL,
null=True
)
```
## API
### Endpoints
- `GET /api/v1/procurement/requests/` - список заявок
- `POST /api/v1/procurement/requests/` - создать заявку
- `GET /api/v1/procurement/equipment/` - список оборудования
- ... (see docs/)
## Тестирование
```bash
pytest tests/
```
## Лицензия
MIT License
```
#### ✅ Шаг 8.2: Обновить docstrings
Добавить полные docstrings во все классы и методы с примерами использования.
---
### **Этап 9: Интеграция в EUSRR (1 день)**
#### ✅ Шаг 9.1: Обновить settings.py проекта
```python
# backend/eusrr_backend/settings.py
# Настройки procurement модуля
PROCUREMENT_USER_MODEL = 'employees.Employee' # Явно указываем
PROCUREMENT_DEPARTMENT_MODEL = 'employees.Department'
PROCUREMENT_APPROVAL_THRESHOLD_LOW = 10000
PROCUREMENT_APPROVAL_THRESHOLD_HIGH = 100000
```
#### ✅ Шаг 9.2: Проверить работу
```bash
# Запустить сервер
python manage.py runserver
# Проверить API
curl http://localhost:9000/api/v1/procurement/requests/
# Проверить админку
# Зайти в /admin/procurement/
```
---
### **Этап 10: Финализация (1 день)**
#### ✅ Шаг 10.1: Запустить полный тест-suite
```bash
# Все тесты procurement
pytest backend/tests/api/v1/procurement/ -v --cov=procurement
# Интеграционные тесты
pytest backend/tests/integration/ -k procurement
# Проверить покрытие
pytest --cov=procurement --cov-report=html
# Цель: 80%+
```
#### ✅ Шаг 10.2: Code quality checks
```bash
# Форматирование
black backend/procurement/
isort backend/procurement/
# Линтинг
flake8 backend/procurement/
mypy backend/procurement/
# Проверка безопасности
bandit -r backend/procurement/
```
#### ✅ Шаг 10.3: Создать git tag
```bash
git add .
git commit -m "refactor: Make procurement module pip-ready
- Add swappable USER/DEPARTMENT models
- Add comprehensive type hints
- Extract configuration to conf.py
- Add setup.py for pip distribution
- Update tests to work with any models
- 100% backward compatible with EUSRR
BREAKING CHANGES: None (backward compatible)
"
git tag -a v0.1.0 -m "Release 0.1.0: pip-ready procurement module"
git push origin refactor/procurement-pip-ready
git push --tags
```
---
## 🧪 Тестирование в изоляции
### Создать тестовый Django проект:
```bash
# Создать чистый проект для проверки
mkdir test-procurement
cd test-procurement
python -m venv venv
source venv/bin/activate
# Установить procurement как pip пакет
pip install -e ../EUSRR/backend/procurement/
# Создать minimal Django проект
django-admin startproject testproject
cd testproject
# Добавить в INSTALLED_APPS
# settings.py
INSTALLED_APPS = [
...
'procurement',
]
# Использовать стандартный auth.User
PROCUREMENT_USER_MODEL = 'auth.User'
# Создать минимальную Department модель
# testproject/models.py
class Department(models.Model):
name = models.CharField(max_length=255)
head = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.SET_NULL,
null=True
)
PROCUREMENT_DEPARTMENT_MODEL = 'testproject.Department'
# Запустить миграции
python manage.py migrate
# Проверить что всё работает!
python manage.py shell
>>> from procurement.models import ProcurementRequest
>>> ProcurementRequest.objects.create(...)
```
---
## ✅ Критерии успеха
### Функциональные:
- [ ] Все 69 тестов проходят
- [ ] Модуль работает с `auth.User` вместо `employees.Employee`
- [ ] Модуль работает с custom Department моделью
- [ ] API endpoints работают
- [ ] Админка работает
- [ ] Миграции применяются без ошибок
### Код quality:
- [ ] 80%+ test coverage
- [ ] Type hints везде
- [ ] Docstrings для всех public API
- [ ] Black + isort пройдены
- [ ] Flake8 без ошибок
- [ ] Mypy пройден
### Packaging:
- [ ] `pip install -e .` работает
- [ ] setup.py корректен
- [ ] pyproject.toml корректен
- [ ] MANIFEST.in включает все нужные файлы
- [ ] README.md полный
### Backward compatibility:
- [ ] EUSRR проект работает БЕЗ изменений (кроме settings)
- [ ] Существующие данные в БД не затронуты
- [ ] Frontend не требует изменений
---
## 📊 Метрики прогресса
### До рефакторинга:
```
Зависимости: 19 жестких связей (13 внешних)
Покрытие тестами: 81% (56/69 passed)
Документация: Minimal
Type hints: Частичные
pip-ready: ❌
```
### После рефакторинга:
```
Зависимости: 0 жестких связей (все swappable)
Покрытие тестами: 80%+ (все проходят)
Документация: Полная (README, docstrings)
Type hints: 100%
pip-ready: ✅
```
---
## 🚨 Риски и митигация
### Риск 1: Миграции сломают БД
**Митигация:**
- Полный бэкап перед началом
- Тестировать на копии БД
- Rollback plan готов
### Риск 2: Тесты не проходят
**Митигация:**
- Запустить baseline до изменений
- Фиксить по одному тесту
- Не коммитить если есть failed tests
### Риск 3: Breaking changes для EUSRR
**Митигация:**
- Backward compatible подход
- Только добавление, не удаление
- Settings опциональные
### Риск 4: Performance деградация
**Митигация:**
- Бенчмарки до/после
- EXPLAIN ANALYZE запросов
- Индексы на месте
---
## 📝 Чеклист перед коммитом
- [ ] Бэкап БД создан
- [ ] Все тесты проходят
- [ ] Black + isort применены
- [ ] Type hints добавлены
- [ ] Docstrings обновлены
- [ ] README.md актуален
- [ ] Миграции протестированы
- [ ] EUSRR проект работает
- [ ] Coverage 80%+
- [ ] Нет TODO/FIXME в коде
---
## 🎉 После завершения
### Опубликовать на TestPyPI:
```bash
cd backend/procurement
python -m build
twine upload --repository testpypi dist/*
```
### Опубликовать на PyPI (когда готово):
```bash
twine upload dist/*
```
### Документация:
- Создать docs/ сайт (MkDocs/Sphinx)
- Опубликовать на ReadTheDocs
---
**Время начала:** 27.02.2026
**Ожидаемое завершение:** 15.03.2026 (2-3 недели)
**Ответственный:** [Ваше имя]
**Статус:** 📝 План готов, ожидает утверждения