Asyncio: проверка реакции задачи на cancel() с таймаутом

Аттестация отмены корутин

Инспектор тестовой платформы проверяет, насколько корректно корутины подключаемых плагинов реагируют на отмену выполнения. Для проверки инспектор запускает задачу плагина, дожидается её входа в рабочий цикл через переключение контекста, отправляет запрос task.cancel() и ограничивает время ожидания реакции с помощью asyncio.timeout.

Если задача успевает завершиться до истечения таймаута, инспектор регистрирует её фактический исход: штатный возврат значения, подтверждённое исключение CancelledError или возникшую прикладную ошибку. Однако недобросовестный плагин может подавить отмену и продолжить работу. Если срок ожидания истёк, а задача осталась незавершённой со статусом done() is False, инспектор не пытается её принудительно уничтожить, а передаёт саму работающую задачу вместе с флагом зависания внешнему стенду для перевода цикла событий в режим изоляции.

Реализуйте функцию аттестации корутины, которая определяет тип завершения задачи после сигнала отмены либо фиксирует её зависание за пределами таймаута с сохранением активного объекта задачи.

Что уже дано

В среде выполнения доступна вспомогательная корутина wait_for_task_completion(tasks), которая пассивно ожидает завершения набора задач без отмены или изменения их состояния. Также тестовое окружение предоставляет фикстуры make_attestation_probe, make_attestation_example_probe, observe_attestation и observe_prepared_attestation для запуска сценариев аттестации.

Вам необходимо реализовать асинхронную функцию attest_cancellation.

Что нужно сделать

Реализуйте корутину attest_cancellation(plugin_coro, timeout_seconds).

Функция принимает два аргумента:

• plugin_coro — однократный объект корутины подключаемого плагина;
• timeout_seconds — положительное число секунд ожидания (от 0.01 до 0.1 с).

Порядок выполнения аттестации:

1. Оберните корутину plugin_coro в отдельную нативную задачу asyncio.Task.
2. Передайте управление циклу событий (например, через await asyncio.sleep(0)), чтобы задача плагина успела начать выполнение.
3. Запросите отмену созданной задачи вызовом task.cancel() (учитывайте, что задача могла завершиться уже на предыдущем шаге).
4. Ожидайте реакцию задачи на отмену в рамках контекста asyncio.timeout(timeout_seconds).

Функция должна вернуть словарь ровно с пятью ключами: {"kind", "value", "error", "task", "hung"}:

• "returned" — задача штатно завершилась возвратом значения (включая None, False, пустые списки или словари): kind="returned", value содержит возвращённое значение без изменений, error=None, task=None, hung=False.
• "cancelled" — задача подтвердила отмену через asyncio.CancelledError (или его подкласс): kind="cancelled", value=None, error=None, task=None, hung=False.
• "error" — задача завершилась прикладным исключением из семейства Exception (включая прикладные TimeoutError плагина): kind="error", error содержит исходный объект исключения плагина, value=None, task=None, hung=False.
• "pending" — время ожидания asyncio.timeout истекло, а задача плагина так и не перешла в состояние done() is True: kind="pending", task содержит сам живой объект задачи плагина, value=None, error=None, hung=True. Не уничтожайте и не отменяйте повторно зависшую задачу.

Прикладное исключение TimeoutError, возникшее внутри самого плагина до истечения внешнего лимита времени, классифицируется как "error", а не как зависание "pending".

Ввод и вывод

Стандартные потоки ввода stdin и вывода stdout в решении не используются: считывать ввод и вызывать print() не нужно. После выполнения кода учащегося проверяющая система считывает из пространства имён модуля функцию верхнего уровня attest_cancellation и вызывает её с проверочными аргументами.

В примере ниже корутина плагина ожидает событие, перехватывает доставленный asyncio.CancelledError и возвращает число 7. Так как задача завершилась штатным возвратом до истечения лимита в 0.04 секунды, итоговый отчёт содержит статус 'returned', само значение 7, пустые поля ошибки и задачи, а флаг зависания равен False.

Пример использования

import asyncio


async def plugin():
    try:
        await asyncio.Event().wait()
    except asyncio.CancelledError:
        return 7


report = asyncio.run(attest_cancellation(plugin(), 0.04))
print(report['kind'], report['value'], report['error'], report['task'], report['hung'])

Вывод:

returned 7 None None False