آزمون خطای آماده‌سازی ورودی حسگر؛ ExitStack و ترتیب پاک‌سازی

برنامهٔ آماده‌سازیِ دادهٔ مهندسی ممکن است ورودیِ نخست را آماده کند، هنگامِ گرفتنِ دومی خطا بدهد و پیش از آغازِ پردازش متوقف شود. پیامِ «ورودی آماده نشد» به‌تنهایی نمی‌گوید منبعِ قبلی آزاد شده یا نه. این درس، مالکیت و ترتیبِ پاک‌سازی را با ExitStack واقعی و سه منبعِ کاملاً ساختگی می‌آزماید. هیچ فایل، پورت، سوکت یا حسگرِ واقعی باز نمی‌شود؛ رخدادهای قابل‌مشاهده، مرجعِ رفتارند و زمانِ اجرای سخت‌افزار اندازه گرفته نمی‌شود.

۱. منبعِ گرفته‌شده را از تلاشِ ناموفق جدا کنیم

منابع با نام‌های A، B و C مشخص‌اند. هر نمونه چهار حالتِ آموزشی دارد: تازه، باز، بسته یا ورودِ ناموفق. ساختِ شیء، در این مثال فقط حالتِ پایتونی می‌سازد و منبعِ بیرونی نمی‌گیرد. __enter__ یا موفق می‌شود و حالت را باز می‌کند، یا پیش از گرفتنِ هر منبع، خطای ساختگی می‌دهد. بنابراین «تلاش برای ورود» با «مالکِ منبعِ آماده بودن» یک رخداد نیست.

مرجعِ دستورِ with، مراحلِ ورود، اجرای بدنه و خروج را جدا تعریف می‌کند. پس از بازگشتِ موفقِ __enter__، مسیرِ خروج به قراردادِ مدیرِ زمینه سپرده می‌شود. اما اگر خودِ ورود، بخشی از منبع را گرفته و سپس شکست بخورد، همان پیاده‌سازی باید پاک‌سازیِ داخلیِ آن بخش را طراحی کند. مثالِ ما آن وضعیتِ نیمه‌گرفته را جعل نمی‌کند؛ شکستِ تزریقی پیش از تخصیص است.

تابعِ سناریو حداکثر سه نامِ یکتای همان مجموعه را می‌گیرد؛ نام‌های تکراری، نامِ دیگر یا فهرست به‌جای tuple رد می‌شوند. هدفِ شکست باید در انتخاب باشد و پرچمِ خطای بدنه، bool دقیق است. اعتبارسنجی پیش از ساختِ منابع انجام می‌شود. این محدودیت، دامنهٔ خوانای آزمون است؛ تعدادِ مجازِ منابع در خودِ ExitStack به سه محدود نیست.

۲. ثبتِ خروج پس از ورودِ موفق

ExitStack در مرجعِ contextlib برای ترکیبِ مدیرهای زمینه و عملیاتِ پاک‌سازی است. enter_context، ورود را اجرا می‌کند و پس از موفقیت، خروجِ آن مدیر را ثبت می‌کند. نتیجهٔ برگشتی، همان مقدارِ __enter__ است؛ در نمونهٔ ما خودِ شیء است. به همین دلیل، مالکیت را از روی ثبتِ موفقِ ورود می‌سنجیم، نه فقط از شمارِ شیءهای ساخته‌شده.

در سناریوی A و Bِ موفق، سپس بدنه اجرا و خروج‌ها به‌ترتیبِ B و A فراخوانی می‌شوند. این ترتیب، معکوسِ ثبت است. برای ورودی‌های وابسته، همین ترتیب می‌تواند مهم باشد؛ ولی کدِ این درس وابستگیِ واقعیِ درایور یا فایل را شبیه‌سازی نمی‌کند. هر خروج در حالتِ صحیحِ باز، فقط یک رخداد ثبت می‌کند، حالت را بسته می‌کند و False برمی‌گرداند؛ پاک‌سازیِ معمولِ ما استثنا نمی‌سازد.

اگر ورودِ B شکست بخورد، ردِ موردِ انتظار ورودِ A، تلاشِ B، شکستِ B و خروجِ A با اطلاعاتِ AcquireError است. خروجِ B ثبت نشده؛ C حتی امتحان نشده و بدنه نیز اجرا نمی‌شود. نبودِ رخدادِ C با «باز شد و بسته شد» یکی نیست. آزمون، این تفاوت را با فهرستِ دقیقِ رخدادها و حالت‌های نهایی کنترل می‌کند.

۳. کدِ کامل و خطای بدنهٔ پردازش

در سناریوی سوم، هر سه ورود موفق‌اند، سپس بدنه BodyError می‌دهد. انتظارِ دستی، خروجِ C، B و A با همان نوعِ خطاست. تابعِ آزمون علاوه بر نامِ نوع، هویتِ همان شیء استثنای ساخته‌شده را هم مقایسه می‌کند. این شاهد فقط برای خروج‌های عادیِ بدونِ خطا و بدونِ سرکوبِ استثنای خودمان است؛ اصلِ خطا در حضورِ هر نوع پاک‌سازیِ خراب، خودکار حفظ نمی‌شود.

"""Actual ExitStack on deterministic simulated input resources; no device/files."""
from contextlib import ExitStack
import json
import platform


class AcquireError(Exception):
    pass


class BodyError(Exception):
    pass


class InputResource:
    def __init__(self, name, events, fail_enter=False):
        self.name = name
        self.events = events
        self.fail_enter = fail_enter
        self.state = "new"
        self.failure = AcquireError("synthetic_acquisition_failure")

    def __enter__(self):
        if self.state != "new":
            raise RuntimeError("single_use_resource")
        self.events.append("enter:" + self.name)
        if self.fail_enter:
            # No simulated resource has been acquired before this injected failure.
            self.state = "failed"
            self.events.append("fail:" + self.name)
            raise self.failure
        self.state = "open"
        return self

    def __exit__(self, exc_type, exc, traceback):
        if self.state != "open":
            raise RuntimeError("exit_without_open_resource")
        label = "none" if exc_type is None else exc_type.__name__
        self.events.append("exit:" + self.name + ":" + label)
        self.state = "closed"
        return False  # Own normal cleanup never throws or suppresses an exception.


def simulate(names, *, fail_at=None, fail_body=False):
    if type(names) is not tuple or len(names) > 3 or any(type(n) is not str or n not in ("A", "B", "C") for n in names) or len(set(names)) != len(names):
        raise ValueError("resource_names")
    if fail_at is not None and (type(fail_at) is not str or fail_at not in names):
        raise ValueError("failure_target")
    if type(fail_body) is not bool:
        raise ValueError("body_failure_flag")
    events, resources = [], []
    body_error = BodyError("synthetic_processing_failure")
    caught = None
    with_error_identity = True
    try:
        with ExitStack() as stack:
            for name in names:
                resource = InputResource(name, events, name == fail_at)
                resources.append(resource)
                stack.enter_context(resource)
            events.append("body")
            if fail_body:
                raise body_error
    except (AcquireError, BodyError) as exc:
        caught = type(exc).__name__
        expected = body_error if type(exc) is BodyError else resources[-1].failure
        with_error_identity = exc is expected
    return {"events": events, "states": [[r.name, r.state] for r in resources],
            "caught_error": caught, "same_original_error_object": with_error_identity}


def invalid_code(call):
    try:
        call()
    except ValueError as exc:
        return str(exc)
    return None


def run_example():
    checks = {}
    def check(name, value):
        checks[name] = bool(value)
        if not checks[name]:
            raise AssertionError(name)
    clean = simulate(("A", "B"))
    check("two_successful_entries_exit_in_literal_reverse_order", clean["events"] == ["enter:A", "enter:B", "body", "exit:B:none", "exit:A:none"] and clean["caught_error"] is None)
    check("successful_enter_values_are_closed_after_scope", clean["states"] == [["A", "closed"], ["B", "closed"]])
    failed = simulate(("A", "B", "C"), fail_at="B")
    check("second_acquisition_failure_unwinds_only_acquired_first", failed["events"] == ["enter:A", "enter:B", "fail:B", "exit:A:AcquireError"])
    check("failed_enter_has_no_exit_and_later_resource_not_attempted", failed["states"] == [["A", "closed"], ["B", "failed"]] and failed["caught_error"] == "AcquireError")
    body = simulate(("A", "B", "C"), fail_body=True)
    check("body_failure_cleanup_receives_exception_in_literal_lifo_order", body["events"] == ["enter:A", "enter:B", "enter:C", "body", "exit:C:BodyError", "exit:B:BodyError", "exit:A:BodyError"])
    check("bounded_nonthrowing_nonsuppressing_cleanup_preserves_error_identity", body["same_original_error_object"] and failed["same_original_error_object"] and body["caught_error"] == "BodyError")
    first_fail = simulate(("A", "B"), fail_at="A")
    check("first_enter_failure_registers_no_exit", first_fail["events"] == ["enter:A", "fail:A"] and first_fail["states"] == [["A", "failed"]])
    check("empty_resource_selection_still_runs_body", simulate(())["events"] == ["body"] and simulate(())["states"] == [])
    callback_events = []
    stack = ExitStack()
    stack.callback(callback_events.append, "first")
    stack.callback(callback_events.append, "second")
    check("callback_registration_itself_does_not_invoke_cleanup", callback_events == [])
    stack.close()
    check("explicit_close_unwinds_registered_callbacks_lifo", callback_events == ["second", "first"])
    stack.close()
    check("closed_stack_does_not_repeat_its_removed_callbacks", callback_events == ["second", "first"])
    callback_error = BodyError("callback_test")
    survived = None
    callback_log = []
    def true_cleanup():
        callback_log.append("callback")
        return True
    try:
        with ExitStack() as stack:
            stack.callback(true_cleanup)
            raise callback_error
    except BodyError as exc:
        survived = exc
    check("callback_returning_true_cannot_suppress_body_error", survived is callback_error and callback_log == ["callback"])
    exit_log = []
    with ExitStack() as stack:
        def explicit_suppressor(exc_type, exc, traceback):
            exit_log.append(exc_type.__name__)
            return True
        stack.push(explicit_suppressor)
        raise BodyError("separate_exit_protocol_demonstration")
    check("push_exit_protocol_can_suppress_distinct_from_callback", exit_log == ["BodyError"])
    events = []
    resource = InputResource("A", events)
    with ExitStack() as stack:
        returned = stack.enter_context(resource)
        check("enter_context_returns_the_resource_enter_value", returned is resource and resource.state == "open")
    cases = 0
    for names in (("A",), ("A", "B"), ("A", "B", "C")):
        for failure_index, failure_name in enumerate(names):
            result = simulate(names, fail_at=failure_name)
            prefix = names[:failure_index]
            expected = ["enter:" + n for n in prefix] + ["enter:" + failure_name, "fail:" + failure_name]
            expected += ["exit:" + n + ":AcquireError" for n in reversed(prefix)]
            if result["events"] != expected or not result["same_original_error_object"]:
                raise AssertionError("independent_acquired_prefix_reference")
            cases += 1
    check("six_acquired_prefix_failure_references_one_group", cases == 6)
    check("bounded_names_and_failure_policy_reject_invalid_inputs", invalid_code(lambda: simulate(["A"])) == "resource_names"
          and invalid_code(lambda: simulate(("A", "A"))) == "resource_names"
          and invalid_code(lambda: simulate(("D",))) == "resource_names"
          and invalid_code(lambda: simulate(("A",), fail_at="B")) == "failure_target"
          and invalid_code(lambda: simulate(("A",), fail_body=1)) == "body_failure_flag")
    return {"python_version": platform.python_version(), "scope": "actual contextlib.ExitStack with only simulated input-resource objects; max3 uniqueA/B/C names, no file, socket, hardware, async or network",
            "clean_scenario": clean, "second_enter_failure": failed, "body_failure": body,
            "failure_prefix_reference_cases_one_group": cases,
            "callback_cleanup_order": callback_events,
            "limitations": "Own ordinary resource cleanup is nonthrowing/nonsuppressing; it does not undo processing or repair failed acquisition internals. A manager whose __enter__ partially acquires then fails must handle that internal cleanup itself. Separate push demonstration intentionally suppresses; callback truthy return cannot. No crash, garbage-collection cleanup, durability, timing or rollback guarantee.",
            "checks": checks}


if __name__ == "__main__":
    print(json.dumps(run_example(), ensure_ascii=False, indent=2))

هر __exit__، نوعِ خطا یا نبودِ آن را در دفترِ رخداد ثبت می‌کند. برای مسیرِ موفق، none و برای شکستِ بدنه BodyError دیده می‌شود. False یعنی همان استثنا در این مسیر سرکوب نمی‌شود. پوشانندهٔ سناریو، خطاهای ساختگیِ شناخته‌شده را برای گزارش می‌گیرد؛ این تصمیم، الگوی عمومیِ بلعیدنِ خطای برنامه نیست. گزارشِ «بسته» نیز فقط حالتِ همین شیء است، نه رسیدِ بسته‌شدنِ دستگاهی در بیرون.

بعد از خطا، پردازشِ انجام‌شده عقب کشیده نمی‌شود. این آزمایش حتی داده‌ای در ذخیره‌ساز نمی‌نویسد که بتواند دربارهٔ تراکنشِ آن حرف بزند. خروجِ مرتب، با rollback، دوامِ داده و بازیابیِ پس از قطعِ برق فرق دارد. از دست‌رفتنِ ناگهانیِ فرایند هم ممکن است فرصتِ خروجِ معمول را ندهد؛ مشاهدهٔ موفقِ finally یا خروجِ زمینه، آزمونِ crash واقعی نیست.

۴. callback، خروجِ پروتکلی و مرجعِ رویداد

callback یک تابعِ پاک‌سازی با آرگومان‌های مشخص ثبت می‌کند و اطلاعاتِ استثنا به آن داده نمی‌شود. در شاهدِ جدا، تابع True برمی‌گرداند، اما BodyError همچنان عبور می‌کند؛ آن مقدارِ برگشتی، استثنا را سرکوب نمی‌کند. در مقابل، تابعی که با push و امضای کاملِ خروج ثبت شده، می‌تواند با True خطا را سرکوب کند. آن نمایشِ جدا، سیاستِ منابعِ عادیِ ما نیست و نباید با callback جابه‌جا شود.

{
  "python_version": "3.14.5",
  "scope": "actual contextlib.ExitStack with only simulated input-resource objects; max3 uniqueA/B/C names, no file, socket, hardware, async or network",
  "clean_scenario": {
    "events": [
      "enter:A",
      "enter:B",
      "body",
      "exit:B:none",
      "exit:A:none"
    ],
    "states": [
      [
        "A",
        "closed"
      ],
      [
        "B",
        "closed"
      ]
    ],
    "caught_error": null,
    "same_original_error_object": true
  },
  "second_enter_failure": {
    "events": [
      "enter:A",
      "enter:B",
      "fail:B",
      "exit:A:AcquireError"
    ],
    "states": [
      [
        "A",
        "closed"
      ],
      [
        "B",
        "failed"
      ]
    ],
    "caught_error": "AcquireError",
    "same_original_error_object": true
  },
  "body_failure": {
    "events": [
      "enter:A",
      "enter:B",
      "enter:C",
      "body",
      "exit:C:BodyError",
      "exit:B:BodyError",
      "exit:A:BodyError"
    ],
    "states": [
      [
        "A",
        "closed"
      ],
      [
        "B",
        "closed"
      ],
      [
        "C",
        "closed"
      ]
    ],
    "caught_error": "BodyError",
    "same_original_error_object": true
  },
  "failure_prefix_reference_cases_one_group": 6,
  "callback_cleanup_order": [
    "second",
    "first"
  ],
  "limitations": "Own ordinary resource cleanup is nonthrowing/nonsuppressing; it does not undo processing or repair failed acquisition internals. A manager whose __enter__ partially acquires then fails must handle that internal cleanup itself. Separate push demonstration intentionally suppresses; callback truthy return cannot. No crash, garbage-collection cleanup, durability, timing or rollback guarantee.",
  "checks": {
    "two_successful_entries_exit_in_literal_reverse_order": true,
    "successful_enter_values_are_closed_after_scope": true,
    "second_acquisition_failure_unwinds_only_acquired_first": true,
    "failed_enter_has_no_exit_and_later_resource_not_attempted": true,
    "body_failure_cleanup_receives_exception_in_literal_lifo_order": true,
    "bounded_nonthrowing_nonsuppressing_cleanup_preserves_error_identity": true,
    "first_enter_failure_registers_no_exit": true,
    "empty_resource_selection_still_runs_body": true,
    "callback_registration_itself_does_not_invoke_cleanup": true,
    "explicit_close_unwinds_registered_callbacks_lifo": true,
    "closed_stack_does_not_repeat_its_removed_callbacks": true,
    "callback_returning_true_cannot_suppress_body_error": true,
    "push_exit_protocol_can_suppress_distinct_from_callback": true,
    "enter_context_returns_the_resource_enter_value": true,
    "six_acquired_prefix_failure_references_one_group": true,
    "bounded_names_and_failure_policy_reject_invalid_inputs": true
  }
}

این تفاوت به معنی بی‌خطر بودنِ هر callback نیست؛ خودِ تابع می‌تواند هنگامِ اجرا خطا بدهد و مسیرِ خطا را عوض کند. چنین پاک‌سازیِ خطازایی در قراردادِ معمولِ این نمونه نداریم. پیش از اجرای close، ثبتِ دو callback هیچ رخدادِ پاک‌سازی تولید نکرده است؛ close آن‌ها را به‌ترتیبِ دوم و اول اجرا می‌کند. close دوباره، callbackهای ازپشته‌برداشته‌شده را تکرار نمی‌کند.

شش حالتِ شکستِ ورود، برای انتخابِ یک، دو و سه منبع، با مرجعِ مستقلِ پیشوندِ موفق مقایسه شده‌اند. انتظار از فهرستِ نام‌های اعلام‌شده و ترتیبِ معکوسِ پیشوند می‌آید؛ از خروجیِ ExitStack ساخته نمی‌شود. همهٔ شش حالت، یک گروه از شانزده گروهِ بررسی‌اند. حالتِ بدونِ منبع، شکستِ نخستین ورود، مقدارِ برگشتیِ ورود و ورودیِ نامناسب نیز جدا سنجیده می‌شوند؛ تکرارِ اجرا بررسیِ تازه نیست.

نباید خروج را به جمع‌آوریِ زباله واگذار کرد؛ مستندات، اجرای ضمنیِ callbackهای ExitStack در آن زمان را وعده نمی‌دهند. کد از with یا close صریح استفاده می‌کند و زمان‌سنج ندارد. اجرای ثبت‌شده Python 3.14.5 و نسخهٔ مستندات 3.14.8 است؛ منابع در ۲۵ مهر ۱۴۰۵ خوانده شده‌اند. هیچ کتابخانه‌ای نصب یا رفتارِ یک پورتِ واقعی آزمایش نشده است.

عمرِ وظیفه‌های TaskGroup مسئولیتِ لغو و انتظارِ ناهمگام را توضیح می‌دهد؛ اینجا خروجِ همگامِ منابع مهم است. نتیجهٔ خطا در جایگزینیِ فایل نیز اثرِ محلی را از پیامِ شکست جدا می‌کند. برای تجربهٔ خطای دقیق، ابتدا باید بدانیم کدام منبع واقعاً آماده شده و کدام اثر واقعاً رخ داده است؛ یک پیامِ کلی، جایِ این قرارداد را نمی‌گیرد.

۵. چهار پرسش با پاسخِ کامل

پرسش اول: با شکستِ ورودِ B، خروجِ چه منبعی انتظار می‌رود؟ پاسخ: فقط A که ورودش موفق بوده است. B در این fixture پیش از گرفتنِ منبع شکست می‌خورد؛ C و بدنه هنوز آغاز نشده‌اند. خروجِ خودِ B توسط پشته ثبت نشده است.

پرسش دوم: خطای بدنه پس از سه ورود، چه ترتیبی دارد؟ پاسخ: C، سپس B و A؛ هرکدام اطلاعاتِ BodyError را می‌بیند. با سیاستِ خروجِ بدونِ خطا و False، همان شیء خطای اولیه به پوشاننده می‌رسد.

پرسش سوم: True برگشتی از callback خطا را حذف می‌کند؟ پاسخ: خیر؛ این API اطلاعاتِ استثنا را به callback نمی‌دهد و مقدارِ برگشتیِ آن برای سرکوب نیست. push با پروتکلِ خروج قراردادِ متفاوتی دارد و در شاهدِ جدا می‌تواند سرکوب کند.

پرسش چهارم: بستنِ همهٔ منابع یعنی اثرِ پردازش برگشته است؟ پاسخ: خیر؛ فقط محدودهٔ عمرِ منابعِ ثبت‌شده را آزموده‌ایم. rollback، بازیابیِ داده، پاک‌سازیِ داخلِ ورودِ نیمه‌موفق و دوامِ پس از خرابی، قراردادهای جدا می‌خواهند.

منابع و مطالعهٔ بیشتر

• Python contextlib.ExitStack؛ ورود، خروج و callback
• مرجع زبان پایتون؛ مراحل with و استثنای enter/exit

بررسی منابع: ۲۵ مهر ۱۴۰۵. تحلیل‌ها و پیشنهادهای اجرایی، جمع‌بندی نویسنده‌اند.