解析失败、重试与回退：超时不等于失败

"""模拟创建订单后回执丢失，用同一业务编号重试并检查订单数量。"""

from copy import deepcopy
import json
import time


class OrderService:
    """按业务编号保存订单参数和结果，重复收到同一请求时返回已有结果。"""

    def __init__(self):
        self._records = {}
        self.effects = 0

    def create_order(self, operation_key, payload, *, lose_response=False):
        if operation_key in self._records:
            saved_payload, result = self._records[operation_key]
            if payload != saved_payload:
                raise ValueError("operation key reused with different payload")
            return deepcopy(result)

        self.effects += 1
        result = {"order_id": f"order-{self.effects}"}
        self._records[operation_key] = (deepcopy(payload), result)
        # 订单已经创建；响应超时仍需核对业务结果。
        if lose_response:
            raise TimeoutError("order committed, response lost")
        return deepcopy(result)


def retry_operation(operation, *, idempotent, max_attempts=3, sleep=time.sleep):
    """仅重试 TimeoutError；次数包含第一次调用，示例不加入随机抖动。

    idempotent=True 表示调用者已按业务契约确认可安全重试，
    这个标志本身不能让任意操作自动具备幂等性。
    """
    if type(max_attempts) is not int or max_attempts < 1:
        raise ValueError("max_attempts must be a positive integer")
    for attempt in range(1, max_attempts + 1):
        try:
            return operation()
        except TimeoutError:
            if not idempotent:
                return {"status": "manual_review"}
            if attempt == max_attempts:
                raise
            sleep(0.1 * 2 ** (attempt - 1))


def run_demo():
    """四个固定场景；注入等待函数，只记录间隔而不实际等待。"""
    results = []
    for scenario in ("retry-safe", "unsafe", "invalid", "exhausted"):
        service = OrderService()
        attempts = 0
        unsafe_effects = 0
        delays = []
        payload = {"item": "book", "quantity": 1}
        if scenario == "invalid":
            service.create_order("checkout-1", payload)

        def operation():
            nonlocal attempts, unsafe_effects
            attempts += 1
            if scenario == "unsafe":
                unsafe_effects += 1
                raise TimeoutError("non-idempotent order committed, response lost")
            if scenario == "exhausted":
                raise TimeoutError("service unavailable before commit")
            if scenario == "invalid":
                return service.create_order("checkout-1", {"item": "pen", "quantity": 1})
            return service.create_order("checkout-1", payload, lose_response=True)

        try:
            result = retry_operation(operation, idempotent=scenario != "unsafe",
                                     sleep=delays.append)
            status = result.get("status", "succeeded")
        except ValueError:
            status = "rejected"
        except TimeoutError:
            status = "exhausted"
        results.append({"scenario": scenario, "status": status, "attempts": attempts,
                        "effects": service.effects + unsafe_effects,
                        "backoff_seconds": delays})
    return results


if __name__ == "__main__":
    print(json.dumps(run_demo(), indent=2))

# 核对解析失败、重试与回退的正常结果和错误分支；断言失败时会显示对应的测试。
"""Offline tests: python -B -m unittest -v test_retry_lab.py"""

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()