功能：新增托盘状态机

shierfeng-fastapi-backend/module_business/domain/enums.py

@@ -0,0 +1,30 @@
+from enum import Enum
+
+
+class PalletFlowState(str, Enum):
+    UNSTOCKED = 'UNSTOCKED'
+    EMPTY_IN_WAREHOUSE = 'EMPTY_IN_WAREHOUSE'
+    FILLED_IN_WAREHOUSE = 'FILLED_IN_WAREHOUSE'
+    AT_SITE_UNUSED = 'AT_SITE_UNUSED'
+    AT_SITE_PARTIAL = 'AT_SITE_PARTIAL'
+    AT_SITE_EMPTY = 'AT_SITE_EMPTY'
+    RETURNING = 'RETURNING'
+
+
+class DocumentStatus(str, Enum):
+    DRAFT = 'DRAFT'
+    CONFIRMED = 'CONFIRMED'
+
+
+class ReturnStatus(str, Enum):
+    REQUESTED = 'REQUESTED'
+    CONFIRMED = 'CONFIRMED'
+
+
+class PalletEventType(str, Enum):
+    INBOUND_CONFIRMED = 'INBOUND_CONFIRMED'
+    FILLING_CONFIRMED = 'FILLING_CONFIRMED'
+    SHIPMENT_CONFIRMED = 'SHIPMENT_CONFIRMED'
+    USAGE_RECORDED = 'USAGE_RECORDED'
+    RETURN_REQUESTED = 'RETURN_REQUESTED'
+    RETURN_CONFIRMED = 'RETURN_CONFIRMED'

shierfeng-fastapi-backend/module_business/domain/pallet_state_machine.py

@@ -0,0 +1,38 @@
+from module_business.domain.enums import PalletFlowState
+
+
+class InvalidPalletTransition(ValueError):
+    """托盘状态迁移不符合业务规则。"""
+
+
+class PalletStateMachine:
+    _ALLOWED_TRANSITIONS = frozenset(
+        {
+            (PalletFlowState.UNSTOCKED, PalletFlowState.EMPTY_IN_WAREHOUSE),
+            (
+                PalletFlowState.EMPTY_IN_WAREHOUSE,
+                PalletFlowState.FILLED_IN_WAREHOUSE,
+            ),
+            (PalletFlowState.FILLED_IN_WAREHOUSE, PalletFlowState.AT_SITE_UNUSED),
+            (PalletFlowState.AT_SITE_UNUSED, PalletFlowState.AT_SITE_PARTIAL),
+            (PalletFlowState.AT_SITE_UNUSED, PalletFlowState.AT_SITE_EMPTY),
+            (PalletFlowState.AT_SITE_UNUSED, PalletFlowState.RETURNING),
+            (PalletFlowState.AT_SITE_PARTIAL, PalletFlowState.AT_SITE_PARTIAL),
+            (PalletFlowState.AT_SITE_PARTIAL, PalletFlowState.AT_SITE_EMPTY),
+            (PalletFlowState.AT_SITE_PARTIAL, PalletFlowState.RETURNING),
+            (PalletFlowState.AT_SITE_EMPTY, PalletFlowState.RETURNING),
+            (PalletFlowState.RETURNING, PalletFlowState.FILLED_IN_WAREHOUSE),
+            (PalletFlowState.RETURNING, PalletFlowState.EMPTY_IN_WAREHOUSE),
+        }
+    )
+
+    @classmethod
+    def ensure_transition(
+        cls,
+        current: PalletFlowState,
+        target: PalletFlowState,
+    ) -> None:
+        if (current, target) not in cls._ALLOWED_TRANSITIONS:
+            raise InvalidPalletTransition(
+                f'不允许的托盘状态迁移：{current.value} -> {target.value}'
+            )

shierfeng-fastapi-backend/tests/module_business/test_state_machine.py

@@ -0,0 +1,76 @@
+import unittest
+
+from module_business.domain.enums import (
+    DocumentStatus,
+    PalletEventType,
+    PalletFlowState,
+    ReturnStatus,
+)
+from module_business.domain.pallet_state_machine import (
+    InvalidPalletTransition,
+    PalletStateMachine,
+)
+
+
+class PalletStateMachineTest(unittest.TestCase):
+    ALLOWED_TRANSITIONS = (
+        (PalletFlowState.UNSTOCKED, PalletFlowState.EMPTY_IN_WAREHOUSE),
+        (PalletFlowState.EMPTY_IN_WAREHOUSE, PalletFlowState.FILLED_IN_WAREHOUSE),
+        (PalletFlowState.FILLED_IN_WAREHOUSE, PalletFlowState.AT_SITE_UNUSED),
+        (PalletFlowState.AT_SITE_UNUSED, PalletFlowState.AT_SITE_PARTIAL),
+        (PalletFlowState.AT_SITE_UNUSED, PalletFlowState.AT_SITE_EMPTY),
+        (PalletFlowState.AT_SITE_UNUSED, PalletFlowState.RETURNING),
+        (PalletFlowState.AT_SITE_PARTIAL, PalletFlowState.AT_SITE_PARTIAL),
+        (PalletFlowState.AT_SITE_PARTIAL, PalletFlowState.AT_SITE_EMPTY),
+        (PalletFlowState.AT_SITE_PARTIAL, PalletFlowState.RETURNING),
+        (PalletFlowState.AT_SITE_EMPTY, PalletFlowState.RETURNING),
+        (PalletFlowState.RETURNING, PalletFlowState.FILLED_IN_WAREHOUSE),
+        (PalletFlowState.RETURNING, PalletFlowState.EMPTY_IN_WAREHOUSE),
+    )
+
+    def test_confirmed_flow_is_allowed(self) -> None:
+        for current, target in self.ALLOWED_TRANSITIONS:
+            PalletStateMachine.ensure_transition(current, target)
+
+    def test_refill_at_customer_site_is_rejected(self) -> None:
+        with self.assertRaises(InvalidPalletTransition):
+            PalletStateMachine.ensure_transition(
+                PalletFlowState.AT_SITE_EMPTY,
+                PalletFlowState.FILLED_IN_WAREHOUSE,
+            )
+
+    def test_confirmed_flow_is_the_only_allowed_transition_set(self) -> None:
+        actual_allowed = set()
+        for current in PalletFlowState:
+            for target in PalletFlowState:
+                try:
+                    PalletStateMachine.ensure_transition(current, target)
+                except InvalidPalletTransition:
+                    continue
+                actual_allowed.add((current, target))
+
+        self.assertEqual(set(self.ALLOWED_TRANSITIONS), actual_allowed)
+
+
+class PalletEnumTest(unittest.TestCase):
+    def test_supporting_enum_values_match_domain_contract(self) -> None:
+        self.assertEqual(['DRAFT', 'CONFIRMED'], [item.value for item in DocumentStatus])
+        self.assertEqual(
+            ['REQUESTED', 'CONFIRMED'],
+            [item.value for item in ReturnStatus],
+        )
+        self.assertEqual(
+            [
+                'INBOUND_CONFIRMED',
+                'FILLING_CONFIRMED',
+                'SHIPMENT_CONFIRMED',
+                'USAGE_RECORDED',
+                'RETURN_REQUESTED',
+                'RETURN_CONFIRMED',
+            ],
+            [item.value for item in PalletEventType],
+        )
+
+
+if __name__ == '__main__':
+    unittest.main()