From b0ebe75165fb7332dbb709e4471309eb20c9abcc Mon Sep 17 00:00:00 2001 From: gabis Date: Thu, 20 Feb 2025 20:00:46 +0200 Subject: [PATCH 01/18] requirements upgrade --- requirements.txt | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/requirements.txt b/requirements.txt index 23f9461a..1bb81f90 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,8 +3,8 @@ aiohttp==3.10.2 aioquic==1.2.0 asyncstdlib==3.12.5 asyncclick==8.1.7.2 -coverage==6.5.0 -coveralls==3.3.1 +coverage==7.5.4 +coveralls==4.0.1 decoy==2.1.1 flake8==7.1.1 From a03406261c970cb3e5c9658882c0af98fe8446ec Mon Sep 17 00:00:00 2001 From: gabis Date: Thu, 20 Feb 2025 21:46:54 +0200 Subject: [PATCH 02/18] raise protocol test limit --- tests/conftest.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/conftest.py b/tests/conftest.py index fd5cddc0..883638f9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -45,7 +45,7 @@ def setup_logging(level=logging.DEBUG, use_file: bool = False): 'aiohttp' ] -if sys.version_info[:3] < (3, 11, 5): +if sys.version_info[:2] < (3, 13): tested_transports += [ 'quic', 'http3', From aa7761c8080a3eaecda51429e98486a1337b4c94 Mon Sep 17 00:00:00 2001 From: gabis Date: Thu, 20 Feb 2025 21:49:09 +0200 Subject: [PATCH 03/18] requirements upgrades --- requirements.txt | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/requirements.txt b/requirements.txt index 1bb81f90..7fe0bcdb 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,12 +1,12 @@ Rx==3.2.0 -aiohttp==3.10.2 +aiohttp==3.10.11 aioquic==1.2.0 -asyncstdlib==3.12.5 -asyncclick==8.1.7.2 +asyncstdlib==3.13.0 +asyncclick==8.1.8 coverage==7.5.4 coveralls==4.0.1 decoy==2.1.1 -flake8==7.1.1 +flake8==7.1.2 pytest-asyncio==0.25.3; python_version > "3.8" pytest-asyncio==0.23.4; python_version == "3.8" @@ -16,21 +16,21 @@ pytest-cov==4.1.0 pytest-profiling==1.8.1; python_version > "3.8" pytest-profiling==1.7.0; python_version == "3.8" -pytest-rerunfailures==13.0 +pytest-rerunfailures==15.0 pytest-timeout==2.3.1 pytest-xdist==3.6.1 pytest==8.3.4; python_version > "3.8" pytest==7.4.4; python_version == "3.8" -quart==0.19.9 +quart==0.20.0 reactivex==4.0.4 -starlette==0.40.0 +starlette==0.45.3 cbitstruct==1.1.0; python_version <= "3.12" cloudevents==1.11.0 -pydantic==1.10.18 -Werkzeug==3.0.4 -graphql-core==3.2.5 -gql==3.5.0 -websockets==13.1 +pydantic==1.10.21 +Werkzeug==3.1.3 +graphql-core==3.2.6 +gql==3.5.1 +websockets==15.0 asyncwebsockets==0.9.4 \ No newline at end of file From 76f6e2d9ea91c2c688b1a16c97225f1e0699e5ee Mon Sep 17 00:00:00 2001 From: gabis Date: Thu, 20 Feb 2025 21:50:06 +0200 Subject: [PATCH 04/18] requirements upgrades --- requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index 7fe0bcdb..4057eff7 100644 --- a/requirements.txt +++ b/requirements.txt @@ -30,7 +30,7 @@ cbitstruct==1.1.0; python_version <= "3.12" cloudevents==1.11.0 pydantic==1.10.21 Werkzeug==3.1.3 -graphql-core==3.2.6 +graphql-core==3.2.3 gql==3.5.1 websockets==15.0 asyncwebsockets==0.9.4 \ No newline at end of file From eb224bd8566c31f4fbab07322e30d9ee8b3da872 Mon Sep 17 00:00:00 2001 From: gabis Date: Thu, 20 Feb 2025 21:52:05 +0200 Subject: [PATCH 05/18] requirements upgrades --- requirements.txt | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index 4057eff7..f0fb6690 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,7 +2,10 @@ Rx==3.2.0 aiohttp==3.10.11 aioquic==1.2.0 asyncstdlib==3.13.0 -asyncclick==8.1.8 + +asyncclick==8.1.8; python_version > "3.8" +asyncclick==8.1.7.2; python_version == "3.8" + coverage==7.5.4 coveralls==4.0.1 decoy==2.1.1 From 3e935dd0d893d0d6fd44587e6ca55ce7d068cef5 Mon Sep 17 00:00:00 2001 From: gabis Date: Fri, 21 Feb 2025 15:09:50 +0200 Subject: [PATCH 06/18] requirements upgrades --- requirements.txt | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index f0fb6690..ad5be905 100644 --- a/requirements.txt +++ b/requirements.txt @@ -19,7 +19,9 @@ pytest-cov==4.1.0 pytest-profiling==1.8.1; python_version > "3.8" pytest-profiling==1.7.0; python_version == "3.8" -pytest-rerunfailures==15.0 +pytest-rerunfailures==15.0; python_version > "3.8" +pytest-rerunfailures==14.0; python_version == "3.8" + pytest-timeout==2.3.1 pytest-xdist==3.6.1 From 17e0a91cfa97ecc672649bd7225f975fff8079b6 Mon Sep 17 00:00:00 2001 From: gabis Date: Fri, 21 Feb 2025 15:12:26 +0200 Subject: [PATCH 07/18] requirements upgrades --- requirements.txt | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index ad5be905..149e2942 100644 --- a/requirements.txt +++ b/requirements.txt @@ -28,7 +28,9 @@ pytest-xdist==3.6.1 pytest==8.3.4; python_version > "3.8" pytest==7.4.4; python_version == "3.8" -quart==0.20.0 +quart==0.20.0; python_version > "3.8" +quart==0.19.9; python_version == "3.8" + reactivex==4.0.4 starlette==0.45.3 cbitstruct==1.1.0; python_version <= "3.12" From c121f3bac3adab1639eeb6f225560a3c306bea02 Mon Sep 17 00:00:00 2001 From: gabis Date: Fri, 21 Feb 2025 15:13:29 +0200 Subject: [PATCH 08/18] requirements upgrades --- requirements.txt | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index 149e2942..7b844f41 100644 --- a/requirements.txt +++ b/requirements.txt @@ -32,7 +32,10 @@ quart==0.20.0; python_version > "3.8" quart==0.19.9; python_version == "3.8" reactivex==4.0.4 -starlette==0.45.3 + +starlette==0.45.3; python_version > "3.8" +starlette==0.44.0; python_version == "3.8" + cbitstruct==1.1.0; python_version <= "3.12" cloudevents==1.11.0 pydantic==1.10.21 From cde0382243224e82652ea131c6266976e7676c49 Mon Sep 17 00:00:00 2001 From: gabis Date: Fri, 21 Feb 2025 15:14:38 +0200 Subject: [PATCH 09/18] requirements upgrades --- requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index 7b844f41..5ea9ace7 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ Rx==3.2.0 -aiohttp==3.10.11 +aiohttp==3.11.12 aioquic==1.2.0 asyncstdlib==3.13.0 From 73fa3aef81d14237d67b342b9269d90c895ae1cc Mon Sep 17 00:00:00 2001 From: gabis Date: Fri, 21 Feb 2025 15:15:28 +0200 Subject: [PATCH 10/18] requirements upgrades --- requirements.txt | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index 5ea9ace7..20d36787 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,8 @@ Rx==3.2.0 -aiohttp==3.11.12 + +aiohttp==3.11.12; python_version > "3.8" +aiohttp==3.10.11; python_version == "3.8" + aioquic==1.2.0 asyncstdlib==3.13.0 From 799526ba48c5e4f9ac5bcf49b3c0c2a8934dd843 Mon Sep 17 00:00:00 2001 From: gabis Date: Fri, 21 Feb 2025 15:16:14 +0200 Subject: [PATCH 11/18] requirements upgrades --- requirements.txt | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index 20d36787..45fc1d96 100644 --- a/requirements.txt +++ b/requirements.txt @@ -42,7 +42,10 @@ starlette==0.44.0; python_version == "3.8" cbitstruct==1.1.0; python_version <= "3.12" cloudevents==1.11.0 pydantic==1.10.21 -Werkzeug==3.1.3 + +Werkzeug==3.1.3; python_version > "3.8" +Werkzeug==3.0.6; python_version == "3.8" + graphql-core==3.2.3 gql==3.5.1 websockets==15.0 From f6117f7f3baa126b3e2985aba7c71aba2aac81f9 Mon Sep 17 00:00:00 2001 From: gabis Date: Fri, 21 Feb 2025 15:24:10 +0200 Subject: [PATCH 12/18] requirements upgrades --- requirements.txt | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index 45fc1d96..0d8b488d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -48,5 +48,8 @@ Werkzeug==3.0.6; python_version == "3.8" graphql-core==3.2.3 gql==3.5.1 -websockets==15.0 + +websockets==15.0; python_version > "3.8" +websockets==13.1; python_version == "3.8" + asyncwebsockets==0.9.4 \ No newline at end of file From 77cbd869e6808859c172796a2a856d8df9b62f5c Mon Sep 17 00:00:00 2001 From: gabis Date: Sat, 22 Feb 2025 15:02:56 +0200 Subject: [PATCH 13/18] logging and dependency fixes --- requirements.txt | 4 ++-- rsocket/transports/aioquic_transport.py | 5 +++++ tests/rsocket/test_concurrency.py | 3 +++ tests/tools/fixtures_aioquic.py | 10 ++++++++++ 4 files changed, 20 insertions(+), 2 deletions(-) diff --git a/requirements.txt b/requirements.txt index 0d8b488d..7ec6a707 100644 --- a/requirements.txt +++ b/requirements.txt @@ -9,8 +9,8 @@ asyncstdlib==3.13.0 asyncclick==8.1.8; python_version > "3.8" asyncclick==8.1.7.2; python_version == "3.8" -coverage==7.5.4 -coveralls==4.0.1 +coverage==6.5.0 +coveralls==3.3.1 decoy==2.1.1 flake8==7.1.2 diff --git a/rsocket/transports/aioquic_transport.py b/rsocket/transports/aioquic_transport.py index bee8de7c..7b50704e 100644 --- a/rsocket/transports/aioquic_transport.py +++ b/rsocket/transports/aioquic_transport.py @@ -102,6 +102,8 @@ async def send_frame(self, frame: Frame): async def incoming_data_listener(self): try: + logger().debug('Quic - Listener started') + await self._quic_protocol.wait_connected() while True: @@ -118,8 +120,11 @@ async def incoming_data_listener(self): logger().debug('Asyncio task canceled: incoming_data_listener') except Exception: self._incoming_frame_queue.put_nowait(RSocketTransportError()) + finally: + logger().debug('Quic - Listener stopped') async def close(self): + logger().debug('Quic - Closing transport') await cancel_if_task_exists(self._listener) self._quic_protocol.close() diff --git a/tests/rsocket/test_concurrency.py b/tests/rsocket/test_concurrency.py index 7a3d2893..813ce4de 100644 --- a/tests/rsocket/test_concurrency.py +++ b/tests/rsocket/test_concurrency.py @@ -1,4 +1,5 @@ import asyncio +import logging from typing import Tuple, Optional import pytest @@ -76,6 +77,8 @@ async def run(): request_2 = asyncio.create_task(measure_time(client.request_response(Payload(b'10')))) return await request_1, await request_2 + logging.debug("Starting concurrent requests") + measure_result = await measure_time(run()) results = measure_result.result diff --git a/tests/tools/fixtures_aioquic.py b/tests/tools/fixtures_aioquic.py index 34f56032..2310b272 100644 --- a/tests/tools/fixtures_aioquic.py +++ b/tests/tools/fixtures_aioquic.py @@ -1,3 +1,4 @@ +import logging from asyncio import Event from contextlib import asynccontextmanager from typing import Optional @@ -34,12 +35,16 @@ def store_server(new_server): server = new_server wait_for_server.set() + logging.debug('test quic - starting server') + quic_server = await rsocket_serve(host='localhost', port=unused_tcp_port, configuration=server_configuration, on_server_create=store_server, **(server_arguments or {})) + logging.debug('test quic - server started') + try: # from datetime import timedelta # test_overrides = {'keep_alive_period': timedelta(minutes=20)} @@ -49,7 +54,12 @@ def store_server(new_server): configuration=quic_client_configuration(certificate)) as transport: async with RSocketClient(single_transport_provider(transport), **client_arguments) as client: + logging.debug('test quic - waiting for server to be ready') + await wait_for_server.wait() + + logging.debug('test quic - server and client ready, starting test') + yield server, client finally: if server is not None: From 07843fd2f9fafaae85aed64a6c39f5d65451a2d7 Mon Sep 17 00:00:00 2001 From: gabis Date: Sat, 22 Feb 2025 15:14:08 +0200 Subject: [PATCH 14/18] server setup unit test --- tests/rsocket/test_routing.py | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/tests/rsocket/test_routing.py b/tests/rsocket/test_routing.py index 32f94900..d452c5c1 100644 --- a/tests/rsocket/test_routing.py +++ b/tests/rsocket/test_routing.py @@ -5,12 +5,14 @@ from reactivestreams.subscriber import DefaultSubscriber from rsocket.awaitable.awaitable_rsocket import AwaitableRSocket +from rsocket.error_codes import ErrorCode from rsocket.extensions.authentication import Authentication, AuthenticationSimple from rsocket.extensions.composite_metadata import CompositeMetadata from rsocket.extensions.helpers import route, composite, authenticate_simple from rsocket.extensions.mimetypes import WellKnownMimeTypes from rsocket.helpers import create_response from rsocket.payload import Payload +from rsocket.request_handler import BaseRequestHandler from rsocket.routing.request_router import RequestRouter from rsocket.routing.routing_request_handler import RoutingRequestHandler from rsocket.rx_support.rx_rsocket import RxRSocket @@ -357,3 +359,33 @@ def handler_factory(): 'pass')))) assert result.data == b'result' + + +@pytest.mark.allow_error_log(regex_filter='(RSocket error REJECTED_SETUP|Setup error)') +async def test_invalid_metadata_for_routing(lazy_pipe): + router = RequestRouter() + + async def authenticate(path: str, authentication: Authentication): + if not isinstance(authentication, AuthenticationSimple) or authentication.password != b'pass': + raise Exception('Invalid credentials') + + error_wait = asyncio.Event() + + def client_handler_factory(): + class ClientHandler(BaseRequestHandler): + async def on_error(self, error_code: ErrorCode, payload: Payload): + error_wait.set() + + return ClientHandler() + + @router.response('test.path') + async def response(): + return create_response(b'result') + + def handler_factory(): + return RoutingRequestHandler(router, authentication_verifier=authenticate) + + async with lazy_pipe( + client_arguments={'handler_factory': client_handler_factory}, + server_arguments={'handler_factory': handler_factory}) as (server, client): + await error_wait.wait() From b4959aecc9da7c950927aa33037353a0dd210d67 Mon Sep 17 00:00:00 2001 From: gabis Date: Sat, 22 Feb 2025 15:37:20 +0200 Subject: [PATCH 15/18] revert protocol test for python versions limitation --- tests/conftest.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/conftest.py b/tests/conftest.py index 883638f9..07d5544c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -45,7 +45,7 @@ def setup_logging(level=logging.DEBUG, use_file: bool = False): 'aiohttp' ] -if sys.version_info[:2] < (3, 13): +if sys.version_info[:2] < (3, 11, 5): tested_transports += [ 'quic', 'http3', From 37b479fd92bc5ab98d134b01d27a3d0ac57f53e6 Mon Sep 17 00:00:00 2001 From: gabis Date: Sat, 22 Feb 2025 16:00:09 +0200 Subject: [PATCH 16/18] revert protocol test for python versions limitation --- tests/conftest.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/conftest.py b/tests/conftest.py index 07d5544c..fd5cddc0 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -45,7 +45,7 @@ def setup_logging(level=logging.DEBUG, use_file: bool = False): 'aiohttp' ] -if sys.version_info[:2] < (3, 11, 5): +if sys.version_info[:3] < (3, 11, 5): tested_transports += [ 'quic', 'http3', From d14710bc115a5aaa48a3de258488ac16dbab3ff5 Mon Sep 17 00:00:00 2001 From: gabis Date: Sat, 22 Feb 2025 16:25:19 +0200 Subject: [PATCH 17/18] fix graphql send cancel on break from subscription loop --- rsocket/graphql/rsocket_transport.py | 19 ++++++++++++- tests/test_integrations/test_graphql.py | 38 ++++++++++++++++++++++++- 2 files changed, 55 insertions(+), 2 deletions(-) diff --git a/rsocket/graphql/rsocket_transport.py b/rsocket/graphql/rsocket_transport.py index 407a013f..e892c765 100644 --- a/rsocket/graphql/rsocket_transport.py +++ b/rsocket/graphql/rsocket_transport.py @@ -9,6 +9,7 @@ from reactivestreams.subscriber import DefaultSubscriber from rsocket.extensions.helpers import composite, route from rsocket.frame_helpers import str_to_bytes +from rsocket.logger import logger from rsocket.payload import Payload from rsocket.rsocket_client import RSocketClient @@ -104,6 +105,14 @@ def on_complete(self): def on_error(self, exception: Exception): self._received_queue.put_nowait(exception) + def cancel(self): + if self.subscription is not None: + self.subscription.cancel() + + def request(self, n: int): + if self.subscription is not None: + self.subscription.request(n) + rsocket_payload = self._create_rsocket_payload(document, variable_values, operation_name) received_queue = Queue() @@ -120,4 +129,12 @@ def on_error(self, exception: Exception): if response is complete_object: break - yield self._response_to_execution_result(response) + execution_result = self._response_to_execution_result(response) + + try: + yield execution_result + except GeneratorExit: + logger().debug('Generator exited') + subscriber.cancel() + return + diff --git a/tests/test_integrations/test_graphql.py b/tests/test_integrations/test_graphql.py index 0bfd2771..6aff5327 100644 --- a/tests/test_integrations/test_graphql.py +++ b/tests/test_integrations/test_graphql.py @@ -1,3 +1,5 @@ +import asyncio + from gql import Client, gql from rsocket.extensions.mimetypes import WellKnownMimeTypes @@ -75,4 +77,38 @@ def handler_factory(): gql("""{__schema { types { name } } }"""), get_execution_result=True) - assert response.data == expected_schema \ No newline at end of file + assert response.data == expected_schema + + +async def test_graphql_break_loop(lazy_pipe, graphql_schema): + def handler_factory(): + return RoutingRequestHandler(graphql_handler(graphql_schema, 'graphql')) + + async with lazy_pipe( + client_arguments={'metadata_encoding': WellKnownMimeTypes.MESSAGE_RSOCKET_COMPOSITE_METADATA}, + server_arguments={'handler_factory': handler_factory}) as (server, client): + graphql = Client( + schema=graphql_schema, + transport=RSocketTransport(client), + ) + + responses = [] + i = 0 + async for response in graphql.subscribe_async( + document=gql(""" + subscription { + greetings {message} + } + """), + get_execution_result=True): + responses.append(response.data) + i += 1 + if i > 4: + break + + assert len(responses) == 5 + assert responses[0] == {'greetings': {'message': 'Hello world 0'}} + + await asyncio.sleep(1) + + # assert responses[9] == {'greetings': {'message': 'Hello world 9'}} From aa6a73d7117701624aab0cb6a3115e99d0f4290e Mon Sep 17 00:00:00 2001 From: gabis Date: Sat, 22 Feb 2025 16:27:46 +0200 Subject: [PATCH 18/18] fix graphql send cancel on break from subscription loop --- rsocket/graphql/rsocket_transport.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/rsocket/graphql/rsocket_transport.py b/rsocket/graphql/rsocket_transport.py index e892c765..19188760 100644 --- a/rsocket/graphql/rsocket_transport.py +++ b/rsocket/graphql/rsocket_transport.py @@ -121,17 +121,17 @@ def request(self, n: int): self._rsocket_client.request_stream(rsocket_payload).subscribe(subscriber) while True: - response = await received_queue.get() + try: + response = await received_queue.get() - if isinstance(response, Exception): - raise response + if isinstance(response, Exception): + raise response - if response is complete_object: - break + if response is complete_object: + break - execution_result = self._response_to_execution_result(response) + execution_result = self._response_to_execution_result(response) - try: yield execution_result except GeneratorExit: logger().debug('Generator exited')