feat: 实现受控真实下单安全边界 (#99)

2026-08-10 16:35:38 +08:00
parent 7106f6b024
commit e960cab1fb
30 changed files with 1562 additions and 66 deletions
@@ -0,0 +1,87 @@
+"""真实下单授权服务测试；不连接 Admin 或手机。"""
+
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+from src.live_purchase_authorization import (
+    LIVE_CONFIRMATION_TEXT,
+    LIVE_ENABLED_KEY,
+    LivePurchaseAuthorizationService,
+)
+from src.settings_repository import SettingsRepository
+
+
+class LivePurchaseAuthorizationTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temporary_directory = tempfile.TemporaryDirectory()
+        repository = SettingsRepository(
+            Path(self.temporary_directory.name) / "client.db"
+        )
+        self.repository = repository
+        self.service = LivePurchaseAuthorizationService(repository)
+
+    def tearDown(self) -> None:
+        self.temporary_directory.cleanup()
+
+    def test_missing_setting_and_environment_always_default_to_dry_run(self):
+        os.environ["CMAUTOBUY_PURCHASE_MODE"] = "live"
+        try:
+            self.assertFalse(self.service.load().enabled)
+            self.assertEqual(
+                self.service.purchase_mode_for(
+                    "CLIENT-1", "USB-1", live_adapter_ready=True
+                ),
+                "dry_run",
+            )
+        finally:
+            os.environ.pop("CMAUTOBUY_PURCHASE_MODE", None)
+
+    def test_enable_requires_exact_confirmation_and_binds_identity(self):
+        with self.assertRaisesRegex(ValueError, LIVE_CONFIRMATION_TEXT):
+            self.service.enable("CLIENT-1", "USB-1", "确认")
+        self.assertIsNone(self.repository.get(LIVE_ENABLED_KEY))
+
+        authorization = self.service.enable(
+            "CLIENT-1", "USB-1", LIVE_CONFIRMATION_TEXT
+        )
+
+        self.assertTrue(authorization.enabled)
+        self.assertTrue(authorization.matches("CLIENT-1", "USB-1"))
+        self.assertEqual(
+            self.service.purchase_mode_for(
+                "CLIENT-1", "USB-1", live_adapter_ready=True
+            ),
+            "live",
+        )
+
+    def test_binding_mismatch_or_missing_adapter_stays_dry_run(self):
+        self.service.enable("CLIENT-1", "USB-1", LIVE_CONFIRMATION_TEXT)
+
+        self.assertEqual(
+            self.service.purchase_mode_for(
+                "CLIENT-2", "USB-1", live_adapter_ready=True
+            ),
+            "dry_run",
+        )
+        self.assertEqual(
+            self.service.purchase_mode_for(
+                "CLIENT-1", "USB-2", live_adapter_ready=True
+            ),
+            "dry_run",
+        )
+        self.assertEqual(
+            self.service.purchase_mode_for(
+                "CLIENT-1", "USB-1", live_adapter_ready=False
+            ),
+            "dry_run",
+        )
+
+    def test_disable_is_immediate_and_does_not_require_confirmation(self):
+        self.service.enable("CLIENT-1", "USB-1", LIVE_CONFIRMATION_TEXT)
+
+        authorization = self.service.disable()
+
+        self.assertFalse(authorization.enabled)
+        self.assertFalse(self.service.load().enabled)