Corp/backend/tests/finance/test_balance_adjustments.py
Игорь Надеин 5ded148251
All checks were successful
Deploy production / deploy (push) Successful in 12m30s
Show balance-only employees and later adjustment notices
2026-10-20 14:39:45 +03:00

579 lines
22 KiB
Python

import calendar
import uuid
from concurrent.futures import ThreadPoolExecutor
from datetime import date
from decimal import Decimal
from threading import Barrier
import pytest
from django.contrib.auth import get_user_model
from django.db import close_old_connections, connection
from django.urls import reverse
from finance.balance_adjustments.services import create_adjustment, void_adjustment
from finance.models import (
FinanceSettings,
PayrollAuditEvent,
PayrollBalanceAdjustment,
PayrollInputLine,
PayrollPeriod,
PayrollRun,
PayrollStatement,
PayrollStatementLine,
PayrollWorkRecord,
)
from finance.payroll.balances import opening_balances
from finance.payroll.exceptions import PayrollOperationError, PayrollPermissionDenied
from finance.payroll.reporting import build_payroll_period_table
from finance.payroll.services import calculate_period
from tests.finance.test_payroll_admin_api import grant, make_approved_period_inputs
pytestmark = pytest.mark.django_db
def url(name, **kwargs):
return reverse(f"api:v1:finance-payroll:{name}", kwargs=kwargs or None)
@pytest.fixture
def setup(user_factory, auth_client_factory):
manager = user_factory()
employee = 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}",
name=f"Месяц {month}",
date_from=date(2026, month, 1),
date_to=date(2026, month, calendar.monthrange(2026, month)[1]),
status="closed" if month < 10 else "open",
created_by=manager,
)
payload = {
"employee_id": employee.pk,
"effective_period_id": periods[10].pk,
"source_period_id": periods[8].pk,
"direction": "company_owes",
"amount": "1234.56",
"request_id": str(uuid.uuid4()),
}
return manager, employee, periods, payload, auth_client_factory(manager)
def create(setup, **overrides):
_, _, _, payload, client = setup
return client.post(
url("admin-balance-adjustment-create"), {**payload, **overrides}, format="json"
)
@pytest.mark.parametrize(
"direction,contribution",
[("company_owes", "1234.56"), ("employee_owes", "-1234.56")],
)
def test_effective_from_october_once_and_no_payroll_inputs(
setup, direction, contribution
):
manager, employee, periods, _, client = setup
frozen = list(
PayrollPeriod.objects.filter(pk__in=[periods[8].pk, periods[9].pk]).values()
)
result = create(setup, direction=direction)
assert result.status_code == 201, result.content
adjustment = PayrollBalanceAdjustment.objects.get()
assert adjustment.reason == ""
assert adjustment.created_by == manager
assert not PayrollInputLine.objects.exists()
assert not PayrollStatement.objects.exists()
assert (
PayrollAuditEvent.objects.filter(
action="payroll.balance_adjustment_created"
).count()
== 1
)
for month in (8, 9):
assert opening_balances(periods[month], [employee.pk]) == {}
ledger = client.get(url("admin-period-carryovers", pk=periods[month].pk)).json()
assert ledger["entries"] == []
for month in (10, 11):
assert opening_balances(periods[month], [employee.pk]) == {
employee.pk: Decimal(contribution)
}
ledger = client.get(url("admin-period-carryovers", pk=periods[month].pk)).json()
assert ledger["summary"]["balance"] == contribution
assert len(ledger["entries"]) == 1
entry = ledger["entries"][0]
assert entry["entry_type"] == "manual_adjustment"
assert entry["can_void"] is True
assert entry["source_period"]["id"] == periods[8].pk
assert frozen == list(
PayrollPeriod.objects.filter(pk__in=[periods[8].pk, periods[9].pk]).values()
)
def test_idempotency_and_conflicting_retries(setup):
assert create(setup).status_code == 201
assert create(setup).status_code == 200
result = create(setup, amount="99")
assert result.status_code == 409, result.content
assert result.json()["code"] == "BALANCE_ADJUSTMENT_REQUEST_CONFLICT"
assert PayrollBalanceAdjustment.objects.count() == 1
@pytest.mark.django_db(transaction=True)
def test_concurrent_retries_are_exactly_once_on_postgres(setup):
if connection.vendor != "postgresql":
pytest.skip("PostgreSQL row-lock integration test")
manager, employee, periods, payload, _ = setup
FinanceSettings.get_solo()
barrier = Barrier(2)
def record():
close_old_connections()
try:
actor = get_user_model().objects.get(pk=manager.pk)
person = get_user_model().objects.get(pk=employee.pk)
effective = PayrollPeriod.objects.get(pk=periods[10].pk)
source = PayrollPeriod.objects.get(pk=periods[8].pk)
barrier.wait(timeout=15)
adjustment, created = create_adjustment(
actor=actor,
employee=person,
effective_period=effective,
source_period=source,
direction="company_owes",
amount=Decimal("1234.56"),
reason="",
request_id=uuid.UUID(payload["request_id"]),
)
return adjustment.pk, created
finally:
close_old_connections()
with ThreadPoolExecutor(max_workers=2) as pool:
first = pool.submit(record)
second = pool.submit(record)
results = [first.result(timeout=30), second.result(timeout=30)]
assert results[0][0] == results[1][0]
assert sorted(result[1] for result in results) == [False, True]
assert PayrollBalanceAdjustment.objects.count() == 1
assert (
PayrollAuditEvent.objects.filter(
action="payroll.balance_adjustment_created"
).count()
== 1
)
def test_void_keeps_history_and_cancels_only_balance(setup):
manager, employee, periods, _, client = setup
assert create(setup).status_code == 201
adjustment = PayrollBalanceAdjustment.objects.get()
result = client.post(
url("admin-balance-adjustment-void", pk=adjustment.pk), {}, format="json"
)
assert result.status_code == 200, result.content
assert (
client.post(
url("admin-balance-adjustment-void", pk=adjustment.pk), {}, format="json"
).status_code
== 200
)
adjustment.refresh_from_db()
assert adjustment.voided_by == manager
assert adjustment.voided_at
assert not opening_balances(periods[10], [employee.pk])
entry = client.get(url("admin-period-carryovers", pk=periods[10].pk)).json()[
"entries"
][0]
assert entry["status"] == "voided"
assert entry["contribution"] == "0.00"
assert entry["can_void"] is False
assert entry["voided_by"]["id"] == manager.pk
assert (
PayrollAuditEvent.objects.filter(
action="payroll.balance_adjustment_voided"
).count()
== 1
)
assert not PayrollInputLine.objects.exists()
@pytest.mark.parametrize(
"status", ["calculated", "review", "approved", "published", "closed"]
)
def test_calculated_and_later_frozen_periods_block_create_and_void(setup, status):
_, employee, periods, _, client = setup
assert create(setup).status_code == 201
adjustment = PayrollBalanceAdjustment.objects.get()
PayrollPeriod.objects.filter(pk=periods[11].pk).update(status=status)
assert create(setup, request_id=str(uuid.uuid4())).status_code == 409
result = client.post(
url("admin-balance-adjustment-void", pk=adjustment.pk), {}, format="json"
)
assert result.status_code == 409
assert opening_balances(periods[10], [employee.pk])[employee.pk] == Decimal(
"1234.56"
)
entry = client.get(url("admin-period-carryovers", pk=periods[10].pk)).json()[
"entries"
][0]
assert entry["can_void"] is False
def test_closed_effective_period_rejected(setup):
assert create(setup, effective_period_id=setup[2][9].pk).status_code == 409
assert not PayrollBalanceAdjustment.objects.exists()
def test_future_effective_period_does_not_change_current_balance(setup):
_, employee, periods, _, client = setup
assert create(setup, effective_period_id=periods[11].pk).status_code == 201
assert opening_balances(periods[10], [employee.pk]) == {}
assert (
client.get(url("admin-period-carryovers", pk=periods[10].pk)).json()["entries"]
== []
)
assert opening_balances(periods[11], [employee.pk])[employee.pk] == Decimal(
"1234.56"
)
def test_retry_of_already_voided_adjustment_after_closing_is_read_only(setup):
_, _, periods, _, client = setup
assert create(setup).status_code == 201
adjustment = PayrollBalanceAdjustment.objects.get()
endpoint = url("admin-balance-adjustment-void", pk=adjustment.pk)
assert client.post(endpoint, {}, format="json").status_code == 200
PayrollPeriod.objects.filter(pk=periods[10].pk).update(status="closed")
assert client.post(endpoint, {}, format="json").status_code == 200
assert (
PayrollAuditEvent.objects.filter(
action="payroll.balance_adjustment_voided"
).count()
== 1
)
@pytest.mark.parametrize(
"overrides",
[
{"amount": "0"},
{"amount": "-1"},
{"amount": "1.001"},
{"direction": "unknown"},
{"request_id": "invalid"},
],
)
def test_invalid_input(setup, overrides):
assert create(setup, **overrides).status_code == 400
assert not PayrollBalanceAdjustment.objects.exists()
def test_source_is_reference_only_and_must_be_earlier_same_currency(setup):
periods = setup[2]
assert create(setup, source_period_id=periods[10].pk).status_code == 409
PayrollPeriod.objects.filter(pk=periods[8].pk).update(currency="USD")
assert create(setup).status_code == 409
assert create(setup, source_period_id=None).status_code == 201
def test_inactive_allowed_hidden_forbidden(setup):
employee = setup[1]
employee.is_active = False
employee.save(update_fields=["is_active"])
assert create(setup).status_code == 201
employee.is_hidden_in_ui = True
employee.save(update_fields=["is_hidden_in_ui"])
assert create(setup, request_id=str(uuid.uuid4())).status_code == 400
def test_permissions_and_management_allowlist(setup, user_factory, auth_client_factory):
manager, employee, periods, payload, client = setup
outsider = user_factory()
grant(outsider, "view_all_payroll")
outsider_client = auth_client_factory(outsider)
assert (
outsider_client.post(
url("admin-balance-adjustment-create"), payload, format="json"
).status_code
== 403
)
with pytest.raises(PayrollPermissionDenied):
create_adjustment(
actor=outsider, employee=employee, effective_period=periods[10]
)
settings = FinanceSettings.get_solo()
settings.management_users.add(outsider)
assert create(setup).status_code == 403
with pytest.raises(PayrollPermissionDenied):
create_adjustment(
actor=manager, employee=employee, effective_period=periods[10]
)
def test_preview_and_calculation_change_only_balance_preserve_september_snapshot(
setup, user_factory
):
manager, employee, periods, _, client = setup
approver = user_factory()
september = periods[9]
september.status = "open"
september.save(update_fields=["status"])
make_approved_period_inputs(
period=september, employee=employee, creator=manager, approver=approver
)
september_run = calculate_period(september.pk, actor=manager)
september.refresh_from_db()
september.status = "closed"
september.save(update_fields=["status"])
snapshot_before = list(PayrollStatement.objects.filter(run=september_run).values())
work = PayrollWorkRecord.objects.get(period=september, employee=employee)
work.pk = None
work.period = periods[10]
work.save()
before = next(
row
for row in build_payroll_period_table(periods[10])["rows"]
if row["employee"]["id"] == employee.pk
)
assert create(setup).status_code == 201
after = next(
row
for row in build_payroll_period_table(periods[10])["rows"]
if row["employee"]["id"] == employee.pk
)
assert Decimal(after["opening_balance"]) - Decimal(
before["opening_balance"]
) == Decimal("1234.56")
assert Decimal(after["closing_balance"]) - Decimal(
before["closing_balance"]
) == Decimal("1234.56")
for key in (
"gross_total",
"deduction_total",
"net_pay",
"payment_total",
"payable",
):
assert before[key] == after[key], key
run = calculate_period(periods[10].pk, actor=manager)
statement = PayrollStatement.objects.get(run=run, employee=employee)
assert statement.opening_balance == Decimal(after["opening_balance"])
assert statement.closing_balance == Decimal(after["closing_balance"])
assert statement.payable == Decimal(before["payable"])
assert snapshot_before == list(
PayrollStatement.objects.filter(run=september_run).values()
)
assert not PayrollInputLine.objects.exists()
adjustment = PayrollBalanceAdjustment.objects.get()
assert (
client.post(
url("admin-balance-adjustment-void", pk=adjustment.pk), {}, format="json"
).status_code
== 409
)
with pytest.raises(PayrollOperationError):
void_adjustment(adjustment.pk, actor=manager)
@pytest.mark.parametrize(
"direction,expected", [("company_owes", "1234.56"), ("employee_owes", "-1234.56")]
)
def test_inactive_employee_with_only_manual_balance_is_visible_without_salary(
setup, user_factory, direction, expected
):
manager, employee, periods, _, client = setup
# An old approved salary rate must not turn a balance-only row into salary.
make_approved_period_inputs(
period=periods[8], employee=employee, creator=manager, approver=user_factory()
)
employee.is_active = False
employee.save(update_fields=["is_active"])
assert create(setup, direction=direction).status_code == 201
for month in (10, 11):
table = client.get(url("admin-period-table", pk=periods[month].pk))
assert table.status_code == 200, table.content
row = next(
r for r in table.json()["rows"] if r["employee"]["id"] == employee.pk
)
assert row["status"] == "balance_only"
assert row["opening_balance"] == expected
assert row["closing_balance"] == expected
assert row["rate_amount"] is None
assert row["attendance_alert"] is None
for key in (
"gross_total",
"deduction_total",
"net_pay",
"payment_total",
"payable",
):
assert Decimal(row[key]) == 0, key
assert len(row["manual_balance_adjustments"]) == int(month == 10)
assert not PayrollStatement.objects.exists()
assert not PayrollInputLine.objects.exists()
def test_current_offsetting_adjustments_are_visible_but_settled_balance_is_not_carried(
setup,
):
employee, periods = setup[1:3]
employee.is_active = False
employee.save(update_fields=["is_active"])
assert create(setup).status_code == 201
assert (
create(
setup, direction="employee_owes", request_id=str(uuid.uuid4())
).status_code
== 201
)
october = build_payroll_period_table(periods[10])
row = next(r for r in october["rows"] if r["employee"]["id"] == employee.pk)
assert row["status"] == "balance_only" and Decimal(row["closing_balance"]) == 0
assert len(row["manual_balance_adjustments"]) == 2
assert employee.pk not in {
r["employee"]["id"] for r in build_payroll_period_table(periods[11])["rows"]
}
def test_later_adjustment_is_reference_only_in_original_closed_period(
setup, user_factory
):
manager, employee, periods, _, client = setup
source = periods[8]
source.status = "open"
source.save(update_fields=["status"])
make_approved_period_inputs(
period=source, employee=employee, creator=manager, approver=user_factory()
)
calculate_period(source.pk, actor=manager)
source.refresh_from_db()
source.status = "closed"
source.save(update_fields=["status"])
before_table = client.get(url("admin-period-table", pk=source.pk)).json()
tracked_models = (
PayrollPeriod,
PayrollRun,
PayrollStatement,
PayrollStatementLine,
PayrollInputLine,
)
before = {
model: list(model.objects.order_by("pk").values()) for model in tracked_models
}
assert (
create(setup, direction="employee_owes", reason="Доплата за август").status_code
== 201
)
adjustment = PayrollBalanceAdjustment.objects.get()
for _ in range(2):
response = client.get(url("admin-period-table", pk=source.pk))
assert response.status_code == 200, response.content
after_table = response.json()
reference = after_table["later_balance_adjustments"][0]
assert reference["id"] == adjustment.pk
assert reference["employee"]["id"] == employee.pk
assert reference["effective_period"]["id"] == periods[10].pk
assert reference["source_period_id"] == source.pk
assert reference["contribution"] == "-1234.56"
assert reference["reason"] == "Доплата за август"
assert reference["is_voided"] is False
assert after_table["rows"] == before_table["rows"]
assert after_table["summary"] == before_table["summary"]
for model in tracked_models:
assert list(model.objects.order_by("pk").values()) == before[model], model
def test_reference_is_shown_even_without_historical_employee_row_and_tracks_voiding(
setup,
):
_, employee, periods, _, client = setup
employee.is_active = False
employee.save(update_fields=["is_active"])
assert create(setup).status_code == 201
adjustment = PayrollBalanceAdjustment.objects.get()
source = build_payroll_period_table(periods[8])
assert employee.pk not in {row["employee"]["id"] for row in source["rows"]}
assert source["later_balance_adjustments"][0]["employee"]["id"] == employee.pk
assert (
build_payroll_period_table(periods[9])["later_balance_adjustments"][0]["id"]
== adjustment.pk
)
assert not build_payroll_period_table(periods[10])["later_balance_adjustments"]
assert (
client.post(
url("admin-balance-adjustment-void", pk=adjustment.pk), {}, format="json"
).status_code
== 200
)
after = build_payroll_period_table(periods[8])
assert after["later_balance_adjustments"][0]["is_voided"] is True
assert employee.pk not in {
row["employee"]["id"] for row in build_payroll_period_table(periods[10])["rows"]
}
employee.is_hidden_in_ui = True
employee.save(update_fields=["is_hidden_in_ui"])
assert not build_payroll_period_table(periods[8])["later_balance_adjustments"]
def test_balance_only_employee_is_not_added_to_calculation_or_payslips(
setup, user_factory
):
manager, former, periods, _, client = setup
former.is_active = False
former.save(update_fields=["is_active"])
worker = user_factory()
make_approved_period_inputs(
period=periods[10], employee=worker, creator=manager, approver=user_factory()
)
assert create(setup).status_code == 201
run = calculate_period(periods[10].pk, actor=manager)
assert set(run.statements.values_list("employee_id", flat=True)) == {worker.pk}
periods[10].refresh_from_db()
table = client.get(url("admin-period-table", pk=periods[10].pk)).json()
row = next(r for r in table["rows"] if r["employee"]["id"] == former.pk)
assert row["status"] == "balance_only"
assert row["closing_balance"] == "1234.56"
assert Decimal(row["payable"]) == 0
assert row["payment_confirmation"] is None
def test_future_notice_covers_source_and_intermediate_months_without_changing_totals(
setup,
):
_, employee, periods, _, _ = setup
july = PayrollPeriod.objects.create(
code="2026-07",
date_from=date(2026, 7, 1),
date_to=date(2026, 7, 31),
status="closed",
)
before = {
month: build_payroll_period_table(periods[month]) for month in (8, 9, 10, 11)
}
frozen = list(PayrollPeriod.objects.order_by("pk").values())
assert create(setup, effective_period_id=periods[11].pk).status_code == 201
adjustment = PayrollBalanceAdjustment.objects.get()
for month in (8, 9, 10):
after = build_payroll_period_table(periods[month])
assert after["rows"] == before[month]["rows"]
assert after["summary"] == before[month]["summary"]
reference = after["later_balance_adjustments"][0]
assert reference["id"] == adjustment.pk
assert reference["source_period"]["id"] == periods[8].pk
assert reference["effective_period"]["id"] == periods[11].pk
assert opening_balances(periods[month], [employee.pk]) == {}
assert not build_payroll_period_table(july)["later_balance_adjustments"]
assert not build_payroll_period_table(periods[11])["later_balance_adjustments"]
assert frozen == list(PayrollPeriod.objects.order_by("pk").values())
def test_no_intermediate_notice_when_source_period_was_not_specified(setup):
assert create(setup, source_period_id=None).status_code == 201
for month in (8, 9):
assert not build_payroll_period_table(setup[2][month])[
"later_balance_adjustments"
]