PayrollManagementSystem/backend/tests/integration/test_requirement_completion.py

"""隔离 MySQL 验证需求补齐的页面接口、版本和工资原子提交。"""

from datetime import date, timedelta
from decimal import Decimal
from io import BytesIO

import pytest
from openpyxl import load_workbook
from sqlalchemy import func, select

from app.core.config import get_settings
from app.core.db import get_session_factory
from app.models import ParamChangeBatch, ProbationPayment
from tests.integration.test_formal_payroll import formal_chain as formal_chain

pytestmark = pytest.mark.integration


def test_系数表未来版本与工作簿往返(client, admin):
    headers = admin["headers"]
    base = "/api/v1/system/coefficients"
    data = client.get(base, headers=headers).json()["data"]
    assert len(data["rows"]) == 10
    rows = [
        {
            k: row.get(k)
            for k in ("product_category", "customer_type", "stage", "value", "slope", "boundary")
        }
        for row in data["rows"]
        if not row.get("readonly")
    ]
    rows[-1]["value"] = "5"
    payload = {
        "effective_date": str(date.today() + timedelta(days=2)),
        "base_date": str(date.today()),
        "fingerprint": data["fingerprint"],
        "rows": rows,
    }
    response = client.put(base, headers=headers, json=payload)
    assert response.status_code == 200, response.text
    assert client.put(base, headers=headers, json=payload).status_code == 400
    future = client.get(base, headers=headers, params={"as_of": payload["effective_date"]}).json()[
        "data"
    ]
    assert future["rows"][-2]["value"] == "5"
    export = client.get(base + "/export", headers=headers)
    assert export.status_code == 200
    preview = client.post(
        base + "/import-preview",
        headers=headers,
        files={
            "file": (
                "coefficients.xlsx",
                export.content,
                "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
            )
        },
    )
    assert preview.status_code == 200, preview.text
    assert preview.json()["data"]["can_apply"]
    book = load_workbook(BytesIO(export.content))
    book["系数参数"].cell(2, 4, "bad-value")
    output = BytesIO()
    book.save(output)
    with get_session_factory()() as session:
        before = session.scalar(select(func.count()).select_from(ParamChangeBatch))
    bad = client.post(
        base + "/import-preview", headers=headers, files={"file": ("bad.xlsx", output.getvalue())}
    )
    assert bad.status_code == 200, bad.text
    assert not bad.json()["data"]["can_apply"]
    with get_session_factory()() as session:
        assert session.scalar(select(func.count()).select_from(ParamChangeBatch)) == before


def test_业务提取产品及重复预览不重复建档(client, admin, formal_chain):
    headers = admin["headers"]
    business, _, _, _, _ = formal_chain
    detail = client.get(f"/api/v1/business/businesses/{business['id']}", headers=headers).json()[
        "data"
    ]
    ids = [row["id"] for row in detail["items"]]
    base = "/api/v1/base/products/extract"
    plan = client.post(base + "/preview", headers=headers, json={"item_ids": ids})
    assert plan.status_code == 200, plan.text
    data = plan.json()["data"]
    applied = client.post(
        base, headers=headers, json={"item_ids": ids, "fingerprint": data["fingerprint"]}
    )
    assert applied.status_code == 200, applied.text
    target = applied.json()["data"]["rows"][0]["target_id"]
    assert target
    again = client.post(base + "/preview", headers=headers, json={"item_ids": ids}).json()["data"]
    assert again["rows"][0]["action"] == "LINK"
    assert again["rows"][0]["target_id"] == target


def test_无绩效试用期工资及冻结分项导出(client, admin, monkeypatch):
    monkeypatch.setattr(get_settings(), "payroll_allow_legacy_calculation", False)
    headers = admin["headers"]
    roles = client.get("/api/v1/system/business-roles", headers=headers).json()["data"]
    if isinstance(roles, dict):
        roles = roles.get("list", [])
    role = next(row for row in roles if row["role_code"] == "salesperson")
    person = client.post(
        "/api/v1/base/personnel",
        headers=headers,
        json={
            "personnel_name": "t_无绩效基本工资",
            "business_role_id": role["id"],
            "salary_standard": "3000",
            "hire_date": "2026-01-01",
            "probation_start_date": "2026-01-01",
            "regularization_date": "2027-01-01",
            "is_probation": 1,
        },
    )
    assert person.status_code == 200, person.text
    generated = client.post(
        "/api/v1/settlements/generate", headers=headers, json={"month": "2026-11"}
    )
    assert generated.status_code == 200, generated.text
    item = next(
        row
        for row in generated.json()["data"]["items"]
        if row["personnel_id"] == person.json()["data"]["id"]
    )
    assert Decimal(item["net_amount"]) == Decimal("3000")
    assert Decimal(item["normal_payable_sum"]) == 0
    assert (
        client.post(
            "/api/v1/settlements/generate", headers=headers, json={"month": "2026-11"}
        ).status_code
        == 400
    )
    export = client.get("/api/v1/payroll/salary-export/2026-11", headers=headers)
    assert export.status_code == 200
    assert "工资分项" in load_workbook(BytesIO(export.content)).sheetnames
    with get_session_factory()() as session:
        assert session.scalar(select(func.count()).select_from(ProbationPayment)) == 0


def test_只读账号不能维护系数或执行提取(client, normal):
    headers = normal["headers"]
    response = client.post("/api/v1/base/products/extract", headers=headers, json={"item_ids": [1]})
    assert response.status_code == 403
    response = client.put("/api/v1/system/coefficients", headers=headers, json={})
    assert response.status_code == 403


def test_已生成工资的年度续付走当月调整且防重(client, admin, formal_chain, monkeypatch):
    from app.models import PerformanceLedger
    from app.services import payroll_months

    monkeypatch.setattr(get_settings(), "payroll_allow_legacy_calculation", False)
    monkeypatch.setattr(get_settings(), "payroll_approval_enabled", False)
    monkeypatch.setattr(payroll_months, "current_month", lambda: "2027-01")
    headers = admin["headers"]
    business, _, _, owner, _ = formal_chain
    response = client.post(
        f"/api/v1/payroll/businesses/{business['id']}/calculate",
        headers=headers,
        json={"month": "2026-03"},
    )
    assert response.status_code == 200, response.text
    # 隔离合成台账：年度计提20000、年底仅1000符合业务支付条件。
    with get_session_factory()() as session:
        row = session.scalar(
            select(PerformanceLedger).where(PerformanceLedger.business_id == business["id"])
        )
        row.is_probation = 1
        row.performance_amount = Decimal("20000")
        row.payable_amount = Decimal("1000")
        session.commit()
    december = client.post(
        "/api/v1/settlements/generate", headers=headers, json={"month": "2026-12"}
    )
    assert december.status_code == 200, december.text
    first = next(
        row for row in december.json()["data"]["items"] if row["personnel_id"] == owner["id"]
    )
    assert Decimal(first["probation_payment_sum"]) == Decimal("1000")
    january = client.post(
        "/api/v1/settlements/generate", headers=headers, json={"month": "2027-01"}
    )
    assert january.status_code == 200, january.text
    # 工资生成后确认额外满足支付条件的1000，原年度权益只剩1000。
    with get_session_factory()() as session:
        row = session.scalar(
            select(PerformanceLedger).where(
                PerformanceLedger.business_id == business["id"], PerformanceLedger.calc_type == 1
            )
        )
        row.payable_amount = Decimal("3000")
        session.commit()
    preview = client.get(
        "/api/v1/payroll/probation",
        headers=headers,
        params={"month": "2027-01", "personnel_id": owner["id"]},
    )
    assert preview.status_code == 200, preview.text
    balance = preview.json()["data"]["rows"][0]
    assert Decimal(balance["delta"]) == Decimal("1000")
    payload = {
        "personnel_id": owner["id"],
        "year": "2026",
        "fingerprint": balance["fingerprint"],
        "reason": "隔离样本：原年度后续支付条件满足",
    }
    adjustment = client.post("/api/v1/payroll/probation/adjustments", headers=headers, json=payload)
    assert adjustment.status_code == 200, adjustment.text
    assert Decimal(adjustment.json()["data"]["amount"]) == Decimal("1000")
    assert (
        client.post(
            "/api/v1/payroll/probation/adjustments", headers=headers, json=payload
        ).status_code
        == 400
    )
    final = client.get(
        "/api/v1/payroll/probation",
        headers=headers,
        params={"month": "2027-01", "personnel_id": owner["id"]},
    ).json()["data"]["rows"][0]
    assert Decimal(final["paid"]) == Decimal("2000")
    assert Decimal(final["delta"]) == 0