Игорь Надеин
3da3100328
All checks were successful
Deploy production / deploy (push) Successful in 11m10s
998 lines
34 KiB
Python
998 lines
34 KiB
Python
"""Deterministic reference implementation of the payroll calculator."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from datetime import date, datetime
|
|
from decimal import Context, Decimal, DecimalException, localcontext
|
|
from typing import Protocol
|
|
|
|
from .balances import carryover_contribution
|
|
from .domain import (
|
|
CalculationWarning,
|
|
ExpectedTotals,
|
|
InputLine,
|
|
LineKind,
|
|
PayrollLine,
|
|
PayrollPeriod,
|
|
PayrollRequest,
|
|
PayrollResult,
|
|
PayrollRules,
|
|
PayrollTotals,
|
|
PointPolicy,
|
|
RoundingPolicy,
|
|
decimal_text,
|
|
)
|
|
from .exceptions import PayrollIssue, PayrollValidationError
|
|
|
|
_CURRENCY_RE = re.compile(r"^[A-Z]{3}$")
|
|
_CODE_RE = re.compile(r"^[A-Z][A-Z0-9_.-]{0,63}$")
|
|
_SYSTEM_LINE_PREFIX = "system:"
|
|
|
|
|
|
class PayrollCalculator(Protocol):
|
|
"""Port implemented by any compatible calculation provider."""
|
|
|
|
def calculate(
|
|
self,
|
|
request: PayrollRequest,
|
|
rules: PayrollRules,
|
|
) -> PayrollResult: ...
|
|
|
|
|
|
def _canonical_hash(payload: object) -> str:
|
|
encoded = json.dumps(
|
|
payload,
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
return hashlib.sha256(encoded).hexdigest()
|
|
|
|
|
|
def _is_finite_decimal(value: object) -> bool:
|
|
return isinstance(value, Decimal) and value.is_finite()
|
|
|
|
|
|
class DeterministicPayrollCalculator:
|
|
"""Pure calculator with explicit Decimal and rounding semantics."""
|
|
|
|
CALCULATOR_VERSION = "1.4.0"
|
|
DECIMAL_PRECISION = 80
|
|
MAX_DECIMAL_DIGITS = 30
|
|
MAX_FRACTIONAL_PLACES = 12
|
|
MAX_INPUT_LINES = 10_000
|
|
|
|
def calculate(
|
|
self,
|
|
request: PayrollRequest,
|
|
rules: PayrollRules,
|
|
) -> PayrollResult:
|
|
issues = self._validate(request, rules)
|
|
if issues:
|
|
raise PayrollValidationError(issues)
|
|
|
|
decimal_context = Context(
|
|
prec=self.DECIMAL_PRECISION,
|
|
rounding=rules.rounding.value,
|
|
)
|
|
with localcontext(decimal_context):
|
|
quantize = self._quantizer(rules)
|
|
lines: list[PayrollLine] = [
|
|
PayrollLine(
|
|
line_id="system:base",
|
|
code="BASE",
|
|
label="Base accrual",
|
|
kind=LineKind.EARNING,
|
|
amount=quantize(request.base_accrual),
|
|
source_ref=request.base_source_ref,
|
|
calculated=True,
|
|
)
|
|
]
|
|
opening_balance = quantize(request.opening_balance)
|
|
warnings: list[CalculationWarning] = []
|
|
point_delta: Decimal | None = None
|
|
|
|
if rules.point_policy is PointPolicy.PROPORTIONAL_WITH_EXCESS:
|
|
point_delta = request.actual_points - request.target_points
|
|
point_base_accrual = (
|
|
request.point_base_accrual
|
|
if request.point_base_accrual is not None
|
|
else request.base_accrual
|
|
)
|
|
raw_point_amount = (
|
|
point_delta * request.point_rate
|
|
if point_delta > 0 and request.point_rate_overridden
|
|
else point_base_accrual
|
|
* (request.actual_points / request.target_points)
|
|
- point_base_accrual
|
|
)
|
|
point_kind = (
|
|
LineKind.ADJUSTMENT_DEBIT
|
|
if raw_point_amount < 0
|
|
else LineKind.ADJUSTMENT_CREDIT
|
|
)
|
|
lines.append(
|
|
PayrollLine(
|
|
line_id="system:point-adjustment",
|
|
code="POINT_ADJUSTMENT",
|
|
label="Point-based recalculation",
|
|
kind=point_kind,
|
|
amount=quantize(abs(raw_point_amount)),
|
|
source_ref="calculation:point-policy",
|
|
calculated=True,
|
|
)
|
|
)
|
|
elif rules.point_policy is PointPolicy.EXCESS_ONLY:
|
|
point_delta = request.actual_points - request.target_points
|
|
eligible_points = max(point_delta, Decimal("0"))
|
|
point_amount = quantize(eligible_points * request.point_rate)
|
|
lines.append(
|
|
PayrollLine(
|
|
line_id="system:point-excess",
|
|
code="POINT_EXCESS",
|
|
label="Points above target",
|
|
kind=LineKind.EARNING,
|
|
amount=point_amount,
|
|
source_ref="calculation:point-policy",
|
|
calculated=True,
|
|
)
|
|
)
|
|
if point_delta < 0:
|
|
warnings.append(
|
|
CalculationWarning(
|
|
code="POINTS_BELOW_TARGET_NO_DEDUCTION",
|
|
field="actual_points",
|
|
message=(
|
|
"Actual points are below target; the active "
|
|
"policy does not reduce pay."
|
|
),
|
|
context=(("point_delta", decimal_text(point_delta)),),
|
|
)
|
|
)
|
|
|
|
for input_line in sorted(
|
|
request.lines,
|
|
key=lambda item: item.line_id,
|
|
):
|
|
lines.append(self._to_result_line(input_line, quantize))
|
|
|
|
totals = self._calculate_totals(
|
|
lines,
|
|
opening_balance=opening_balance,
|
|
balance_rules_version=request.balance_rules_version,
|
|
)
|
|
if totals.payable < 0 and not rules.allow_negative_payable:
|
|
raise PayrollValidationError(
|
|
[
|
|
PayrollIssue(
|
|
code="NEGATIVE_PAYABLE_NOT_ALLOWED",
|
|
field="lines",
|
|
message=(
|
|
"Deductions and recorded payments exceed available pay."
|
|
),
|
|
context={"payable": decimal_text(totals.payable)},
|
|
)
|
|
]
|
|
)
|
|
|
|
warnings.extend(
|
|
self._reconcile_expected(
|
|
request.expected_totals,
|
|
lines,
|
|
totals,
|
|
quantize,
|
|
)
|
|
)
|
|
|
|
input_hash = _canonical_hash(request.calculation_dict())
|
|
rules_payload = rules.to_dict()
|
|
ruleset_hash = _canonical_hash(rules_payload)
|
|
calculator_version = (
|
|
"1.3.0"
|
|
if request.balance_rules_version == 1
|
|
else self.CALCULATOR_VERSION
|
|
)
|
|
result_payload = {
|
|
"employee_ref": request.employee_ref,
|
|
"period": request.period.to_dict(),
|
|
"currency": request.currency,
|
|
"lines": [line.to_dict() for line in lines],
|
|
"point_delta": (
|
|
decimal_text(point_delta) if point_delta is not None else None
|
|
),
|
|
"totals": totals.to_dict(),
|
|
"ruleset_id": rules.ruleset_id,
|
|
"ruleset_version": rules.version,
|
|
"ruleset_hash": ruleset_hash,
|
|
"calculator_version": calculator_version,
|
|
"rounding_policy": rules.rounding.value,
|
|
"input_hash": input_hash,
|
|
}
|
|
result_hash = _canonical_hash(result_payload)
|
|
reconciliation_hash = _canonical_hash(
|
|
{
|
|
"result_hash": result_hash,
|
|
"warnings": [warning.to_dict() for warning in warnings],
|
|
}
|
|
)
|
|
|
|
return PayrollResult(
|
|
employee_ref=request.employee_ref,
|
|
period=request.period,
|
|
currency=request.currency,
|
|
lines=tuple(lines),
|
|
point_delta=point_delta,
|
|
totals=totals,
|
|
ruleset_id=rules.ruleset_id,
|
|
ruleset_version=rules.version,
|
|
ruleset_hash=ruleset_hash,
|
|
calculator_version=calculator_version,
|
|
rounding_policy=rules.rounding.value,
|
|
input_hash=input_hash,
|
|
result_hash=result_hash,
|
|
reconciliation_hash=reconciliation_hash,
|
|
warnings=tuple(warnings),
|
|
)
|
|
|
|
@staticmethod
|
|
def _quantizer(rules: PayrollRules):
|
|
def quantize(value: Decimal) -> Decimal:
|
|
try:
|
|
units = (value / rules.money_quantum).quantize(
|
|
Decimal("1"),
|
|
rounding=rules.rounding.value,
|
|
)
|
|
return units * rules.money_quantum
|
|
except DecimalException as exc:
|
|
raise PayrollValidationError(
|
|
[
|
|
PayrollIssue(
|
|
code="ARITHMETIC_ERROR",
|
|
message="A monetary value cannot be rounded safely.",
|
|
)
|
|
]
|
|
) from exc
|
|
|
|
return quantize
|
|
|
|
@staticmethod
|
|
def _to_result_line(input_line: InputLine, quantize) -> PayrollLine:
|
|
return PayrollLine(
|
|
line_id=input_line.line_id,
|
|
code=input_line.code,
|
|
label=input_line.label,
|
|
kind=input_line.kind,
|
|
amount=quantize(input_line.amount),
|
|
source_ref=input_line.source_ref,
|
|
reason=input_line.reason,
|
|
source_period=input_line.source_period,
|
|
is_retro=input_line.is_retro,
|
|
calculated=False,
|
|
include_in_carryover=input_line.include_in_carryover,
|
|
include_in_salary=input_line.include_in_salary,
|
|
)
|
|
|
|
@staticmethod
|
|
def _calculate_totals(
|
|
lines: list[PayrollLine],
|
|
*,
|
|
opening_balance: Decimal = Decimal("0"),
|
|
balance_rules_version: int = 2,
|
|
) -> PayrollTotals:
|
|
earnings = sum(
|
|
(
|
|
line.amount
|
|
for line in lines
|
|
if line.kind is LineKind.EARNING and line.include_in_salary
|
|
),
|
|
Decimal("0"),
|
|
)
|
|
adjustment_credits = sum(
|
|
(
|
|
line.amount
|
|
for line in lines
|
|
if line.kind is LineKind.ADJUSTMENT_CREDIT and line.include_in_salary
|
|
),
|
|
Decimal("0"),
|
|
)
|
|
adjustment_debits = sum(
|
|
(
|
|
line.amount
|
|
for line in lines
|
|
if line.kind is LineKind.ADJUSTMENT_DEBIT and line.include_in_salary
|
|
),
|
|
Decimal("0"),
|
|
)
|
|
deductions = sum(
|
|
(
|
|
line.amount
|
|
for line in lines
|
|
if line.kind is LineKind.DEDUCTION and line.include_in_salary
|
|
),
|
|
Decimal("0"),
|
|
)
|
|
payments = sum(
|
|
(line.amount for line in lines if line.kind is LineKind.PAYMENT),
|
|
Decimal("0"),
|
|
)
|
|
non_salary_payments = sum(
|
|
(
|
|
line.amount
|
|
for line in lines
|
|
if line.kind is LineKind.PAYMENT and not line.include_in_salary
|
|
),
|
|
Decimal("0"),
|
|
)
|
|
carryover_change = sum(
|
|
(
|
|
carryover_contribution(
|
|
line.kind.value, line.amount, version=balance_rules_version
|
|
)
|
|
for line in lines
|
|
if not line.calculated and line.include_in_carryover
|
|
),
|
|
Decimal("0"),
|
|
)
|
|
adjustment_total = adjustment_credits - adjustment_debits
|
|
gross = earnings + adjustment_total
|
|
net_pay = gross - deductions
|
|
return PayrollTotals(
|
|
gross_before_adjustments=earnings,
|
|
adjustment_total=adjustment_total,
|
|
gross_after_adjustments=gross,
|
|
deduction_total=deductions,
|
|
net_pay=net_pay,
|
|
payment_total=payments,
|
|
payable=net_pay - payments + non_salary_payments,
|
|
non_salary_payment_total=non_salary_payments,
|
|
closing_balance=opening_balance + carryover_change,
|
|
)
|
|
|
|
@staticmethod
|
|
def _reconcile_expected(
|
|
expected: ExpectedTotals | None,
|
|
lines: list[PayrollLine],
|
|
totals: PayrollTotals,
|
|
quantize,
|
|
) -> list[CalculationWarning]:
|
|
if expected is None:
|
|
return []
|
|
|
|
point_line = next(
|
|
(
|
|
line
|
|
for line in lines
|
|
if line.line_id in {"system:point-excess", "system:point-adjustment"}
|
|
),
|
|
None,
|
|
)
|
|
point_amount = (
|
|
-point_line.amount
|
|
if point_line is not None and point_line.kind is LineKind.ADJUSTMENT_DEBIT
|
|
else point_line.amount
|
|
if point_line is not None
|
|
else Decimal("0")
|
|
)
|
|
gross_before_point_recalculation = totals.gross_after_adjustments - point_amount
|
|
checks = (
|
|
("point_amount", expected.point_amount, point_amount),
|
|
("gross", expected.gross, gross_before_point_recalculation),
|
|
(
|
|
"recalculated_gross",
|
|
expected.recalculated_gross,
|
|
totals.recalculated_gross,
|
|
),
|
|
("payable", expected.payable, totals.payable),
|
|
)
|
|
warnings = []
|
|
for field_name, expected_value, actual_value in checks:
|
|
if expected_value is None:
|
|
continue
|
|
normalized_expected = quantize(expected_value)
|
|
if normalized_expected != actual_value:
|
|
warnings.append(
|
|
CalculationWarning(
|
|
code="LEGACY_EXPECTED_TOTAL_MISMATCH",
|
|
field=f"expected_totals.{field_name}",
|
|
message=(
|
|
"The calculated amount differs from the legacy "
|
|
"control total."
|
|
),
|
|
context=(
|
|
("expected", decimal_text(normalized_expected)),
|
|
("actual", decimal_text(actual_value)),
|
|
),
|
|
)
|
|
)
|
|
return warnings
|
|
|
|
def _validate(
|
|
self,
|
|
request: PayrollRequest,
|
|
rules: PayrollRules,
|
|
) -> list[PayrollIssue]:
|
|
if not isinstance(request, PayrollRequest):
|
|
return [
|
|
PayrollIssue(
|
|
code="INVALID_REQUEST",
|
|
field="request",
|
|
message="Request must be a PayrollRequest instance.",
|
|
)
|
|
]
|
|
if not isinstance(rules, PayrollRules):
|
|
return [
|
|
PayrollIssue(
|
|
code="INVALID_RULESET",
|
|
field="rules",
|
|
message="Rules must be a PayrollRules instance.",
|
|
)
|
|
]
|
|
|
|
issues: list[PayrollIssue] = []
|
|
period_is_valid = self._validate_request_fields(request, issues)
|
|
self._validate_rules(rules, request, period_is_valid, issues)
|
|
self._validate_point_metrics(request, rules, issues)
|
|
self._validate_input_lines(request, period_is_valid, issues)
|
|
self._validate_expected_totals(request, rules, issues)
|
|
return issues
|
|
|
|
def _validate_request_fields(
|
|
self,
|
|
request: PayrollRequest,
|
|
issues: list[PayrollIssue],
|
|
) -> bool:
|
|
if not self._is_nonempty_string(request.employee_ref):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="MISSING_EMPLOYEE_REF",
|
|
field="employee_ref",
|
|
message="Employee reference is required.",
|
|
)
|
|
)
|
|
if type(
|
|
request.balance_rules_version
|
|
) is not int or request.balance_rules_version not in (1, 2):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_BALANCE_RULES_VERSION",
|
|
field="balance_rules_version",
|
|
message="Unsupported carryover rules version.",
|
|
)
|
|
)
|
|
period_is_valid = self._validate_period(request.period, "period", issues)
|
|
if not isinstance(request.currency, str) or not _CURRENCY_RE.fullmatch(
|
|
request.currency
|
|
):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_CURRENCY",
|
|
field="currency",
|
|
message="Currency must be a three-letter uppercase code.",
|
|
)
|
|
)
|
|
self._validate_decimal(
|
|
request.base_accrual, "base_accrual", issues, minimum=Decimal("0")
|
|
)
|
|
self._validate_decimal(
|
|
request.opening_balance, "opening_balance", issues, minimum=None
|
|
)
|
|
self._validate_decimal(
|
|
request.point_base_accrual,
|
|
"point_base_accrual",
|
|
issues,
|
|
minimum=Decimal("0"),
|
|
allow_none=True,
|
|
)
|
|
if not self._is_nonempty_string(request.base_source_ref):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="MISSING_SOURCE_REF",
|
|
field="base_source_ref",
|
|
message="Base accrual must identify its source.",
|
|
)
|
|
)
|
|
self._validate_decimal(
|
|
request.point_rate, "point_rate", issues, minimum=Decimal("0")
|
|
)
|
|
if not isinstance(request.point_rate_overridden, bool):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_POINT_RATE_MODE",
|
|
field="point_rate_overridden",
|
|
message="Point rate override marker must be boolean.",
|
|
)
|
|
)
|
|
return period_is_valid
|
|
|
|
def _validate_rules(
|
|
self,
|
|
rules: PayrollRules,
|
|
request: PayrollRequest,
|
|
period_is_valid: bool,
|
|
issues: list[PayrollIssue],
|
|
) -> None:
|
|
if not self._is_nonempty_string(rules.ruleset_id):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_RULESET",
|
|
field="rules.ruleset_id",
|
|
message="Ruleset identifier is required.",
|
|
)
|
|
)
|
|
if not self._is_nonempty_string(rules.version):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_RULESET",
|
|
field="rules.version",
|
|
message="Ruleset version is required.",
|
|
)
|
|
)
|
|
|
|
rules_from_valid = isinstance(rules.effective_from, date) and not isinstance(
|
|
rules.effective_from, datetime
|
|
)
|
|
rules_to_valid = rules.effective_to is None or (
|
|
isinstance(rules.effective_to, date)
|
|
and not isinstance(rules.effective_to, datetime)
|
|
)
|
|
if not rules_from_valid:
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_RULESET_PERIOD",
|
|
field="rules.effective_from",
|
|
message="Ruleset effective_from must be a date.",
|
|
)
|
|
)
|
|
if not rules_to_valid:
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_RULESET_PERIOD",
|
|
field="rules.effective_to",
|
|
message="Ruleset effective_to must be a date or None.",
|
|
)
|
|
)
|
|
if (
|
|
rules_from_valid
|
|
and rules_to_valid
|
|
and rules.effective_to is not None
|
|
and rules.effective_from > rules.effective_to
|
|
):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_RULESET_PERIOD",
|
|
field="rules.effective_to",
|
|
message="Ruleset effective dates are invalid.",
|
|
)
|
|
)
|
|
if (
|
|
period_is_valid
|
|
and rules_from_valid
|
|
and rules_to_valid
|
|
and (
|
|
request.period.start < rules.effective_from
|
|
or rules.effective_to is not None
|
|
and request.period.end > rules.effective_to
|
|
)
|
|
):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="RULESET_NOT_EFFECTIVE",
|
|
field="period",
|
|
message="Ruleset is not effective for the full period.",
|
|
)
|
|
)
|
|
if not isinstance(rules.point_policy, PointPolicy):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="UNSUPPORTED_POINT_POLICY",
|
|
field="rules.point_policy",
|
|
message="Point policy is not supported.",
|
|
)
|
|
)
|
|
if not isinstance(rules.rounding, RoundingPolicy):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="UNSUPPORTED_ROUNDING_POLICY",
|
|
field="rules.rounding",
|
|
message="Rounding policy is not supported.",
|
|
)
|
|
)
|
|
if not isinstance(rules.allow_negative_payable, bool):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_RULE_OPTION",
|
|
field="rules.allow_negative_payable",
|
|
message="allow_negative_payable must be a boolean.",
|
|
)
|
|
)
|
|
self._validate_decimal(
|
|
rules.money_quantum,
|
|
"rules.money_quantum",
|
|
issues,
|
|
minimum=Decimal("0"),
|
|
exclusive_minimum=True,
|
|
)
|
|
|
|
def _validate_point_metrics(
|
|
self,
|
|
request: PayrollRequest,
|
|
rules: PayrollRules,
|
|
issues: list[PayrollIssue],
|
|
) -> None:
|
|
if rules.point_policy in {
|
|
PointPolicy.EXCESS_ONLY,
|
|
PointPolicy.PROPORTIONAL_WITH_EXCESS,
|
|
}:
|
|
self._validate_decimal(
|
|
request.target_points,
|
|
"target_points",
|
|
issues,
|
|
minimum=Decimal("0"),
|
|
exclusive_minimum=True,
|
|
)
|
|
self._validate_decimal(
|
|
request.actual_points,
|
|
"actual_points",
|
|
issues,
|
|
minimum=Decimal("0"),
|
|
)
|
|
return
|
|
if rules.point_policy is not PointPolicy.DISABLED:
|
|
return
|
|
|
|
has_target = request.target_points is not None
|
|
has_actual = request.actual_points is not None
|
|
if has_target != has_actual:
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INCOMPLETE_POINT_METRICS",
|
|
field="target_points",
|
|
message="Target and actual points must be supplied together.",
|
|
)
|
|
)
|
|
if has_target:
|
|
self._validate_decimal(
|
|
request.target_points,
|
|
"target_points",
|
|
issues,
|
|
minimum=Decimal("0"),
|
|
exclusive_minimum=True,
|
|
)
|
|
if has_actual:
|
|
self._validate_decimal(
|
|
request.actual_points,
|
|
"actual_points",
|
|
issues,
|
|
minimum=Decimal("0"),
|
|
)
|
|
|
|
def _validate_input_lines(
|
|
self,
|
|
request: PayrollRequest,
|
|
period_is_valid: bool,
|
|
issues: list[PayrollIssue],
|
|
) -> None:
|
|
if not isinstance(request.lines, tuple):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_LINES_CONTAINER",
|
|
field="lines",
|
|
message="Input lines must be an immutable tuple.",
|
|
)
|
|
)
|
|
lines = ()
|
|
else:
|
|
lines = request.lines
|
|
if len(lines) > self.MAX_INPUT_LINES:
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="TOO_MANY_INPUT_LINES",
|
|
field="lines",
|
|
message=f"At most {self.MAX_INPUT_LINES} input lines are allowed.",
|
|
)
|
|
)
|
|
|
|
seen_line_ids: set[str] = set()
|
|
for index, line in enumerate(lines):
|
|
self._validate_input_line(
|
|
line,
|
|
index=index,
|
|
request=request,
|
|
period_is_valid=period_is_valid,
|
|
seen_line_ids=seen_line_ids,
|
|
issues=issues,
|
|
)
|
|
|
|
def _validate_input_line(
|
|
self,
|
|
line: object,
|
|
*,
|
|
index: int,
|
|
request: PayrollRequest,
|
|
period_is_valid: bool,
|
|
seen_line_ids: set[str],
|
|
issues: list[PayrollIssue],
|
|
) -> None:
|
|
prefix = f"lines[{index}]"
|
|
if not isinstance(line, InputLine):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_COMPONENT",
|
|
field=prefix,
|
|
message="Every input line must be an InputLine instance.",
|
|
)
|
|
)
|
|
return
|
|
|
|
for flag in ("include_in_salary", "include_in_carryover"):
|
|
if not isinstance(getattr(line, flag), bool):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_COMPONENT_FLAG",
|
|
field=f"{prefix}.{flag}",
|
|
message="Component participation flags must be boolean.",
|
|
)
|
|
)
|
|
if not self._is_nonempty_string(line.line_id):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="MISSING_COMPONENT_ID",
|
|
field=f"{prefix}.line_id",
|
|
message="Input line identifier is required.",
|
|
)
|
|
)
|
|
else:
|
|
if line.line_id.startswith(_SYSTEM_LINE_PREFIX):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="RESERVED_COMPONENT_ID",
|
|
field=f"{prefix}.line_id",
|
|
message="Input line uses a reserved system prefix.",
|
|
)
|
|
)
|
|
if line.line_id in seen_line_ids:
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="DUPLICATE_COMPONENT_ID",
|
|
field=f"{prefix}.line_id",
|
|
message="Input line identifiers must be unique.",
|
|
)
|
|
)
|
|
seen_line_ids.add(line.line_id)
|
|
|
|
if not isinstance(line.code, str) or not _CODE_RE.fullmatch(line.code):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_COMPONENT_CODE",
|
|
field=f"{prefix}.code",
|
|
message="Component code must use canonical uppercase form.",
|
|
)
|
|
)
|
|
if not self._is_nonempty_string(line.label):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="MISSING_COMPONENT_LABEL",
|
|
field=f"{prefix}.label",
|
|
message="Input line label is required.",
|
|
)
|
|
)
|
|
kind_is_valid = isinstance(line.kind, LineKind)
|
|
if not kind_is_valid:
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_COMPONENT_KIND",
|
|
field=f"{prefix}.kind",
|
|
message="Input line kind is not supported.",
|
|
)
|
|
)
|
|
self._validate_decimal(
|
|
line.amount, f"{prefix}.amount", issues, minimum=Decimal("0")
|
|
)
|
|
if not self._is_nonempty_string(line.source_ref):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="MISSING_SOURCE_REF",
|
|
field=f"{prefix}.source_ref",
|
|
message="Every input line must identify its source.",
|
|
)
|
|
)
|
|
self._validate_retro_line(
|
|
line,
|
|
prefix=prefix,
|
|
request=request,
|
|
period_is_valid=period_is_valid,
|
|
issues=issues,
|
|
)
|
|
|
|
def _validate_retro_line(
|
|
self,
|
|
line: InputLine,
|
|
*,
|
|
prefix: str,
|
|
request: PayrollRequest,
|
|
period_is_valid: bool,
|
|
issues: list[PayrollIssue],
|
|
) -> None:
|
|
is_retro_valid = isinstance(line.is_retro, bool)
|
|
if not is_retro_valid:
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_RETRO_FLAG",
|
|
field=f"{prefix}.is_retro",
|
|
message="is_retro must be a boolean.",
|
|
)
|
|
)
|
|
if line.is_retro is True and line.source_period is None:
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="MISSING_RETRO_SOURCE_PERIOD",
|
|
field=f"{prefix}.source_period",
|
|
message="Retro input must reference the affected period.",
|
|
)
|
|
)
|
|
if line.source_period is None:
|
|
return
|
|
|
|
source_period_valid = self._validate_period(
|
|
line.source_period, f"{prefix}.source_period", issues
|
|
)
|
|
if is_retro_valid and line.is_retro is not True:
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="UNEXPECTED_SOURCE_PERIOD",
|
|
field=f"{prefix}.source_period",
|
|
message="A source period requires is_retro=True.",
|
|
)
|
|
)
|
|
elif (
|
|
is_retro_valid
|
|
and line.is_retro is True
|
|
and source_period_valid
|
|
and period_is_valid
|
|
and line.source_period.end >= request.period.start
|
|
):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_RETRO_SOURCE_PERIOD",
|
|
field=f"{prefix}.source_period",
|
|
message="Retro source period must end before this period.",
|
|
)
|
|
)
|
|
|
|
def _validate_expected_totals(
|
|
self,
|
|
request: PayrollRequest,
|
|
rules: PayrollRules,
|
|
issues: list[PayrollIssue],
|
|
) -> None:
|
|
if request.expected_totals is None:
|
|
return
|
|
if not isinstance(request.expected_totals, ExpectedTotals):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_EXPECTED_TOTALS",
|
|
field="expected_totals",
|
|
message="Expected totals must be an ExpectedTotals instance.",
|
|
)
|
|
)
|
|
return
|
|
for field_name in ("point_amount", "gross", "recalculated_gross", "payable"):
|
|
allow_negative = field_name == "point_amount" or (
|
|
field_name == "payable" and rules.allow_negative_payable
|
|
)
|
|
self._validate_decimal(
|
|
getattr(request.expected_totals, field_name),
|
|
f"expected_totals.{field_name}",
|
|
issues,
|
|
minimum=None if allow_negative else Decimal("0"),
|
|
allow_none=True,
|
|
)
|
|
|
|
@staticmethod
|
|
def _is_nonempty_string(value: object) -> bool:
|
|
return isinstance(value, str) and bool(value.strip())
|
|
|
|
@staticmethod
|
|
def _validate_period(
|
|
value: object,
|
|
field_name: str,
|
|
issues: list[PayrollIssue],
|
|
) -> bool:
|
|
if not isinstance(value, PayrollPeriod):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_PERIOD",
|
|
field=field_name,
|
|
message="Period must be a PayrollPeriod instance.",
|
|
)
|
|
)
|
|
return False
|
|
if (
|
|
not isinstance(value.start, date)
|
|
or isinstance(value.start, datetime)
|
|
or not isinstance(value.end, date)
|
|
or isinstance(value.end, datetime)
|
|
):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_PERIOD",
|
|
field=field_name,
|
|
message="Period boundaries must be dates.",
|
|
)
|
|
)
|
|
return False
|
|
if value.start > value.end:
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_PERIOD",
|
|
field=field_name,
|
|
message="Period start must not be after period end.",
|
|
)
|
|
)
|
|
return False
|
|
return True
|
|
|
|
@staticmethod
|
|
def _validate_decimal(
|
|
value: object,
|
|
field_name: str,
|
|
issues: list[PayrollIssue],
|
|
*,
|
|
minimum: Decimal | None,
|
|
exclusive_minimum: bool = False,
|
|
allow_none: bool = False,
|
|
) -> None:
|
|
if value is None and allow_none:
|
|
return
|
|
if not isinstance(value, Decimal):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="INVALID_DECIMAL_TYPE",
|
|
field=field_name,
|
|
message="Monetary and quantity values must use Decimal.",
|
|
)
|
|
)
|
|
return
|
|
if not _is_finite_decimal(value):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="NON_FINITE_DECIMAL",
|
|
field=field_name,
|
|
message="NaN and infinite values are not supported.",
|
|
)
|
|
)
|
|
return
|
|
decimal_tuple = value.as_tuple()
|
|
fractional_places = max(-decimal_tuple.exponent, 0)
|
|
integer_digits = max(value.adjusted() + 1, 0) if value else 0
|
|
if (
|
|
len(decimal_tuple.digits)
|
|
> DeterministicPayrollCalculator.MAX_DECIMAL_DIGITS
|
|
or integer_digits > DeterministicPayrollCalculator.MAX_DECIMAL_DIGITS
|
|
or fractional_places > DeterministicPayrollCalculator.MAX_FRACTIONAL_PLACES
|
|
):
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="DECIMAL_OUT_OF_RANGE",
|
|
field=field_name,
|
|
message="Decimal precision or scale exceeds safe limits.",
|
|
)
|
|
)
|
|
return
|
|
below_minimum = minimum is not None and value < minimum
|
|
equals_exclusive_minimum = (
|
|
minimum is not None and exclusive_minimum and value == minimum
|
|
)
|
|
if below_minimum or equals_exclusive_minimum:
|
|
comparator = "greater than" if exclusive_minimum else "at least"
|
|
issues.append(
|
|
PayrollIssue(
|
|
code="NEGATIVE_OR_INVALID_AMOUNT",
|
|
field=field_name,
|
|
message=(f"Value must be {comparator} {decimal_text(minimum)}."),
|
|
)
|
|
)
|