Игорь Надеин
3da3100328
All checks were successful
Deploy production / deploy (push) Successful in 11m10s
284 lines
10 KiB
Python
284 lines
10 KiB
Python
import calendar
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
from importlib import import_module
|
|
|
|
import pytest
|
|
from django.apps import apps
|
|
from django.core.exceptions import ValidationError
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
|
|
from api.v1.finance.admin.workspace_payload import _period_input_state
|
|
from api.v1.finance.serializers import OwnPayrollStatementSerializer
|
|
from finance.models import (
|
|
PayrollCustomComponent,
|
|
PayrollInputLine,
|
|
PayrollPeriod,
|
|
)
|
|
from finance.payroll.balances import opening_balances
|
|
from finance.payroll.components import (
|
|
build_payroll_component_snapshot,
|
|
get_payroll_components,
|
|
)
|
|
from finance.payroll.exceptions import PayrollOperationError
|
|
from finance.payroll.services import (
|
|
_ensure_run_is_fresh,
|
|
calculate_period,
|
|
)
|
|
from tests.finance.test_payroll_admin_api import grant, make_approved_period_inputs
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
@pytest.fixture
|
|
def setup(user_factory, auth_client_factory):
|
|
manager, employee, checker = user_factory(), user_factory(), user_factory()
|
|
grant(manager, "manage_payroll_inputs", "view_all_payroll", "calculate_payroll")
|
|
periods = {}
|
|
for month in (8, 9, 10, 11):
|
|
periods[month] = PayrollPeriod.objects.create(
|
|
code=f"2026-{month:02}",
|
|
date_from=date(2026, month, 1),
|
|
date_to=date(2026, month, calendar.monthrange(2026, month)[1]),
|
|
created_by=manager,
|
|
status="closed" if month < 10 else "open",
|
|
balance_rules_version=1 if month < 10 else 2,
|
|
)
|
|
component = PayrollCustomComponent.objects.create(
|
|
name="Выплата долга",
|
|
kind="payment",
|
|
include_in_salary=False,
|
|
include_in_carryover=True,
|
|
period_scope="all",
|
|
created_by=manager,
|
|
)
|
|
return manager, employee, checker, periods, component, auth_client_factory(manager)
|
|
|
|
|
|
def line(setup, period, **overrides):
|
|
manager, employee, _, _, component, _ = setup
|
|
return PayrollInputLine.objects.create(
|
|
period=period,
|
|
employee=employee,
|
|
component_code=component.code,
|
|
amount="5000",
|
|
status="approved",
|
|
created_by=manager,
|
|
approved_by=setup[2],
|
|
approved_at=timezone.now(),
|
|
**overrides,
|
|
)
|
|
|
|
|
|
def test_previous_period_payments_keep_legacy_sign_and_current_payments_reduce_debt(
|
|
setup,
|
|
):
|
|
_, employee, _, periods, component, client = setup
|
|
for month in (8, 9):
|
|
# An old v1 snapshot contains no salary-participation flag.
|
|
snapshot = build_payroll_component_snapshot(periods[month])
|
|
snapshot["version"] = 1
|
|
for payload in snapshot["components"]:
|
|
payload.pop("include_in_salary")
|
|
PayrollPeriod.objects.filter(pk=periods[month].pk).update(
|
|
component_snapshot=snapshot
|
|
)
|
|
line(setup, periods[month])
|
|
line(setup, periods[10])
|
|
assert opening_balances(periods[10], [employee.pk])[employee.pk] == Decimal("10000")
|
|
assert opening_balances(periods[11], [employee.pk])[employee.pk] == Decimal("5000")
|
|
ledger = client.get(
|
|
reverse(
|
|
"api:v1:finance-payroll:admin-period-carryovers",
|
|
kwargs={"pk": periods[10].pk},
|
|
)
|
|
)
|
|
assert ledger.status_code == 200, ledger.content
|
|
contributions = {
|
|
entry["period"]["id"]: Decimal(entry["contribution"])
|
|
for entry in ledger.json()["entries"]
|
|
}
|
|
assert contributions == {
|
|
periods[8].pk: Decimal("5000"),
|
|
periods[9].pk: Decimal("5000"),
|
|
periods[10].pk: Decimal("-5000"),
|
|
}
|
|
for month in (8, 9):
|
|
periods[month].refresh_from_db()
|
|
assert (
|
|
get_payroll_components(periods[month])
|
|
.get(code=component.code)
|
|
.include_in_salary
|
|
is True
|
|
)
|
|
|
|
|
|
def test_excluded_payment_does_not_reduce_current_salary_or_point_base(setup):
|
|
manager, employee, checker, periods, component, _ = setup
|
|
make_approved_period_inputs(
|
|
period=periods[10], employee=employee, creator=manager, approver=checker
|
|
)
|
|
component.include_in_point_base = (
|
|
True # Defensive adapter protection, even against bad ORM data.
|
|
)
|
|
component.save(update_fields=["include_in_point_base"])
|
|
line(setup, periods[10])
|
|
run = calculate_period(periods[10].pk, actor=manager)
|
|
statement = run.statements.get(employee=employee)
|
|
assert statement.payable == Decimal("80000")
|
|
assert statement.gross_total == Decimal("80000")
|
|
assert statement.payment_total == Decimal("5000")
|
|
assert statement.closing_balance == Decimal("-5000")
|
|
assert statement.input_snapshot["point_base_accrual"] == "80000"
|
|
assert statement.result_snapshot["totals"]["non_salary_payment_total"] == "5000"
|
|
assert statement.input_snapshot["lines"][0]["include_in_salary"] is False
|
|
detail = OwnPayrollStatementSerializer(statement).data
|
|
assert Decimal(detail["non_salary_payment_total"]) == Decimal("5000")
|
|
assert (
|
|
next(item for item in detail["lines"] if item["code"] == component.code)[
|
|
"include_in_salary"
|
|
]
|
|
is False
|
|
)
|
|
|
|
|
|
def test_inactive_employee_payment_without_work_does_not_block_or_generate_payslip(
|
|
setup, user_factory
|
|
):
|
|
manager, employee, checker, periods, component, client = setup
|
|
make_approved_period_inputs(
|
|
period=periods[10], employee=employee, creator=manager, approver=checker
|
|
)
|
|
former = user_factory(active=False)
|
|
PayrollInputLine.objects.create(
|
|
period=periods[10],
|
|
employee=former,
|
|
component_code=component.code,
|
|
amount="3500",
|
|
status="approved",
|
|
created_by=manager,
|
|
approved_by=checker,
|
|
approved_at=timezone.now(),
|
|
)
|
|
state = _period_input_state(periods[10])
|
|
assert state.approved_input_outside_roster_employee_ids == []
|
|
run = calculate_period(periods[10].pk, actor=manager)
|
|
assert not run.statements.filter(employee=former).exists()
|
|
table = client.get(
|
|
reverse(
|
|
"api:v1:finance-payroll:admin-period-table", kwargs={"pk": periods[10].pk}
|
|
)
|
|
).json()
|
|
row = next(item for item in table["rows"] if item["employee"]["id"] == former.pk)
|
|
assert row["status"] == "balance_only"
|
|
assert Decimal(row["payment_total"]) == Decimal("3500")
|
|
assert Decimal(row["payable"]) == 0
|
|
assert Decimal(row["closing_balance"]) == Decimal("-3500")
|
|
assert Decimal(table["recorded_payments"]["total_recorded"]) == Decimal("3500")
|
|
_ensure_run_is_fresh(run, periods[10])
|
|
PayrollInputLine.objects.create(
|
|
period=periods[10],
|
|
employee=former,
|
|
component_code=component.code,
|
|
amount="100",
|
|
status="approved",
|
|
created_by=manager,
|
|
approved_by=checker,
|
|
approved_at=timezone.now(),
|
|
)
|
|
with pytest.raises(PayrollOperationError) as error:
|
|
_ensure_run_is_fresh(run, periods[10])
|
|
assert error.value.code == "RUN_STALE_RECALCULATION_REQUIRED"
|
|
|
|
|
|
@pytest.mark.parametrize("builtin", [False, True])
|
|
def test_component_setting_is_saved_and_point_base_is_cleared(setup, builtin):
|
|
_, _, _, periods, component, client = setup
|
|
code = "BONUS" if builtin else component.code
|
|
payload = {
|
|
"name": "Отдельная выплата" if builtin else component.name,
|
|
"kind": "payment",
|
|
"period_scope": "current_and_future",
|
|
"period_id": periods[10].pk,
|
|
"requires_reason": False,
|
|
"include_in_salary": False,
|
|
"include_in_carryover": True,
|
|
"include_in_point_base": True,
|
|
"visible_to_employee": False,
|
|
}
|
|
url = reverse(
|
|
"api:v1:finance-payroll:admin-component-detail", kwargs={"code": code}
|
|
)
|
|
response = client.patch(url, payload, format="json")
|
|
assert response.status_code == 200, response.content
|
|
assert response.json()["include_in_salary"] is False
|
|
assert response.json()["include_in_point_base"] is False
|
|
payload.pop("include_in_salary")
|
|
response = client.patch(url, payload, format="json")
|
|
assert response.status_code == 200, response.content
|
|
assert response.json()["include_in_salary"] is False
|
|
assert response.json()["include_in_point_base"] is False
|
|
|
|
|
|
def test_migration_preserves_old_snapshots_and_only_activates_new_open_periods(setup):
|
|
_, _, _, periods, _, _ = setup
|
|
old_snapshot = {"version": 1, "components": []}
|
|
PayrollPeriod.objects.filter(pk=periods[8].pk).update(
|
|
component_snapshot=old_snapshot
|
|
)
|
|
PayrollPeriod.objects.all().update(balance_rules_version=1)
|
|
migration = import_module(
|
|
"finance.migrations.0041_salary_participation_and_balance_rules"
|
|
)
|
|
migration.activate_current_rules(apps, None)
|
|
for month, period in periods.items():
|
|
period.refresh_from_db()
|
|
assert period.balance_rules_version == (1 if month < 10 else 2)
|
|
assert periods[8].component_snapshot == old_snapshot
|
|
september = periods[9]
|
|
assert september.component_snapshot["version"] == 1
|
|
assert all(
|
|
component.include_in_salary for component in get_payroll_components(september)
|
|
)
|
|
assert not periods[10].component_snapshot
|
|
with pytest.raises(ValidationError):
|
|
september.balance_rules_version = 2
|
|
september.save(update_fields=["balance_rules_version"])
|
|
|
|
|
|
def test_migration_keeps_existing_unclosed_run_fresh_and_freezes_its_settings(setup):
|
|
manager, employee, checker, periods, component, _ = setup
|
|
PayrollPeriod.objects.all().update(balance_rules_version=1)
|
|
current = PayrollPeriod.objects.get(pk=periods[10].pk)
|
|
component.include_in_salary = True
|
|
component.save(update_fields=["include_in_salary"])
|
|
make_approved_period_inputs(
|
|
period=current, employee=employee, creator=manager, approver=checker
|
|
)
|
|
line(setup, current)
|
|
run = calculate_period(current.pk, actor=manager)
|
|
previous = list(
|
|
run.statements.values(
|
|
"input_snapshot", "result_snapshot", "input_hash", "result_hash"
|
|
)
|
|
)
|
|
import_module(
|
|
"finance.migrations.0041_salary_participation_and_balance_rules"
|
|
).activate_current_rules(apps, None)
|
|
current.refresh_from_db()
|
|
periods[11].refresh_from_db()
|
|
assert current.balance_rules_version == 1
|
|
assert periods[11].balance_rules_version == 2
|
|
component.include_in_salary = False
|
|
component.save(update_fields=["include_in_salary"])
|
|
_ensure_run_is_fresh(run, current)
|
|
assert (
|
|
list(
|
|
run.statements.values(
|
|
"input_snapshot", "result_snapshot", "input_hash", "result_hash"
|
|
)
|
|
)
|
|
== previous
|
|
)
|