PayrollManagementSystem/backend/tests/integration/test_receipt_confirmation.py

"""隔离MySQL/API验证独立确认、默认门槛、凭证及内容变更，不写真实工资。"""

import copy
from decimal import Decimal

import pytest
from sqlalchemy import func, select

from app.core.db import get_session_factory
from app.models import OperationLog, PayrollInput, PerformanceLedger, Receipt
from tests.integration.test_formal_payroll import (
    post,
)
from tests.integration.test_formal_payroll import (
    unconfirmed_formal_chain as unconfirmed_formal_chain,
)

pytestmark = pytest.mark.integration
ROOT = "/api/v1/business/receipts"


def confirm(client, headers, receipt):
    return client.post(f"{ROOT}/{receipt['id']}/confirm", headers=headers,
                       json={"expected_revision": receipt["confirmation_revision"]})


def calculate(client, headers, business):
    return client.post(f"/api/v1/payroll/businesses/{business['id']}/calculate", headers=headers,
                       json={"month": "2026-03"})


def switch_off(client, headers):
    post(client, "/api/v1/system/params/changes", headers, {
        "change_type": 1, "remark": "t_独立确认关闭场景",
        "items": [{"param_code": "receipt.confirmation.enabled", "dimension_value": "", "value": "0"}],
    })


def test_默认开启_保存资料和凭证资格不能代替人工确认(client, admin, unconfirmed_formal_chain):
    business, receipt, *_ = unconfirmed_formal_chain
    headers = admin["headers"]
    assert client.get(ROOT + "/confirmation-policy", headers=headers).json()["data"]["enabled"] is True
    assert receipt["confirmation_status"] == 0
    response = calculate(client, headers, business)
    assert response.status_code == 400 and "人工确认" in response.text
    with get_session_factory()() as s:
        assert s.get(Receipt, receipt["id"]).confirmation_status == 0
        assert s.scalar(select(func.count()).select_from(PerformanceLedger)) == 0


def test_独立确认后核算_重复确认幂等及审计(client, admin, unconfirmed_formal_chain):
    business, receipt, *_ = unconfirmed_formal_chain
    headers = admin["headers"]
    response = confirm(client, headers, receipt)
    assert response.status_code == 200, response.text
    confirmed = response.json()["data"]
    assert confirmed["confirmation_status"] == 1 and confirmed["confirmed_by"]
    again = confirm(client, headers, receipt).json()["data"]
    assert again["confirmed_at"] == confirmed["confirmed_at"]
    result = calculate(client, headers, business)
    assert result.status_code == 200, result.text
    assert result.json()["data"]["performance"] == "0.20"
    with get_session_factory()() as s:
        assert s.get(Receipt, receipt["id"]).settlement_status == 1
        assert s.scalar(select(func.count()).select_from(OperationLog).where(OperationLog.action == "人工确认")) >= 1


def test_关闭仅跳过人工确认_不产生确认状态(client, admin, unconfirmed_formal_chain):
    business, receipt, *_ = unconfirmed_formal_chain
    switch_off(client, admin["headers"])
    result = calculate(client, admin["headers"], business)
    assert result.status_code == 200, result.text
    with get_session_factory()() as s:
        assert s.get(Receipt, receipt["id"]).confirmation_status == 0


@pytest.mark.parametrize("manually_confirmed", [False, True])
def test_关闭开关或人工确认都不能豁免无效凭证(client, admin, unconfirmed_formal_chain, manually_confirmed):
    business, receipt, *_ = unconfirmed_formal_chain
    if manually_confirmed:
        assert confirm(client, admin["headers"], receipt).status_code == 200
    else:
        switch_off(client, admin["headers"])
    with get_session_factory()() as s:
        record = s.scalar(select(PayrollInput).where(PayrollInput.business_id == business["id"]))
        data = copy.deepcopy(record.data_json)
        data["receipts"][0]["voucher_confirmed"] = False
        record.data_json = data
        s.commit()
    result = calculate(client, admin["headers"], business)
    assert result.status_code == 400 and "VOUCHER_STATE_PENDING" in result.text


def test_修改关键内容重置确认_旧页面不能误确认(client, admin, unconfirmed_formal_chain):
    _, receipt, *_ = unconfirmed_formal_chain
    headers = admin["headers"]
    assert confirm(client, headers, receipt).status_code == 200
    changed = client.put(f"{ROOT}/{receipt['id']}", headers=headers,
                         json={"receipt_date": receipt["receipt_date"], "amount": "101"})
    assert changed.status_code == 200, changed.text
    data = changed.json()["data"]
    assert data["confirmation_status"] == 0 and data["confirmed_at"] is None
    assert confirm(client, headers, receipt).status_code == 409
    assert confirm(client, headers, data).status_code == 200


def test_只读拒绝确认及切换开关(client, admin, normal, unconfirmed_formal_chain):
    _, receipt, *_ = unconfirmed_formal_chain
    assert confirm(client, normal["headers"], receipt).status_code == 403
    response = client.post("/api/v1/system/params/changes", headers=normal["headers"], json={
        "change_type": 1, "items": [{"param_code": "receipt.confirmation.enabled", "value": "0"}],
    })
    assert response.status_code == 403


@pytest.mark.parametrize("unconfirmed_formal_chain,expected", [
    ({"unit_cost": "50"}, "0.00"), ({"unit_cost": "45"}, "-20.00"),
], indirect=["unconfirmed_formal_chain"])
def test_正式核算零毛利特例但低阈值仍为负(client, admin, unconfirmed_formal_chain, expected):
    business, receipt, *_ = unconfirmed_formal_chain
    assert confirm(client, admin["headers"], receipt).status_code == 200
    result = calculate(client, admin["headers"], business)
    assert result.status_code == 200, result.text
    with get_session_factory()() as s:
        ledger = s.scalar(select(PerformanceLedger).where(PerformanceLedger.business_id == business["id"]))
        assert ledger.converted_profit == Decimal(expected)
        assert ledger.performance_amount == Decimal(expected) * Decimal("0.01")


@pytest.mark.parametrize("role,profit,converted,performance", [
    ("salesperson", "0", "0", "0"),
    ("salesperson", "10", "-20", "-0.20"),
    ("info_officer", "0", "100", "0.10"),
])
def test_试算零毛利与信息员基数保持独立(client, admin, role, profit, converted, performance):
    data = post(client, "/api/v1/system/params/preview", admin["headers"], {
        "business_role": role, "customer_type": "personal", "product_category": "municipal_water",
        "ship_date": "2026-03-10", "receipt_date": "2026-03-10", "revenue": "100",
        "profit": profit, "received_amount": "100",
    })
    assert Decimal(data["converted_profit"]) == Decimal(converted)
    assert Decimal(data["performance_amount"]) == Decimal(performance)