工具超时以后：重试、幂等与失败恢复

"""校验重复意图、冲突、未知结果和重试预算。"""

import importlib.util
import unittest
from pathlib import Path


class RetryLabTests(unittest.TestCase):
    def setUp(self):
        path = Path(__file__).with_name("retry_lab.py")
        self.assertTrue(path.is_file(), "retry_lab.py must implement the lab")
        spec = importlib.util.spec_from_file_location("retry_lab", path)
        self.lab = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(self.lab)

    def test_same_key_same_payload_returns_original_order(self):
        service = self.lab.OrderService()
        first = service.create_order("op-1", {"item": "book", "quantity": 1})
        second = service.create_order("op-1", {"quantity": 1, "item": "book"})
        self.assertEqual(first, second)
        self.assertEqual(service.effects, 1)

    def test_conflicting_payload_is_rejected_without_effect(self):
        service = self.lab.OrderService()
        service.create_order("op-1", {"item": "book"})
        with self.assertRaises(ValueError):
            service.create_order("op-1", {"item": "pen"})
        self.assertEqual(service.effects, 1)

    def test_different_keys_create_different_orders(self):
        service = self.lab.OrderService()
        self.assertNotEqual(service.create_order("a", {"item": "book"}),
                            service.create_order("b", {"item": "book"}))
        self.assertEqual(service.effects, 2)

    def test_input_and_output_mutation_cannot_change_saved_record(self):
        service = self.lab.OrderService()
        payload = {"items": ["book"]}
        result = service.create_order("a", payload)
        payload["items"].append("pen")
        result["order_id"] = "tampered"
        self.assertEqual(service.create_order("a", {"items": ["book"]}),
                         {"order_id": "order-1"})
        with self.assertRaises(ValueError):
            service.create_order("a", payload)

    def test_lost_response_after_commit_is_safe_to_retry(self):
        # 回执丢失后的第二次尝试必须复用业务键，效果仍只能一次。
        service = self.lab.OrderService()
        attempts = []
        delays = []

        def operation():
            attempts.append(1)
            return service.create_order("a", {"item": "book"}, lose_response=True)

        result = self.lab.retry_operation(operation, idempotent=True, sleep=delays.append)
        self.assertEqual(result, {"order_id": "order-1"})
        self.assertEqual(len(attempts), 2)
        self.assertEqual(service.effects, 1)
        self.assertEqual(delays, [0.1])

    def test_unsafe_unknown_outcome_requires_manual_review(self):
        calls, delays = [], []

        def operation():
            calls.append(1)
            raise TimeoutError("response lost; outcome unknown")

        self.assertEqual(self.lab.retry_operation(operation, idempotent=False,
                                                 sleep=delays.append),
                         {"status": "manual_review"})
        self.assertEqual(len(calls), 1)
        self.assertEqual(delays, [])

    def test_non_timeout_error_is_not_retried(self):
        for error_type in (ValueError, RuntimeError):
            with self.subTest(error_type=error_type):
                calls, delays = [], []

                def operation():
                    calls.append(1)
                    raise error_type("rejected")

                with self.assertRaises(error_type):
                    self.lab.retry_operation(operation, idempotent=True, sleep=delays.append)
                self.assertEqual(len(calls), 1)
                self.assertEqual(delays, [])

    def test_exhaustion_has_three_attempts_and_two_backoffs(self):
        calls, delays = [], []

        def operation():
            calls.append(1)
            raise TimeoutError("still unavailable")

        with self.assertRaises(TimeoutError):
            self.lab.retry_operation(operation, idempotent=True, sleep=delays.append)
        self.assertEqual(len(calls), 3)
        self.assertEqual(delays, [0.1, 0.2])

    def test_one_attempt_does_not_sleep_after_failure(self):
        delays = []

        def operation():
            raise TimeoutError("unavailable")

        with self.assertRaises(TimeoutError):
            self.lab.retry_operation(operation, idempotent=True, max_attempts=1,
                                     sleep=delays.append)
        self.assertEqual(delays, [])

    def test_success_returns_immediately(self):
        delays = []
        self.assertEqual(self.lab.retry_operation(lambda: "ok", idempotent=False,
                                                 sleep=delays.append), "ok")
        self.assertEqual(delays, [])

    def test_invalid_attempt_budget_is_rejected_before_call(self):
        for budget in (0, -1, 1.5, "3", True, None):
            with self.subTest(budget=budget):
                calls = []
                with self.assertRaises(ValueError):
                    self.lab.retry_operation(lambda: calls.append(1), idempotent=True,
                                             max_attempts=budget, sleep=lambda _: None)
                self.assertEqual(calls, [])

    def test_demo_reports_four_scenarios(self):
        results = self.lab.run_demo()
        self.assertEqual([row["scenario"] for row in results],
                         ["retry-safe", "unsafe", "invalid", "exhausted"])
        self.assertEqual([row["attempts"] for row in results], [2, 1, 1, 3])
        self.assertEqual(results[0]["effects"], 1)
        self.assertEqual(results[1]["status"], "manual_review")
        self.assertEqual(results[2]["status"], "rejected")
        self.assertEqual(results[3]["status"], "exhausted")


if __name__ == "__main__":
    unittest.main()