"""隔离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)