|
| 1 | +# Part of the EDG Compiler Project, under the Apache License v2.0 with LLVM |
| 2 | +# Exceptions. |
| 3 | +# See https://edgcpp.org/LICENSE.txt for license information. |
| 4 | +# SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception |
| 5 | + |
| 6 | +'''Cancelling AcknowlEDG client requests undoes them, as failing does.''' |
| 7 | + |
| 8 | +import asyncio |
| 9 | +import uuid |
| 10 | + |
| 11 | +from typing import Any |
| 12 | + |
| 13 | +from edgacknowledg.client import ( |
| 14 | + BatchClientSession, |
| 15 | + _BatchClientReview, |
| 16 | + _DirectClientReview, |
| 17 | +) |
| 18 | + |
| 19 | + |
| 20 | +class _Unanswered: |
| 21 | + '''A batch connection the server never answers on.''' |
| 22 | + |
| 23 | + def __init__(self) -> None: |
| 24 | + self.sent: list[Any] = [] |
| 25 | + |
| 26 | + async def send(self, data: Any) -> None: |
| 27 | + self.sent.append(data) |
| 28 | + |
| 29 | + |
| 30 | +class _Handler: |
| 31 | + subscription_type = 'test.src' |
| 32 | + slug = 'test-changes' |
| 33 | + |
| 34 | + |
| 35 | +def _session() -> BatchClientSession: |
| 36 | + session = BatchClientSession( |
| 37 | + status_listener = None, |
| 38 | + api_host = '127.0.0.1', |
| 39 | + api_port = 1, |
| 40 | + api_key = None, |
| 41 | + open_timeout = None |
| 42 | + ) |
| 43 | + session._connection = _Unanswered() # type: ignore[assignment] |
| 44 | + session._session_ready.set() |
| 45 | + return session |
| 46 | + |
| 47 | + |
| 48 | +async def _cancel_once_sent(session: BatchClientSession, |
| 49 | + request: 'asyncio.Task[Any]') -> None: |
| 50 | + connection = session._connection |
| 51 | + assert isinstance(connection, _Unanswered) |
| 52 | + while len(connection.sent) == 0: |
| 53 | + await asyncio.sleep(0.01) |
| 54 | + request.cancel() |
| 55 | + await asyncio.gather(request, return_exceptions = True) |
| 56 | + assert request.cancelled() |
| 57 | + |
| 58 | + |
| 59 | +async def test_cancelled_subscribes_are_not_awaited() -> None: |
| 60 | + session = _session() |
| 61 | + subscribing = asyncio.create_task(session._subscribe({ |
| 62 | + 'id': str(uuid.uuid4()), 'type': 'test.src', 'slug': 'test-changes', |
| 63 | + })) |
| 64 | + await _cancel_once_sent(session, subscribing) |
| 65 | + assert session._pending_subscribe == {} |
| 66 | + |
| 67 | + unsubscribing = asyncio.create_task(session._unsubscribe(uuid.uuid4())) |
| 68 | + session._connection = _Unanswered() # type: ignore[assignment] |
| 69 | + await _cancel_once_sent(session, unsubscribing) |
| 70 | + assert session._pending_unsubscribe == {} |
| 71 | + |
| 72 | + |
| 73 | +async def test_cancelled_batch_sections_are_not_added() -> None: |
| 74 | + session = _session() |
| 75 | + review = _BatchClientReview(session, uuid.uuid4(), [], '') |
| 76 | + adding = asyncio.create_task(review.add_section(_Handler())) # type: ignore |
| 77 | + await _cancel_once_sent(session, adding) |
| 78 | + # Not left to be subscribed on reconnecting. |
| 79 | + assert review.sections() == [] |
| 80 | + assert session._pending_subscribe == {} |
| 81 | + |
| 82 | + |
| 83 | +async def test_cancelled_direct_sections_are_not_added() -> None: |
| 84 | + review = _DirectClientReview(None, uuid.uuid4(), [], '') # type: ignore |
| 85 | + subscribing = asyncio.Event() |
| 86 | + |
| 87 | + async def subscribe_section(handler: Any) -> None: |
| 88 | + subscribing.set() |
| 89 | + await asyncio.Event().wait() |
| 90 | + |
| 91 | + review.subscribe_section = subscribe_section # type: ignore[method-assign] |
| 92 | + adding = asyncio.create_task(review.add_section(_Handler())) # type: ignore |
| 93 | + await subscribing.wait() |
| 94 | + adding.cancel() |
| 95 | + await asyncio.gather(adding, return_exceptions = True) |
| 96 | + assert review.sections() == [] |
0 commit comments