提交

backend/tests/test_market_payment_api.py

@@ -320,7 +320,7 @@ def test_payment_list_filters_progress_from_schedule_status_and_due_date_range()
     assert [item['id'] for item in paid] == ['sp-paid']
 
 
-def test_payment_progress_never_infers_cleared_from_amounts():
+def test_payment_progress_infers_cleared_when_confirmed_paid_covers_due():
     from app.services.market_payment import list_payments
 
     db = _session()
@@ -337,6 +337,13 @@ def test_payment_progress_never_infers_cleared_from_amounts():
         due_amount=100,
         schedule_status='cleared',
     )
+    _payment_tree(
+        db,
+        'partial-open',
+        due_amount=100,
+        schedule_status='open',
+        transfer_statuses=('yjf',),
+    )
 
     partial, partial_total = list_payments(
         db, ctx=_ctx(), user_info=_user(), progress_status='bfyj'
@@ -346,13 +353,18 @@ def test_payment_progress_never_infers_cleared_from_amounts():
     )
 
     assert partial_total == 1
-    assert partial[0]['id'] == 'sp-amount-covered-but-open'
+    assert partial[0]['id'] == 'sp-partial-open'
     assert partial[0]['paymentProgress'] == '部分已缴'
     assert partial[0]['clearedScheduleCount'] == 0
-    assert cleared_total == 1
-    assert cleared[0]['id'] == 'sp-status-cleared-without-transfer'
-    assert cleared[0]['paymentProgress'] == '已缴清'
-    assert cleared[0]['clearedScheduleCount'] == 1
+    assert cleared_total == 2
+    assert {item['id'] for item in cleared} == {
+        'sp-amount-covered-but-open',
+        'sp-status-cleared-without-transfer',
+    }
+    covered = next(item for item in cleared if item['id'] == 'sp-amount-covered-but-open')
+    assert covered['paymentProgress'] == '已缴清'
+    assert covered['clearedScheduleCount'] == 1
+    assert covered['schedules'][0]['scheduleStatus'] == 'cleared'
 
 
 def test_progress_filter_pages_in_sql_before_loading_children(monkeypatch):

backend/tests/test_oa_approver_dedup.py

@@ -0,0 +1,314 @@
+"""OA：审批人去重 — 同一实例内已同意的人，后续节点系统自动通过。"""
+import json
+import unittest
+from datetime import datetime
+from unittest.mock import patch
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+
+def _resolve_from_config(_db, *, assignee_config, **_kwargs):
+    ids = list(assignee_config.get("userIds") or [])
+    return ids, {"resolve_source": "users"}
+
+
+class TestApproverDedup(unittest.TestCase):
+    def setUp(self):
+        from app.models.models import OaAppVersion, OaInstance, OaTask
+
+        self.engine = create_engine("sqlite:///:memory:")
+        OaAppVersion.__table__.create(self.engine)
+        OaInstance.__table__.create(self.engine)
+        OaTask.__table__.create(self.engine)
+        Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)
+        self.db = Session()
+        self.OaAppVersion = OaAppVersion
+        self.OaInstance = OaInstance
+        self.OaTask = OaTask
+        self.now = datetime.now()
+
+    def tearDown(self):
+        self.db.close()
+        self.engine.dispose()
+
+    def _seed(self, *, flow, advanced, initiator="u0"):
+        self.db.add(
+            self.OaAppVersion(
+                id="v1",
+                app_id="app1",
+                version_no=1,
+                status="published",
+                is_latest_published="1",
+                form_schema_json="{}",
+                flow_tree_json=json.dumps(flow),
+                advanced_settings_json=json.dumps(advanced),
+                tenant_no="zkxw",
+                create_dept_id="",
+                create_time=self.now,
+                update_time=self.now,
+                del_flag=0,
+            )
+        )
+        self.db.add(
+            self.OaInstance(
+                id="i1",
+                app_id="app1",
+                app_version_id="v1",
+                title="t",
+                initiator_user_id=initiator,
+                initiator_dept_id="d1",
+                form_data_json="{}",
+                status="running",
+                tenant_no="zkxw",
+                create_dept_id="",
+                create_time=self.now,
+                update_time=self.now,
+                del_flag=0,
+            )
+        )
+        self.db.commit()
+
+    def _load(self):
+        inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
+        ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
+        return inst, ver
+
+    def _tasks(self):
+        self.db.flush()
+        return self.db.query(self.OaTask).filter_by(instance_id="i1").all()
+
+    def _pending(self, user_id):
+        return next(
+            t
+            for t in self._tasks()
+            if t.assignee_user_id == user_id and t.status == "pending"
+        )
+
+    def _approve(self, task):
+        from app.services.oa_engine import _approve_task_core
+
+        inst, ver = self._load()
+        _approve_task_core(
+            self.db,
+            task=task,
+            inst=inst,
+            ver=ver,
+            user_id=task.assignee_user_id,
+            comment="同意",
+        )
+
+    def _chain_flow(self):
+        """第 1 步 A，第 2 步 B，第 3、4 步再是 A。"""
+
+        def approval(node_id, name, user_ids, child):
+            return {
+                "id": node_id,
+                "type": "approval",
+                "name": name,
+                "approverConfig": {
+                    "assigneeType": "users",
+                    "userIds": user_ids,
+                    "signMode": "or",
+                },
+                "child": child,
+            }
+
+        return {
+            "id": "s",
+            "type": "start",
+            "child": approval(
+                "a1",
+                "第一步",
+                ["A"],
+                approval(
+                    "a2",
+                    "第二步",
+                    ["B"],
+                    approval(
+                        "a3",
+                        "第三步",
+                        ["A"],
+                        approval("a4", "第四步", ["A"], {"id": "e1", "type": "end"}),
+                    ),
+                ),
+            ),
+        }
+
+    def test_later_nodes_auto_pass_after_same_person_approved(self):
+        """A 在第 1 步同意后，第 3、4 步不再待办，系统记为自动通过。"""
+        from app.services.oa_engine import advance_from
+
+        self._seed(flow=self._chain_flow(), advanced={"approverDedup": True})
+        inst, ver = self._load()
+        with patch(
+            "app.services.oa_engine.resolve_assignees", side_effect=_resolve_from_config
+        ), patch("app.services.oa_engine._next_task_order_num", return_value=1):
+            advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
+            self._approve(self._pending("A"))
+            self._approve(self._pending("B"))
+
+        by_node = {t.node_key: t for t in self._tasks()}
+        self.assertEqual(by_node["a1"].status, "approved")
+        self.assertEqual(by_node["a1"].comment_text, "同意")
+        self.assertEqual(by_node["a3"].status, "approved")
+        self.assertEqual(by_node["a3"].comment_text, "审批人去重，系统自动通过")
+        self.assertEqual(by_node["a4"].status, "approved")
+        self.assertEqual(by_node["a4"].comment_text, "审批人去重，系统自动通过")
+        self.assertEqual(inst.status, "completed")
+        self.assertFalse(any(t.status == "pending" for t in by_node.values()))
+
+    def test_switch_off_keeps_repeat_approver_pending(self):
+        from app.services.oa_engine import advance_from
+
+        self._seed(flow=self._chain_flow(), advanced={"approverDedup": False})
+        inst, ver = self._load()
+        with patch(
+            "app.services.oa_engine.resolve_assignees", side_effect=_resolve_from_config
+        ), patch("app.services.oa_engine._next_task_order_num", return_value=1):
+            advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
+            self._approve(self._pending("A"))
+            self._approve(self._pending("B"))
+
+        step3 = next(t for t in self._tasks() if t.node_key == "a3")
+        self.assertEqual(step3.assignee_user_id, "A")
+        self.assertEqual(step3.status, "pending")
+        self.assertEqual(inst.status, "running")
+        self.assertFalse(any(t.node_key == "a4" for t in self._tasks()))
+
+    def test_or_sign_auto_pass_closes_whole_node(self):
+        from app.services.oa_engine import advance_from
+
+        flow = {
+            "id": "s",
+            "type": "start",
+            "child": {
+                "id": "a1",
+                "type": "approval",
+                "name": "第一步",
+                "approverConfig": {
+                    "assigneeType": "users",
+                    "userIds": ["A"],
+                    "signMode": "or",
+                },
+                "child": {
+                    "id": "a2",
+                    "type": "approval",
+                    "name": "或签",
+                    "approverConfig": {
+                        "assigneeType": "users",
+                        "userIds": ["A", "C"],
+                        "signMode": "or",
+                    },
+                    "child": {"id": "e1", "type": "end"},
+                },
+            },
+        }
+        self._seed(flow=flow, advanced={"approverDedup": True})
+        inst, ver = self._load()
+        with patch(
+            "app.services.oa_engine.resolve_assignees", side_effect=_resolve_from_config
+        ), patch("app.services.oa_engine._next_task_order_num", return_value=1):
+            advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
+            self._approve(self._pending("A"))
+
+        by_key = {(t.node_key, t.assignee_user_id): t for t in self._tasks()}
+        self.assertEqual(by_key[("a2", "A")].status, "approved")
+        self.assertEqual(by_key[("a2", "A")].comment_text, "审批人去重，系统自动通过")
+        self.assertEqual(by_key[("a2", "C")].status, "done")
+        self.assertEqual(inst.status, "completed")
+
+    def test_and_sign_only_auto_passes_repeat_person(self):
+        from app.services.oa_engine import advance_from
+
+        flow = {
+            "id": "s",
+            "type": "start",
+            "child": {
+                "id": "a1",
+                "type": "approval",
+                "name": "第一步",
+                "approverConfig": {
+                    "assigneeType": "users",
+                    "userIds": ["A"],
+                    "signMode": "or",
+                },
+                "child": {
+                    "id": "a2",
+                    "type": "approval",
+                    "name": "会签",
+                    "approverConfig": {
+                        "assigneeType": "users",
+                        "userIds": ["A", "C"],
+                        "signMode": "and",
+                    },
+                    "child": {"id": "e1", "type": "end"},
+                },
+            },
+        }
+        self._seed(flow=flow, advanced={"approverDedup": True})
+        inst, ver = self._load()
+        with patch(
+            "app.services.oa_engine.resolve_assignees", side_effect=_resolve_from_config
+        ), patch("app.services.oa_engine._next_task_order_num", return_value=1):
+            advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
+            self._approve(self._pending("A"))
+
+        by_user = {
+            t.assignee_user_id: t for t in self._tasks() if t.node_key == "a2"
+        }
+        self.assertEqual(by_user["A"].status, "approved")
+        self.assertEqual(by_user["A"].comment_text, "审批人去重，系统自动通过")
+        self.assertEqual(by_user["C"].status, "pending")
+        self.assertEqual(inst.status, "running")
+
+    def test_non_approved_history_does_not_dedup(self):
+        """仅统计已同意；拒绝等状态不触发后续自动通过。"""
+        from app.services.oa_engine import advance_from
+
+        flow = {
+            "id": "s",
+            "type": "start",
+            "child": {
+                "id": "a1",
+                "type": "approval",
+                "name": "当前",
+                "approverConfig": {
+                    "assigneeType": "users",
+                    "userIds": ["A"],
+                    "signMode": "or",
+                },
+                "child": {"id": "e1", "type": "end"},
+            },
+        }
+        self._seed(flow=flow, advanced={"approverDedup": True})
+        self.db.add(
+            self.OaTask(
+                id="old-reject",
+                instance_id="i1",
+                node_key="old",
+                node_name="历史",
+                task_type="approval",
+                assignee_user_id="A",
+                status="rejected",
+                sign_mode="or",
+                round_no=1,
+                is_read=0,
+                tenant_no="zkxw",
+                order_num=0,
+                create_dept_id="",
+                create_time=self.now,
+                update_time=self.now,
+                del_flag=0,
+            )
+        )
+        self.db.commit()
+        inst, ver = self._load()
+        with patch(
+            "app.services.oa_engine.resolve_assignees", side_effect=_resolve_from_config
+        ), patch("app.services.oa_engine._next_task_order_num", return_value=1):
+            advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
+
+        current = next(t for t in self._tasks() if t.node_key == "a1")
+        self.assertEqual(current.status, "pending")
+        self.assertEqual(inst.status, "running")

backend/tests/test_oa_instance_comment.py

@@ -257,24 +257,32 @@ class TestCommentCrud(unittest.TestCase):
         )
         self.assertEqual(again["status"], "revoked")
 
-    def test_cannot_comment_when_not_running(self):
+    def test_can_comment_and_revoke_when_completed(self):
         from app.services import oa_instance_comment as svc
-        from app.services.oa_engine import OaEngineError
         from app.models.models import OaInstance
 
         inst = self.db.query(OaInstance).filter(OaInstance.id == "i1").one()
         inst.status = "completed"
         self.db.commit()
-        with self.assertRaises(OaEngineError):
-            svc.create_comment(
-                self.db,
-                instance_id="i1",
-                tenant_no="zkxw",
-                user_id="u1",
-                dept_id="d1",
-                content="x",
-                attachment_refs=None,
-            )
+        row = svc.create_comment(
+            self.db,
+            instance_id="i1",
+            tenant_no="zkxw",
+            user_id="u1",
+            dept_id="d1",
+            content="结束后仍可评论",
+            attachment_refs=None,
+        )
+        self.assertEqual(row["status"], "normal")
+        self.assertTrue(row["canRevoke"])
+        revoked = svc.revoke_comment(
+            self.db,
+            instance_id="i1",
+            comment_id=row["id"],
+            tenant_no="zkxw",
+            user_id="u1",
+        )
+        self.assertEqual(revoked["status"], "revoked")
 
 
 class TestCommentMentionGrant(unittest.TestCase):

backend/tests/test_payment_finance_status_regression.py

@@ -0,0 +1,45 @@
+"""两类缴费的财务确认/退回成功路径与重复操作拒绝，纯内存数据库。"""
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from app.crud import presign_order, protocol_sample_payment
+from app.models.base import Base
+from app.models.models import (
+    BizCollectionProtocol, BizPresignOrder, BizPresignTransfer,
+    BizProtocolSamplePayment, BizProtocolPaymentSchedule, BizProtocolPaymentTransfer,
+)
+
+
+@pytest.mark.parametrize('kind', ['presign', 'sample'])
+@pytest.mark.parametrize('operation,expected', [('confirm', 'yjf'), ('reject', 'dfj')])
+def test_finance_transition_succeeds_once_and_rejects_duplicate(kind, operation, expected):
+    engine = create_engine('sqlite:///:memory:')
+    Base.metadata.create_all(engine)
+    with sessionmaker(bind=engine)() as db:
+        common = dict(tenant_no='tenant-a', create_dept_id='', del_flag=0)
+        if kind == 'presign':
+            db.add(BizPresignOrder(id='p', order_status='active', protocol_no='P', **common))
+            db.add(BizPresignTransfer(id='t', presign_order_id='p', pay_status='jfqrz', transfer_amount=100, **common))
+            crud, model = presign_order, BizPresignTransfer
+        else:
+            db.add_all([
+                BizCollectionProtocol(id='p', protocol_no='P', **common),
+                BizProtocolSamplePayment(id='pay', collection_protocol_id='p', sample_code='S', pay_mode='dj', contract_amount=100, **common),
+                BizProtocolPaymentSchedule(id='sch', sample_payment_id='pay', collection_protocol_id='p', sample_code='S', schedule_no=1, due_amount=100, schedule_status='open', **common),
+                BizProtocolPaymentTransfer(id='t', schedule_id='sch', sample_payment_id='pay', collection_protocol_id='p', sample_code='S', transfer_amount=100, pay_status='jfqrz', **common),
+            ])
+            crud, model = protocol_sample_payment, BizProtocolPaymentTransfer
+        db.commit()
+        function = crud.confirm_receipt if operation == 'confirm' else crud.reject_receipt
+        result = function(db, transfer_id='t', tenant_no='tenant-a', user_id='finance')
+        assert result.pay_status == expected
+        with pytest.raises(ValueError):
+            function(db, transfer_id='t', tenant_no='tenant-a', user_id='finance')
+        assert db.query(model).filter_by(id='t').one().pay_status == expected
+        if kind == 'sample' and operation == 'confirm':
+            assert (
+                db.query(BizProtocolPaymentSchedule).filter_by(id='sch').one().schedule_status
+                == 'cleared'
+            )
+    engine.dispose()

backend/tests/test_presign_order_date_validation.py

@@ -0,0 +1,19 @@
+"""预签约日期不应到数据库 CHAR(8) 字段才失败。"""
+import pytest
+from pydantic import ValidationError
+
+from app.api.presign_order import PresignOrderBody
+
+
+@pytest.mark.parametrize('field', ['orderDate', 'expectedDeliveryDate'])
+@pytest.mark.parametrize('value', ['2026-10-15', '2026101', '20261301', '20260230', 'abcdefgh'])
+def test_invalid_presign_date_rejected_before_database(field, value):
+    with pytest.raises(ValidationError):
+        PresignOrderBody(**{field: value})
+
+
+@pytest.mark.parametrize('field', ['orderDate', 'expectedDeliveryDate'])
+@pytest.mark.parametrize('value', ['20261015', '20240229', '', None])
+def test_valid_or_empty_presign_date_keeps_existing_contract(field, value):
+    body = PresignOrderBody(**{field: value})
+    assert body.model_dump(by_alias=True)[field] == value

backend/tests/test_protocol_status_change_history.py

@@ -0,0 +1,71 @@
+"""状态变更历史：同类型追加不覆盖。"""
+from datetime import datetime
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+
+def _session():
+    from app.models.models import BizCollectionProtocol, BizProtocolStatusChange, SysUser
+
+    engine = create_engine("sqlite:///:memory:")
+    BizCollectionProtocol.__table__.create(engine)
+    BizProtocolStatusChange.__table__.create(engine)
+    SysUser.__table__.create(engine)
+    return sessionmaker(bind=engine, autoflush=False, autocommit=False)(), engine
+
+
+def test_same_type_status_change_appends_instead_of_overwrite():
+    from app.crud.collection_protocol import (
+        list_protocol_status_changes,
+        update_collection_protocol,
+    )
+    from app.models.models import BizCollectionProtocol
+
+    db, engine = _session()
+    try:
+        now = datetime(2026, 3, 1, 10, 0, 0)
+        db.add(
+            BizCollectionProtocol(
+                id="p1",
+                protocol_no="XY0001",
+                protocol_status="sxz",
+                tenant_no="zkxw",
+                create_dept_id="",
+                create_time=now,
+                update_time=now,
+                del_flag=0,
+            )
+        )
+        db.commit()
+
+        update_collection_protocol(
+            db,
+            "p1",
+            tenant_no="zkxw",
+            user_id="u1",
+            protocol_status="ygq",
+            expire_time=datetime(2026, 6, 1, 12, 0, 0),
+            expire_desc="第一次过期",
+        )
+        update_collection_protocol(
+            db,
+            "p1",
+            tenant_no="zkxw",
+            user_id="u2",
+            protocol_status="ygq",
+            expire_time=datetime(2026, 9, 15, 18, 0, 0),
+            expire_desc="第二次过期",
+        )
+
+        row = db.query(BizCollectionProtocol).filter(BizCollectionProtocol.id == "p1").one()
+        assert row.expire_desc == "第二次过期"
+        items = list_protocol_status_changes(db, collection_protocol_id="p1", tenant_no="zkxw")
+        assert [item["changeDesc"] for item in items] == ["第一次过期", "第二次过期"]
+        assert [item["changeType"] for item in items] == ["expire", "expire"]
+        assert items[0]["fromStatus"] == "sxz"
+        assert items[1]["fromStatus"] == "ygq"
+        assert items[1]["toStatus"] == "ygq"
+    finally:
+        db.close()
+        engine.dispose()

backend/tests/test_protocol_status_transition_message.py

@@ -0,0 +1,23 @@
+import pytest
+
+from app.core.protocol_status_display import (
+    assert_protocol_status_transition,
+    status_transition_denied_message,
+)
+
+
+def test_expired_protocol_cannot_void_and_message_uses_chinese_labels():
+    assert status_transition_denied_message("ygq", "yzf") == "状态已过期不可转为已作废"
+    assert status_transition_denied_message("ygq", "yzf", kind="donation") == "状态已过期不可转为已作废"
+    with pytest.raises(ValueError, match="状态已过期不可转为已作废"):
+        assert_protocol_status_transition("ygq", "yzf", kind="protocol")
+    with pytest.raises(ValueError, match="状态已过期不可转为已作废"):
+        assert_protocol_status_transition("ygq", "yzf", kind="donation")
+
+
+def test_active_protocol_can_void_and_same_status_is_allowed():
+    assert_protocol_status_transition("sxz", "yzf", kind="protocol")
+    assert_protocol_status_transition("sxz", "zky", kind="protocol")
+    assert_protocol_status_transition("sxz", "zckc", kind="donation")
+    assert_protocol_status_transition("ygq", "ygq", kind="protocol")
+    assert_protocol_status_transition("ygq", "ygq", kind="donation")

backend/tests/test_protocol_v2_payment_mode_summary.py

@@ -0,0 +1,33 @@
+"""协议列表可显式读取 V2 样本方案，其他费用摘要消费者保持原行为。"""
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from app.models.base import Base
+from app.models.models import BizProtocolSamplePayment
+from app.services.qc_work_order_fee_summary import build_protocol_payment_summaries
+
+
+@pytest.mark.parametrize('modes,expected,label', [
+    (['dj'], 'dj', '趸交'),
+    (['fq', 'fq'], 'fq', '分期'),
+    (['fgd'], 'fgd', '分改趸'),
+    (['fq', 'dj'], 'djfq', '趸交+分期'),
+    (['fgd', 'fq'], 'fq+fgd', '分期+分改趸'),
+])
+def test_protocol_list_uses_active_tenant_v2_payment_modes(modes, expected, label):
+    engine = create_engine('sqlite:///:memory:')
+    Base.metadata.create_all(engine)
+    with sessionmaker(bind=engine)() as db:
+        for i, mode in enumerate(modes):
+            db.add(BizProtocolSamplePayment(id=str(i), collection_protocol_id='p1', sample_code=str(i), pay_mode=mode, tenant_no='tenant-a', del_flag=0))
+        db.add_all([
+            BizProtocolSamplePayment(id='deleted', collection_protocol_id='p1', sample_code='deleted', pay_mode='djfq', tenant_no='tenant-a', del_flag=1),
+            BizProtocolSamplePayment(id='foreign', collection_protocol_id='p1', sample_code='foreign', pay_mode='djfq', tenant_no='tenant-b', del_flag=0),
+        ])
+        db.commit()
+        result = build_protocol_payment_summaries(db, ['p1'], 'tenant-a', include_sample_payment_modes=True)
+        assert result['p1']['payMode'] == expected
+        assert result['p1']['payModeLabel'] == label
+        assert build_protocol_payment_summaries(db, ['p1'], 'tenant-a') == {}
+    engine.dispose()

backend/tests/test_sample_payment_progress.py

@@ -67,6 +67,43 @@ class TestSamplePaymentProgress(unittest.TestCase):
         self.assertEqual(r["activeScheduleCount"], 2)
         self.assertTrue(r["canConfirm"])
 
+    def test_open_schedule_is_cleared_when_confirmed_paid_covers_due(self):
+        """财务已确认实缴覆盖应缴时，即使批次仍是 open，也按已缴清计。"""
+        r = compute_sample_payment_progress(
+            schedules=[
+                {
+                    "id": "s1",
+                    "scheduleStatus": "open",
+                    "dueAmount": Decimal("22880.00"),
+                },
+            ],
+            transfers=[
+                {
+                    "scheduleId": "s1",
+                    "payStatus": "yjf",
+                    "transferAmount": Decimal("22880.00"),
+                },
+            ],
+        )
+        self.assertEqual(r["progressStatus"], "yjq")
+        self.assertEqual(r["clearedScheduleCount"], 1)
+        self.assertEqual(r["activeScheduleCount"], 1)
+        self.assertEqual(r["dueRemainAmount"], Decimal("0"))
+        self.assertFalse(r["canConfirm"])
+
+    def test_partial_confirmed_on_open_schedule_stays_partial(self):
+        r = compute_sample_payment_progress(
+            schedules=[
+                {"id": "s1", "scheduleStatus": "open", "dueAmount": Decimal("100")},
+            ],
+            transfers=[
+                {"scheduleId": "s1", "payStatus": "yjf", "transferAmount": Decimal("20")},
+            ],
+        )
+        self.assertEqual(r["progressStatus"], "bfyj")
+        self.assertEqual(r["clearedScheduleCount"], 0)
+        self.assertEqual(r["activeScheduleCount"], 1)
+
     def test_all_cleared(self):
         r = compute_sample_payment_progress(
             schedules=[

backend/tests/test_web_freight_permissions.py

@@ -0,0 +1,137 @@
+"""Web 运费数据权限与写入边界，使用内存数据库避免改动业务数据。"""
+import asyncio
+from datetime import datetime
+from decimal import Decimal
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from app.api import sample_freight as api
+from app.crud.sample_freight import _parse_amount
+from app.models.base import Base
+from app.models.models import (
+    BizCollectionProtocol, BizProtocolSample, BizSampleFreight,
+    BizSampleFreightItem, SysDictData, SysDictType, SysRole,
+    SysRoleDataScopeDetail, SysUser, SysUserRole,
+)
+
+USER = {'user_id': 'u1', 'tenant_no': 'tenant-a', 'role': 'admin', 'dept_id': 'dept-a'}
+
+
+@pytest.fixture
+def db():
+    engine = create_engine('sqlite:///:memory:')
+    Base.metadata.create_all(engine)
+    session = sessionmaker(bind=engine)()
+    now = datetime.now()
+    common = dict(tenant_no='tenant-a', create_dept_id='dept-a', create_time=now, del_flag=0)
+    session.add_all([
+        SysUser(id='u1', user_no='u1', nick_name='用户', password='', dept_id='dept-a', **common),
+        SysRole(id='self', role_name='普通人员', data_scope='4', status='1', **common),
+        SysUserRole(id='ur', user_id='u1', role_id='self', **common),
+        SysDictType(id='dt', dict_name='运输方式', dict_type='sample_transport_mode', **common),
+        SysDictData(id='dd', dict_label='顺丰', dict_value='sf', dict_type='sample_transport_mode', **common),
+    ])
+    for owner in ('u1', 'u2'):
+        session.add_all([
+            BizCollectionProtocol(id='p-' + owner, protocol_no='P-' + owner, create_by=owner, **common),
+            BizProtocolSample(id='s-' + owner, collection_protocol_id='p-' + owner, sample_code='S-' + owner, create_by=owner, **common),
+            BizSampleFreight(id='f-' + owner, freight_no='F-' + owner, transport_mode='sf', freight_amount=Decimal('10'), create_by=owner, **common),
+        ])
+    session.commit()
+    yield session
+    session.close()
+    engine.dispose()
+
+
+def call(endpoint, db, **kwargs):
+    return asyncio.run(endpoint(db=db, user=USER, **kwargs))
+
+
+def test_list_obeys_self_scope(db):
+    result = call(api.list_sample_freights, db, page=1, pageSize=10, includeItems=False)
+    assert [r['id'] for r in result['data']['list']] == ['f-u1']
+    assert result['data']['total'] == 1
+
+
+def test_multiple_roles_union_preserves_allowed_records(db):
+    common = dict(tenant_no='tenant-a', create_dept_id='', del_flag=0)
+    db.add_all([
+        SysRole(id='custom', role_name='扩展权限', data_scope='5', status='1', **common),
+        SysUserRole(id='ur2', user_id='u1', role_id='custom', **common),
+        SysRoleDataScopeDetail(id='detail', role_id='custom', scope_type='user', scope_value='u2', **common),
+    ])
+    db.commit()
+    result = call(api.list_sample_freights, db, page=1, pageSize=10, includeItems=False)
+    assert {r['id'] for r in result['data']['list']} == {'f-u1', 'f-u2'}
+
+
+@pytest.mark.parametrize('operation', ['detail', 'update', 'delete', 'link', 'unlink'])
+def test_outside_scope_batch_is_not_addressable(db, operation):
+    endpoints = {
+        'detail': (api.get_sample_freight, {}),
+        'update': (api.update_sample_freight, {'body': api.SampleFreightUpdateBody(freightAmount=999)}),
+        'delete': (api.delete_sample_freight, {}),
+        'link': (api.link_sample_freight, {'body': api.SampleFreightItemsBody(items=[{'sampleSource': 'cell_storage', 'sampleId': 's-u1'}])}),
+        'unlink': (api.unlink_sample_freight, {'body': api.SampleFreightItemsBody(items=[])}),
+    }
+    endpoint, kwargs = endpoints[operation]
+    result = call(endpoint, db, freight_id='f-u2', **kwargs)
+    assert result['code'] in (403, 404)
+    obj = db.query(BizSampleFreight).filter_by(id='f-u2').one()
+    assert obj.del_flag == 0
+    assert obj.freight_amount == Decimal('10')
+
+
+def test_sample_options_obey_parent_scope(db):
+    result = call(api.list_sample_freight_options, db, page=1, pageSize=20, sampleSource='cell_storage')
+    assert [r['sampleId'] for r in result['data']['list']] == ['s-u1']
+
+
+@pytest.mark.parametrize('sample_id', ['s-u2', 'does-not-exist'])
+@pytest.mark.parametrize('operation', ['create', 'link'])
+def test_cannot_link_hidden_or_missing_sample(db, sample_id, operation):
+    items = [{'sampleSource': 'cell_storage', 'sampleId': sample_id}]
+    if operation == 'create':
+        result = call(api.create_sample_freight, db, body=api.SampleFreightCreateBody(transportMode='sf', freightAmount=10, items=items))
+    else:
+        result = call(api.link_sample_freight, db, freight_id='f-u1', body=api.SampleFreightItemsBody(items=items))
+    assert result['code'] in (400, 403, 404)
+    assert db.query(BizSampleFreightItem).count() == 0
+    assert db.query(BizSampleFreight).count() == 2
+
+
+def test_created_batch_has_server_department_and_sample_snapshot(db):
+    result = call(api.create_sample_freight, db, body=api.SampleFreightCreateBody(
+        transportMode='sf', freightAmount=10,
+        items=[{'sampleSource': 'cell_storage', 'sampleId': 's-u1', 'sampleCode': 'forged', 'protocolId': 'p-u2'}],
+    ))
+    assert result['code'] == 200
+    row = db.query(BizSampleFreight).filter_by(id=result['data']['id']).one()
+    assert row.create_dept_id == 'dept-a'
+    assert result['data']['items'][0]['sampleCode'] == 'S-u1'
+    assert result['data']['items'][0]['protocolId'] == 'p-u1'
+
+
+@pytest.mark.parametrize('amount', ['NaN', 'Infinity', '-Infinity'])
+def test_nonfinite_freight_amount_is_validation_error(amount):
+    with pytest.raises(ValueError):
+        _parse_amount(amount)
+
+
+@pytest.mark.parametrize('operation', ['create', 'link'])
+def test_link_cannot_remove_sample_from_an_inaccessible_batch(db, operation):
+    db.add(BizSampleFreightItem(
+        id='existing-link', freight_id='f-u2', sample_source='cell_storage',
+        sample_id='s-u1', sample_code='S-u1', protocol_id='p-u1',
+        tenant_no='tenant-a', create_dept_id='', del_flag=0,
+    ))
+    db.commit()
+    items = [{'sampleSource': 'cell_storage', 'sampleId': 's-u1'}]
+    if operation == 'create':
+        result = call(api.create_sample_freight, db, body=api.SampleFreightCreateBody(transportMode='sf', freightAmount=10, items=items))
+    else:
+        result = call(api.link_sample_freight, db, freight_id='f-u1', body=api.SampleFreightItemsBody(items=items))
+    assert result['code'] in (403, 404)
+    assert db.query(BizSampleFreightItem).filter_by(id='existing-link').one().freight_id == 'f-u2'

backend/tests/test_web_presign_permissions.py

@@ -0,0 +1,129 @@
+"""预签约 Web 入口使用真实角色范围查询，在越权时不得调用任何读写服务。"""
+import asyncio
+from datetime import datetime
+from unittest.mock import Mock
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from app.api import presign_order as api
+from app.models.base import Base
+from app.models.models import (
+    BizCollectionProtocol, BizPresignOrder, BizPresignTransfer, SysRole, SysRoleDataScopeDetail,
+    SysUser, SysUserRole,
+)
+
+USER = {'user_id': 'u1', 'role': 'admin', 'tenant_no': 'tenant-a', 'dept_id': 'dept-a'}
+
+
+@pytest.fixture
+def db():
+    engine = create_engine('sqlite:///:memory:')
+    Base.metadata.create_all(engine)
+    session = sessionmaker(bind=engine)()
+    common = dict(tenant_no='tenant-a', create_dept_id='dept-a', create_time=datetime.now(), del_flag=0)
+    session.add_all([
+        SysUser(id='u1', user_no='u1', nick_name='用户', password='', dept_id='dept-a', **common),
+        SysRole(id='self', role_name='普通人员', data_scope='4', status='1', **common),
+        SysUserRole(id='ur', user_id='u1', role_id='self', **common),
+    ])
+    for owner in ('u1', 'u2'):
+        session.add_all([
+            BizCollectionProtocol(id='cp-' + owner, protocol_no='CP-' + owner, create_by=owner, **common),
+            BizPresignOrder(id='p-' + owner, protocol_no='P-' + owner, order_status='active', create_by=owner, **common),
+            BizPresignTransfer(id='t-' + owner, presign_order_id='p-' + owner, pay_status='jfqrz', transfer_amount=10, create_by=owner, **common),
+        ])
+    session.commit()
+    yield session
+    session.close()
+    engine.dispose()
+
+
+@pytest.mark.parametrize('action', [
+    'get', 'update', 'delete', 'convert', 'create_transfer',
+    'update_transfer', 'submit', 'withdraw', 'confirm', 'reject',
+])
+@pytest.mark.parametrize('access', ['denied', 'self', 'union'])
+def test_presign_endpoint_enforces_union_data_scope(db, monkeypatch, action, access):
+    if access == 'union':
+        common = dict(tenant_no='tenant-a', create_dept_id='', del_flag=0)
+        db.add_all([
+            SysRole(id='custom', role_name='自定义', data_scope='5', status='1', **common),
+            SysUserRole(id='ur2', user_id='u1', role_id='custom', **common),
+            SysRoleDataScopeDetail(id='detail', role_id='custom', scope_type='user', scope_value='u2', **common),
+        ])
+        db.commit()
+    owner = 'u1' if access == 'self' else 'u2'
+    cases = {
+        'get': (api.get_presign_order, 'get_presign_order', {'order_id': 'p-' + owner}),
+        'update': (api.update_presign_order, 'update_presign_order', {'order_id': 'p-' + owner, 'body': api.PresignOrderBody(remark='test')}),
+        'delete': (api.delete_presign_order, 'soft_delete_presign_order', {'order_id': 'p-' + owner}),
+        'convert': (api.convert_presign_order, 'convert_to_protocol', {'order_id': 'p-' + owner}),
+        'create_transfer': (api.create_presign_transfer, 'add_transfer', {'order_id': 'p-' + owner, 'body': api.PresignTransferBody(transferAmount=10)}),
+        'update_transfer': (api.update_presign_transfer, 'update_transfer_draft', {'transfer_id': 't-' + owner, 'body': api.PresignTransferBody(transferAmount=10)}),
+        'submit': (api.submit_presign_transfer_finance, 'submit_receipt_finance', {'transfer_id': 't-' + owner}),
+        'withdraw': (api.withdraw_presign_transfer, 'withdraw_receipt', {'transfer_id': 't-' + owner}),
+        'confirm': (api.confirm_presign_transfer, 'confirm_receipt', {'transfer_id': 't-' + owner, 'body': None}),
+        'reject': (api.reject_presign_transfer, 'reject_receipt', {'transfer_id': 't-' + owner}),
+    }
+    endpoint, service_name, kwargs = cases[action]
+    service = Mock(return_value={'id': 'result'})
+    monkeypatch.setattr(api.po_svc, service_name, service)
+    monkeypatch.setattr(api.po_svc, 'order_to_dict', lambda *args, **kwargs: {'id': 'result'})
+    monkeypatch.setattr(api.po_svc, '_transfer_to_dict', lambda *args, **kwargs: {'id': 'result'})
+    result = asyncio.run(endpoint(db=db, user=USER, **kwargs))
+    if access == 'denied':
+        assert result['code'] == 403
+        service.assert_not_called()
+    else:
+        assert result['code'] == 200
+        service.assert_called_once()
+
+
+def test_collection_protocol_audit_rejects_outside_scope(db, monkeypatch):
+    from app.api import collection_protocol
+
+    audit = Mock()
+    monkeypatch.setattr(collection_protocol.cp_svc, 'create_audit', audit)
+    result = asyncio.run(collection_protocol.submit_collection_protocol_audit(
+        protocol_id='cp-u2',
+        body=collection_protocol.AuditCreateBody(auditStatus='approved'),
+        db=db, user=USER,
+    ))
+    assert result['code'] == 403
+    audit.assert_not_called()
+
+
+@pytest.mark.parametrize('action', ['create', 'update', 'transfer', 'convert'])
+def test_web_presign_writes_use_server_department(db, monkeypatch, action):
+    cases = {
+        'create': (api.create_presign_order, 'create_presign_order', {'body': api.PresignOrderBody(motherName='客户')}),
+        'update': (api.update_presign_order, 'update_presign_order', {'order_id': 'p-u1', 'body': api.PresignOrderBody(remark='test')}),
+        'transfer': (api.create_presign_transfer, 'add_transfer', {'order_id': 'p-u1', 'body': api.PresignTransferBody(transferAmount=10)}),
+        'convert': (api.convert_presign_order, 'convert_to_protocol', {'order_id': 'p-u1'}),
+    }
+    endpoint, service_name, kwargs = cases[action]
+    service = Mock(return_value={'id': 'result'})
+    monkeypatch.setattr(api.po_svc, service_name, service)
+    monkeypatch.setattr(api.po_svc, 'order_to_dict', lambda *args, **kwargs: {'id': 'result'})
+    monkeypatch.setattr(api.po_svc, '_transfer_to_dict', lambda *args, **kwargs: {'id': 'result'})
+    result = asyncio.run(endpoint(db=db, user={**USER, 'dept_id': 'untrusted-dept'}, **kwargs))
+    assert result['code'] == 200
+    assert service.call_args.kwargs['create_dept_id'] == 'dept-a'
+
+
+def test_collection_protocol_create_uses_server_department(db, monkeypatch):
+    from types import SimpleNamespace
+    from app.api import collection_protocol
+
+    create = Mock(return_value=SimpleNamespace(id='new-protocol', payment_model_version=2))
+    monkeypatch.setattr(collection_protocol.cp_svc, 'create_collection_protocol', create)
+    monkeypatch.setattr(collection_protocol.cp_svc, '_protocol_to_dict', lambda obj: {'id': obj.id})
+    monkeypatch.setattr(collection_protocol, '_persist_protocol_sub_resources', Mock())
+    result = asyncio.run(collection_protocol.create_collection_protocol(
+        body=collection_protocol.ProtocolCreateBody(), db=db,
+        user={**USER, 'dept_id': 'untrusted-dept'},
+    ))
+    assert result['code'] == 200
+    assert create.call_args.kwargs['create_dept_id'] == 'dept-a'

backend/tests/test_web_sensitive_permissions.py

@@ -337,21 +337,9 @@ def _route_strict_codes(router, endpoint_name):
             [
                 "create_research_donation",
                 "update_research_donation",
-                "delete_research_donation",
-                "batch_delete_research_donations",
             ],
             {PROTOCOL_LIST},
         ),
-        (
-            "research_donation",
-            [
-                "submit_donation_audit",
-                "change_donation_status",
-                "convert_from_storage",
-                "convert_to_storage",
-            ],
-            {PROTOCOL_AUDIT},
-        ),
         (
             "protocol_sample_payment",
             ["confirm_payment_transfer", "reject_payment_transfer"],
@@ -362,15 +350,6 @@ def _route_strict_codes(router, endpoint_name):
             ["confirm_presign_transfer", "reject_presign_transfer"],
             {PAYMENT_CONFIRM},
         ),
-        (
-            "protocol_refund_record",
-            [
-                "create_protocol_refund_record",
-                "update_protocol_refund_record",
-                "delete_protocol_refund_record",
-            ],
-            {REFUND_CONFIRM},
-        ),
         (
             "sample_cert_insurance",
             ["put_sample_storage_cert", "put_sample_insurance"],
@@ -391,6 +370,13 @@ def test_sensitive_write_routes_declare_strict_menu_dependencies(
         assert _route_strict_codes(module.router, endpoint_name) == expected_codes
 
 
+def test_research_donation_audit_route_has_no_strict_menu_dependency():
+    """科研捐赠审核只保留运营端登录，不再要求审核菜单权限。"""
+    from app.api import research_donation
+
+    assert _route_strict_codes(research_donation.router, "submit_donation_audit") == set()
+
+
 def test_collection_protocol_write_routes_have_no_strict_menu_dependencies():
     """采集协议单接口仅保留运营端登录守卫，不再挂菜单权限码。"""
     from app.api import collection_protocol
@@ -416,6 +402,19 @@ def test_collection_protocol_write_routes_have_no_strict_menu_dependencies():
         assert _route_strict_codes(collection_protocol.router, endpoint_name) == set()
 
 
+def test_protocol_refund_record_routes_have_no_strict_menu_dependencies():
+    """采集协议单上的退费登记只保留运营端登录，不再要求退费确认菜单。"""
+    from app.api import protocol_refund_record
+
+    for endpoint_name in (
+        "list_protocol_refund_records",
+        "create_protocol_refund_record",
+        "update_protocol_refund_record",
+        "delete_protocol_refund_record",
+    ):
+        assert _route_strict_codes(protocol_refund_record.router, endpoint_name) == set()
+
+
 def test_collection_protocol_write_paths_keep_payment_detail_guard_only():
     source = Path("app/api/collection_protocol.py").read_text()
     assert "assert_protocol_sensitive_fields" in source
@@ -518,7 +517,6 @@ def test_market_only_user_is_stopped_before_sensitive_crud(monkeypatch):
     采集协议单接口已按方案去掉菜单码，不在本用例断言范围内。
     """
     from app.api import (
-        protocol_refund_record,
         protocol_sample_payment,
         sample_cert_insurance,
         sample_storage_cert_qc,
@@ -548,9 +546,6 @@ def test_market_only_user_is_stopped_before_sensitive_crud(monkeypatch):
     monkeypatch.setattr(
         protocol_sample_payment.psp_svc, "confirm_receipt", forbidden_call
     )
-    monkeypatch.setattr(
-        protocol_refund_record.rr_svc, "create_refund_record", forbidden_call
-    )
     monkeypatch.setattr(
         sample_cert_insurance.cert_crud, "upsert_handle", forbidden_call
     )
@@ -559,7 +554,6 @@ def test_market_only_user_is_stopped_before_sensitive_crud(monkeypatch):
     app = FastAPI()
     for router in (
         protocol_sample_payment.router,
-        protocol_refund_record.router,
         sample_cert_insurance.router,
         sample_storage_cert_qc.router,
     ):
@@ -575,11 +569,6 @@ def test_market_only_user_is_stopped_before_sensitive_crud(monkeypatch):
 
     requests = [
         ("post", "/web/admin/payment-transfers/t1/receipt-confirm", {}),
-        (
-            "post",
-            "/web/admin/collection-protocols/p1/refund-records",
-            {"refundAmount": 1},
-        ),
         ("put", "/web/admin/sample-storage-certs/S1", {}),
         ("put", "/web/admin/sample-storage-cert-qc/S1", {"payload": {}}),
     ]