18095678244

Committed 29 Sep 2025 11:38AM UTC coverage: 86.637% (-0.06%) from 86.697%

Build # 18095678244

Build Type

push

github

Committed by

web-flow

Commit Message

Fix entrypoint for console script (#336)

* Fix entrypoint

* Add asyncclick import check for console script

* Make console script exit if required imports aren't found

---------

Co-authored-by: jell-o-fishi <111562631+jell-o-fishi@users.noreply.github.com>

Coverage Stats

780 of 978 branches covered (79.75%)

Branch coverage included in aggregate %.

3 of 7 new or added lines in 1 file covered. (42.86%)

3849 of 4365 relevant lines covered (88.18%)

2.63 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.71

/rsocket/cli/command.py

import asyncio
import logging
import ssl
import sys
from contextlib import asynccontextmanager
from dataclasses import dataclass
from enum import Enum, unique
from importlib.metadata import version as get_version
from typing import Optional, Type, Collection, List, Callable

try:
    import asyncclick as click
except ImportError:
    print("This command requires the CLI extra.")
    print("Install with: pip install rsocket[cli]")
    sys.exit(1)

from rsocket.awaitable.awaitable_rsocket import AwaitableRSocket
from rsocket.extensions.helpers import route, composite, authenticate_simple, authenticate_bearer
from rsocket.extensions.mimetypes import WellKnownMimeTypes
from rsocket.frame import MAX_REQUEST_N
from rsocket.frame_helpers import ensure_bytes, safe_len
from rsocket.helpers import single_transport_provider
from rsocket.payload import Payload
from rsocket.rsocket_client import RSocketClient
from rsocket.transports.abstract_messaging import AbstractMessagingTransport

from rsocket.transports.tcp import TransportTCP


@unique
class RequestType(Enum):
    response = 'REQUEST_RESPONSE'
    stream = 'REQUEST_STREAM'
    channel = 'REQUEST_CHANNEL'
    fnf = 'FIRE_AND_FORGET'
    metadata_push = 'METADATA_PUSH'


interaction_models: List[str] = [str(e.value) for e in RequestType]


@dataclass(frozen=True)
class RSocketUri:
    host: str
    port: int
    schema: str
    path: Optional[str] = None
    original_uri: Optional[str] = None


def parse_uri(uri: str) -> RSocketUri:
    schema, rest = uri.split(':', 1)
    rest = rest.strip('/')
    host_port_path = rest.split('/', 1)
    host_port = host_port_path[0].split(':')
    host = host_port[0]

    if len(host_port) == 1:
        port = None
    else:
        port = int(host_port[1])

    if len(host_port_path) > 1:
        rest = host_port_path[1]
    else:
        rest = None

    return RSocketUri(host, port, schema, rest, uri)


@asynccontextmanager
async def transport_from_uri(uri: RSocketUri,
                             verify_ssl=True,
                             headers: Optional = None,
                             trust_cert: Optional[str] = None) -> Type[AbstractMessagingTransport]:
    if uri.schema == 'tcp':
        connection = await asyncio.open_connection(uri.host, uri.port)
        yield TransportTCP(*connection)
    elif uri.schema in ['wss', 'ws']:
        import aiohttp
        async with aiohttp.ClientSession() as session:
            if trust_cert is not None:
                ssl_context = ssl.create_default_context(cafile=trust_cert)
            else:
                ssl_context = None

            async with session.ws_connect(uri.original_uri,
                                          verify_ssl=verify_ssl,
                                          ssl_context=ssl_context,
                                          headers=headers) as websocket:
                from rsocket.transports.aiohttp_websocket import TransportAioHttpClient

                yield TransportAioHttpClient(websocket=websocket)
    else:
        raise Exception('Unsupported schema in CLI')


def build_composite_metadata(auth_simple: Optional[str],
                             route_value: Optional[str],
                             auth_bearer: Optional[str]) -> List:
    composite_items = []

    if route_value is not None:
        composite_items.append(route(route_value))

    if auth_simple is not None and auth_bearer is not None:
        raise click.UsageError('Multiple authentication methods specified.')

    if auth_simple is not None:
        composite_items.append(authenticate_simple(*auth_simple.split(':')))

    if auth_bearer is not None:
        composite_items.append(authenticate_bearer(auth_bearer))

    return composite_items


@asynccontextmanager
async def create_client(parsed_uri,
                        data_mime_type,
                        metadata_mime_type,
                        setup_payload,
                        allow_untrusted_ssl=False,
                        http_headers=None,
                        trust_cert=None):
    async with transport_from_uri(parsed_uri,
                                  verify_ssl=not allow_untrusted_ssl,
                                  headers=http_headers,
                                  trust_cert=trust_cert) as transport:
        async with RSocketClient(single_transport_provider(transport),
                                 data_encoding=data_mime_type or WellKnownMimeTypes.APPLICATION_JSON,
                                 metadata_encoding=metadata_mime_type or WellKnownMimeTypes.APPLICATION_JSON,
                                 setup_payload=setup_payload) as client:
            yield AwaitableRSocket(client)


def get_request_type(request: bool,
                     stream: bool,
                     fnf: bool,
                     metadata_push: bool,
                     channel: bool,
                     interaction_model: Optional[str]) -> RequestType:
    interaction_options = list(filter(lambda _: _ is True, [request, stream, fnf, channel, metadata_push]))

    if len(interaction_options) >= 2 or (len(interaction_options) >= 1 and interaction_model is not None):
        raise click.UsageError('Multiple interaction methods specified.')

    if interaction_model is not None:
        return RequestType(interaction_model.upper())
    if request:
        return RequestType.response
    if stream:
        return RequestType.stream
    if channel:
        return RequestType.channel
    if fnf:
        return RequestType.fnf
    if metadata_push:
        return RequestType.metadata_push

    raise click.UsageError('No interaction method specified (eg. --request)')


@click.command(name='rsocket-py', help='Supported connection strings: tcp/ws/wss')
@click.option('--im', '--interactionModel', 'interaction_model', is_flag=False,
              type=click.Choice(interaction_models, case_sensitive=False),
              help='Interaction Model')
@click.option('--request', is_flag=True,
              help='Request response')
@click.option('--stream', is_flag=True,
              help='Request stream')
@click.option('--channel', is_flag=True,
              help='Request channel')
@click.option('--fnf', is_flag=True,
              help='Fire and Forget')
@click.option('--metadataPush', 'metadata_push', is_flag=True,
              help='Metadata Push')
@click.option('-d', '--data', '--input', 'data', is_flag=False,
              help='Data. Use "-" to read data from standard input. (default: )')
@click.option('-l', '--load', is_flag=False,
              help='Load a file as Data. (e.g. ./foo.txt, /tmp/foo.txt)')
@click.option('-m', '--metadata', is_flag=False, default=None,
              help='Metadata (default: )')
@click.option('-r', '--route', 'route_value', is_flag=False, default=None,
              help='Enable Routing Metadata Extension')
@click.option('--limitRate', 'limit_rate', is_flag=False, default=None, type=int,
              help='Enable limitRate(rate)')
@click.option('--take', 'take_n', is_flag=False, default=None, type=int,
              help='Enable take(n)')
@click.option('-u', '--as', '--authSimple', 'auth_simple', is_flag=False, default=None,
              help='Enable Authentication Metadata Extension (Simple). The format must be "username: password"')
@click.option('--sd', '--setup', '--setupData', 'setup_data', is_flag=False, default=None,
              help='Data for Setup payload')
@click.option('--sm', '--setupMetadata', 'setup_metadata', is_flag=False, default=None,
              help='Metadata for Setup payload')
@click.option('--ab', '--authBearer', 'auth_bearer', is_flag=False, default=None,
              help='Enable Authentication Metadata Extension (Bearer)')
@click.option('--dataMimeType', '--dataFormat', '--dmt', 'data_mime_type', is_flag=False,
              help='MimeType for data (default: application/json)')
@click.option('--metadataMimeType', '--metadataFormat', '--mmt', 'metadata_mime_type', is_flag=False,
              help='MimeType for metadata (default:application/json)')
@click.option('--allowUntrustedSsl', 'allow_untrusted_ssl', is_flag=True, default=False,
              help='Do not verify SSL certificate (for wss:// urls)')
@click.option('-H', '--header', '--httpHeader', 'http_header', multiple=True,
              help='ws/wss headers')
@click.option('--trustCert', 'trust_cert', is_flag=False,
              help='PEM file for a trusted certificate. (e.g. ./foo.crt, /tmp/foo.crt)')
@click.option('--debug', is_flag=True,
              help='Show debug log')
@click.option('--quiet', '-q', is_flag=True,
              help='Disable the output on next')
@click.option('--timeout', 'timeout_seconds', is_flag=False, type=int,
              help='Timeout in seconds')
@click.option('--version', is_flag=True,
              help='Print version')
@click.argument('uri', required=False)
@click.pass_context
async def command(context, data, load,
                  metadata, route_value, auth_simple, auth_bearer,
                  limit_rate, take_n, allow_untrusted_ssl,
                  setup_data, setup_metadata, interaction_model,
                  http_header, metadata_push, timeout_seconds,
                  data_mime_type, metadata_mime_type,
                  request, stream, channel, fnf, trust_cert,
                  uri, debug, version, quiet):
    if version:
        try:
            print(get_version('rsocket'))
        except Exception:
            print('Failed to find version')
        return

    if uri is None:
        raise click.MissingParameter(param=context.command.params[-1])

    if quiet:
        logging.basicConfig(handlers=[])

    if debug:
        logging.basicConfig(level=logging.DEBUG)

    if take_n == 0:
        return

    request_type = get_request_type(request, stream, fnf, metadata_push, channel, interaction_model)
    http_headers = parse_headers(http_header)
    composite_items = build_composite_metadata(auth_simple, route_value, auth_bearer)
    setup_payload = create_setup_payload(setup_data, setup_metadata)
    metadata_value = get_metadata_value(composite_items, metadata)
    metadata_mime_type = normalize_metadata_mime_type(composite_items, metadata_mime_type)
    parsed_uri = parse_uri(uri)

    def payload_provider():
        return create_request_payload(data, load, metadata_value)

    future = run_request(request_type, limit_rate, payload_provider,
                         http_headers=http_headers,
                         allow_untrusted_ssl=allow_untrusted_ssl,
                         metadata_mime_type=metadata_mime_type,
                         data_mime_type=data_mime_type,
                         setup_payload=setup_payload,
                         trust_cert=trust_cert,
                         parsed_uri=parsed_uri)

    if timeout_seconds is not None:
        result = await asyncio.wait_for(future, timeout_seconds)
    else:
        result = await future

    if not quiet:
        output_result(result)


async def run_request(request_type: RequestType,
                      limit_rate: Optional[int],
                      payload_provider: Callable[[], Payload],
                      **kwargs):
    async with create_client(**kwargs) as client:
        return await execute_request(client,
                                     request_type,
                                     normalize_limit_rate(limit_rate),
                                     payload_provider())


def parse_headers(http_headers):
    if safe_len(http_headers) > 0:
        headers = dict()

        for header in http_headers:
            parts = header.split('=', 2)
            headers[parts[0]] = parts[1]

        return headers

    return None


def normalize_metadata_mime_type(composite_items: List, metadata_mime_type):
    if len(composite_items) > 0:
        metadata_mime_type = WellKnownMimeTypes.MESSAGE_RSOCKET_COMPOSITE_METADATA

    return metadata_mime_type


def create_request_payload(data: Optional[str],
                           load: Optional[str],
                           metadata: Optional[bytes]) -> Payload:
    data = normalize_data(data, load)

    return Payload(data, metadata)


def output_result(result):
    if isinstance(result, Payload):
        print(result.data.decode('utf-8'))
    elif isinstance(result, Collection):
        print([p.data.decode('utf-8') for p in result])


async def execute_request(awaitable_client: AwaitableRSocket,
                          request_type: RequestType,
                          limit_rate: int,
                          payload: Payload):
    result = None

    if request_type is RequestType.response:
        result = await awaitable_client.request_response(payload)
    elif request_type is RequestType.stream:
        result = await awaitable_client.request_stream(payload, limit_rate=limit_rate)
    elif request_type is RequestType.channel:
        result = await awaitable_client.request_channel(payload, limit_rate=limit_rate)
    elif request_type is RequestType.fnf:
        await awaitable_client.fire_and_forget(payload)
    elif request_type is RequestType.metadata_push:
        await awaitable_client.metadata_push(payload.metadata)

    return result


def get_metadata_value(composite_items: List, metadata: Optional[str]) -> bytes:
    if len(composite_items) > 0:
        metadata_value = composite(*composite_items)
    else:
        metadata_value = metadata

    return ensure_bytes(metadata_value)


def create_setup_payload(setup_data: Optional[str], setup_metadata: Optional[str]) -> Optional[Payload]:
    setup_payload = None

    if setup_data is not None or setup_metadata is not None:
        setup_payload = Payload(
            ensure_bytes(setup_data),
            ensure_bytes(setup_metadata)
        )

    return setup_payload


def normalize_data(data: Optional[str], load: Optional[str]) -> bytes:
    if data == '-':
        stdin_text = click.get_text_stream('stdin')
        return ensure_bytes(stdin_text.read())

    if load is not None:
        with open(load) as fd:
            return ensure_bytes(fd.read())

    return ensure_bytes(data)


def normalize_limit_rate(limit_rate):
    if limit_rate is None or limit_rate <= 0:
        limit_rate = MAX_REQUEST_N

    return limit_rate


if __name__ == '__main__':
    command()

1	import asyncio	3✔
2	import logging	3✔
3	import ssl	3✔
4	import sys	3✔
5	from contextlib import asynccontextmanager	3✔
6	from dataclasses import dataclass	3✔
7	from enum import Enum, unique	3✔
8	from importlib.metadata import version as get_version	3✔
9	from typing import Optional, Type, Collection, List, Callable	3✔
10
11	try:	3✔
12	import asyncclick as click	3✔
NEW 13	except ImportError:	×
NEW 14	print("This command requires the CLI extra.")	×
NEW 15	print("Install with: pip install rsocket[cli]")	×
NEW 16	sys.exit(1)	×
17
18	from rsocket.awaitable.awaitable_rsocket import AwaitableRSocket	3✔
19	from rsocket.extensions.helpers import route, composite, authenticate_simple, authenticate_bearer	3✔
20	from rsocket.extensions.mimetypes import WellKnownMimeTypes	3✔
21	from rsocket.frame import MAX_REQUEST_N	3✔
22	from rsocket.frame_helpers import ensure_bytes, safe_len	3✔
23	from rsocket.helpers import single_transport_provider	3✔
24	from rsocket.payload import Payload	3✔
25	from rsocket.rsocket_client import RSocketClient	3✔
26	from rsocket.transports.abstract_messaging import AbstractMessagingTransport	3✔
27
28	from rsocket.transports.tcp import TransportTCP	3✔
29
30
31	@unique	3✔
32	class RequestType(Enum):	3✔
33	response = 'REQUEST_RESPONSE'	3✔
34	stream = 'REQUEST_STREAM'	3✔
35	channel = 'REQUEST_CHANNEL'	3✔
36	fnf = 'FIRE_AND_FORGET'	3✔
37	metadata_push = 'METADATA_PUSH'	3✔
38
39
40	interaction_models: List[str] = [str(e.value) for e in RequestType]	3✔
41
42
43	@dataclass(frozen=True)	3✔
44	class RSocketUri:	3✔
45	host: str	3✔
46	port: int	3✔
47	schema: str	3✔
48	path: Optional[str] = None	3✔
49	original_uri: Optional[str] = None	3✔
50
51
52	def parse_uri(uri: str) -> RSocketUri:	3✔
53	schema, rest = uri.split(':', 1)	3✔
54	rest = rest.strip('/')	3✔
55	host_port_path = rest.split('/', 1)	3✔
56	host_port = host_port_path[0].split(':')	3✔
57	host = host_port[0]	3✔
58
59	if len(host_port) == 1:	3✔
60	port = None	3✔
61	else:
62	port = int(host_port[1])	3✔
63
64	if len(host_port_path) > 1:	3✔
65	rest = host_port_path[1]	3✔
66	else:
67	rest = None	3✔
68
69	return RSocketUri(host, port, schema, rest, uri)	3✔
70
71
72	@asynccontextmanager	3✔
73	async def transport_from_uri(uri: RSocketUri,	3✔
74	verify_ssl=True,
75	headers: Optional = None,
76	trust_cert: Optional[str] = None) -> Type[AbstractMessagingTransport]:
77	if uri.schema == 'tcp':	3✔
78	connection = await asyncio.open_connection(uri.host, uri.port)	3✔
79	yield TransportTCP(*connection)	3✔
80	elif uri.schema in ['wss', 'ws']:	3!
81	import aiohttp	3✔
82	async with aiohttp.ClientSession() as session:	3✔
83	if trust_cert is not None:	3!
84	ssl_context = ssl.create_default_context(cafile=trust_cert)	×
85	else:
86	ssl_context = None	3✔
87
88	async with session.ws_connect(uri.original_uri,	3✔
89	verify_ssl=verify_ssl,
90	ssl_context=ssl_context,
91	headers=headers) as websocket:
92	from rsocket.transports.aiohttp_websocket import TransportAioHttpClient	3✔
93
94	yield TransportAioHttpClient(websocket=websocket)	3✔
95	else:
96	raise Exception('Unsupported schema in CLI')	×
97
98
99	def build_composite_metadata(auth_simple: Optional[str],	3✔
100	route_value: Optional[str],
101	auth_bearer: Optional[str]) -> List:
102	composite_items = []	3✔
103
104	if route_value is not None:	3✔
105	composite_items.append(route(route_value))	3✔
106
107	if auth_simple is not None and auth_bearer is not None:	3✔
108	raise click.UsageError('Multiple authentication methods specified.')	3✔
109
110	if auth_simple is not None:	3✔
111	composite_items.append(authenticate_simple(*auth_simple.split(':')))	3✔
112
113	if auth_bearer is not None:	3✔
114	composite_items.append(authenticate_bearer(auth_bearer))	3✔
115
116	return composite_items	3✔
117
118
119	@asynccontextmanager	3✔
120	async def create_client(parsed_uri,	3✔
121	data_mime_type,
122	metadata_mime_type,
123	setup_payload,
124	allow_untrusted_ssl=False,
125	http_headers=None,
126	trust_cert=None):
127	async with transport_from_uri(parsed_uri,	3✔
128	verify_ssl=not allow_untrusted_ssl,
129	headers=http_headers,
130	trust_cert=trust_cert) as transport:
131	async with RSocketClient(single_transport_provider(transport),	3✔
132	data_encoding=data_mime_type or WellKnownMimeTypes.APPLICATION_JSON,
133	metadata_encoding=metadata_mime_type or WellKnownMimeTypes.APPLICATION_JSON,
134	setup_payload=setup_payload) as client:
135	yield AwaitableRSocket(client)	3✔
136
137
138	def get_request_type(request: bool,	3✔
139	stream: bool,
140	fnf: bool,
141	metadata_push: bool,
142	channel: bool,
143	interaction_model: Optional[str]) -> RequestType:
144	interaction_options = list(filter(lambda _: _ is True, [request, stream, fnf, channel, metadata_push]))	3✔
145
146	if len(interaction_options) >= 2 or (len(interaction_options) >= 1 and interaction_model is not None):	3✔
147	raise click.UsageError('Multiple interaction methods specified.')	3✔
148
149	if interaction_model is not None:	3✔
150	return RequestType(interaction_model.upper())	3✔
151	if request:	3✔
152	return RequestType.response	3✔
153	if stream:	3✔
154	return RequestType.stream	3✔
155	if channel:	3✔
156	return RequestType.channel	3✔
157	if fnf:	3✔
158	return RequestType.fnf	3✔
159	if metadata_push:	3✔
160	return RequestType.metadata_push	3✔
161
162	raise click.UsageError('No interaction method specified (eg. --request)')	3✔
163
164
165	@click.command(name='rsocket-py', help='Supported connection strings: tcp/ws/wss')	3✔
166	@click.option('--im', '--interactionModel', 'interaction_model', is_flag=False,	3✔
167	type=click.Choice(interaction_models, case_sensitive=False),
168	help='Interaction Model')
169	@click.option('--request', is_flag=True,	3✔
170	help='Request response')
171	@click.option('--stream', is_flag=True,	3✔
172	help='Request stream')
173	@click.option('--channel', is_flag=True,	3✔
174	help='Request channel')
175	@click.option('--fnf', is_flag=True,	3✔
176	help='Fire and Forget')
177	@click.option('--metadataPush', 'metadata_push', is_flag=True,	3✔
178	help='Metadata Push')
179	@click.option('-d', '--data', '--input', 'data', is_flag=False,	3✔
180	help='Data. Use "-" to read data from standard input. (default: )')
181	@click.option('-l', '--load', is_flag=False,	3✔
182	help='Load a file as Data. (e.g. ./foo.txt, /tmp/foo.txt)')
183	@click.option('-m', '--metadata', is_flag=False, default=None,	3✔
184	help='Metadata (default: )')
185	@click.option('-r', '--route', 'route_value', is_flag=False, default=None,	3✔
186	help='Enable Routing Metadata Extension')
187	@click.option('--limitRate', 'limit_rate', is_flag=False, default=None, type=int,	3✔
188	help='Enable limitRate(rate)')
189	@click.option('--take', 'take_n', is_flag=False, default=None, type=int,	3✔
190	help='Enable take(n)')
191	@click.option('-u', '--as', '--authSimple', 'auth_simple', is_flag=False, default=None,
192	help='Enable Authentication Metadata Extension (Simple). The format must be "username: password"')
193	@click.option('--sd', '--setup', '--setupData', 'setup_data', is_flag=False, default=None,	3✔
194	help='Data for Setup payload')
195	@click.option('--sm', '--setupMetadata', 'setup_metadata', is_flag=False, default=None,	3✔
196	help='Metadata for Setup payload')
197	@click.option('--ab', '--authBearer', 'auth_bearer', is_flag=False, default=None,	3✔
198	help='Enable Authentication Metadata Extension (Bearer)')
199	@click.option('--dataMimeType', '--dataFormat', '--dmt', 'data_mime_type', is_flag=False,	3✔
200	help='MimeType for data (default: application/json)')
201	@click.option('--metadataMimeType', '--metadataFormat', '--mmt', 'metadata_mime_type', is_flag=False,	3✔
202	help='MimeType for metadata (default:application/json)')
203	@click.option('--allowUntrustedSsl', 'allow_untrusted_ssl', is_flag=True, default=False,	3✔
204	help='Do not verify SSL certificate (for wss:// urls)')
205	@click.option('-H', '--header', '--httpHeader', 'http_header', multiple=True,	3✔
206	help='ws/wss headers')
207	@click.option('--trustCert', 'trust_cert', is_flag=False,	3✔
208	help='PEM file for a trusted certificate. (e.g. ./foo.crt, /tmp/foo.crt)')
209	@click.option('--debug', is_flag=True,	3✔
210	help='Show debug log')
211	@click.option('--quiet', '-q', is_flag=True,	3✔
212	help='Disable the output on next')
213	@click.option('--timeout', 'timeout_seconds', is_flag=False, type=int,	3✔
214	help='Timeout in seconds')
215	@click.option('--version', is_flag=True,	3✔
216	help='Print version')
217	@click.argument('uri', required=False)	3✔
218	@click.pass_context
219	async def command(context, data, load,
220	metadata, route_value, auth_simple, auth_bearer,
221	limit_rate, take_n, allow_untrusted_ssl,
222	setup_data, setup_metadata, interaction_model,
223	http_header, metadata_push, timeout_seconds,
224	data_mime_type, metadata_mime_type,
225	request, stream, channel, fnf, trust_cert,
226	uri, debug, version, quiet):
227	if version:
228	try:
229	print(get_version('rsocket'))
230	except Exception:
231	print('Failed to find version')
232	return
233
234	if uri is None:
235	raise click.MissingParameter(param=context.command.params[-1])
236
237	if quiet:
238	logging.basicConfig(handlers=[])
239
240	if debug:
241	logging.basicConfig(level=logging.DEBUG)
242
243	if take_n == 0:
244	return
245
246	request_type = get_request_type(request, stream, fnf, metadata_push, channel, interaction_model)
247	http_headers = parse_headers(http_header)
248	composite_items = build_composite_metadata(auth_simple, route_value, auth_bearer)
249	setup_payload = create_setup_payload(setup_data, setup_metadata)
250	metadata_value = get_metadata_value(composite_items, metadata)
251	metadata_mime_type = normalize_metadata_mime_type(composite_items, metadata_mime_type)
252	parsed_uri = parse_uri(uri)
253
254	def payload_provider():
255	return create_request_payload(data, load, metadata_value)
256
257	future = run_request(request_type, limit_rate, payload_provider,
258	http_headers=http_headers,
259	allow_untrusted_ssl=allow_untrusted_ssl,
260	metadata_mime_type=metadata_mime_type,
261	data_mime_type=data_mime_type,
262	setup_payload=setup_payload,
263	trust_cert=trust_cert,
264	parsed_uri=parsed_uri)
265
266	if timeout_seconds is not None:
267	result = await asyncio.wait_for(future, timeout_seconds)
268	else:
269	result = await future
270
271	if not quiet:
272	output_result(result)
273
274
275	async def run_request(request_type: RequestType,	3✔
276	limit_rate: Optional[int],
277	payload_provider: Callable[[], Payload],
278	**kwargs):
279	async with create_client(**kwargs) as client:	3✔
280	return await execute_request(client,	3✔
281	request_type,
282	normalize_limit_rate(limit_rate),
283	payload_provider())
284
285
286	def parse_headers(http_headers):	3✔
287	if safe_len(http_headers) > 0:	3✔
288	headers = dict()	3✔
289
290	for header in http_headers:	3✔
291	parts = header.split('=', 2)	3✔
292	headers[parts[0]] = parts[1]	3✔
293
294	return headers	3✔
295
296	return None	3✔
297
298
299	def normalize_metadata_mime_type(composite_items: List, metadata_mime_type):	3✔
300	if len(composite_items) > 0:	3✔
301	metadata_mime_type = WellKnownMimeTypes.MESSAGE_RSOCKET_COMPOSITE_METADATA	3✔
302
303	return metadata_mime_type	3✔
304
305
306	def create_request_payload(data: Optional[str],	3✔
307	load: Optional[str],
308	metadata: Optional[bytes]) -> Payload:
309	data = normalize_data(data, load)	3✔
310
311	return Payload(data, metadata)	3✔
312
313
314	def output_result(result):	3✔
315	if isinstance(result, Payload):	3!
316	print(result.data.decode('utf-8'))	3✔
317	elif isinstance(result, Collection):	×
318	print([p.data.decode('utf-8') for p in result])	×
319
320
321	async def execute_request(awaitable_client: AwaitableRSocket,	3✔
322	request_type: RequestType,
323	limit_rate: int,
324	payload: Payload):
325	result = None	3✔
326
327	if request_type is RequestType.response:	3✔
328	result = await awaitable_client.request_response(payload)	3✔
329	elif request_type is RequestType.stream:	3✔
330	result = await awaitable_client.request_stream(payload, limit_rate=limit_rate)	3✔
331	elif request_type is RequestType.channel:	3✔
332	result = await awaitable_client.request_channel(payload, limit_rate=limit_rate)	3✔
333	elif request_type is RequestType.fnf:	3!
334	await awaitable_client.fire_and_forget(payload)	3✔
335	elif request_type is RequestType.metadata_push:	×
336	await awaitable_client.metadata_push(payload.metadata)	×
337
338	return result	3✔
339
340
341	def get_metadata_value(composite_items: List, metadata: Optional[str]) -> bytes:	3✔
342	if len(composite_items) > 0:	3✔
343	metadata_value = composite(*composite_items)	3✔
344	else:
345	metadata_value = metadata	3✔
346
347	return ensure_bytes(metadata_value)	3✔
348
349
350	def create_setup_payload(setup_data: Optional[str], setup_metadata: Optional[str]) -> Optional[Payload]:	3✔
351	setup_payload = None	3✔
352
353	if setup_data is not None or setup_metadata is not None:	3✔
354	setup_payload = Payload(	3✔
355	ensure_bytes(setup_data),
356	ensure_bytes(setup_metadata)
357	)
358
359	return setup_payload	3✔
360
361
362	def normalize_data(data: Optional[str], load: Optional[str]) -> bytes:	3✔
363	if data == '-':	3✔
364	stdin_text = click.get_text_stream('stdin')	3✔
365	return ensure_bytes(stdin_text.read())	3✔
366
367	if load is not None:	3✔
368	with open(load) as fd:	3✔
369	return ensure_bytes(fd.read())	3✔
370
371	return ensure_bytes(data)	3✔
372
373
374	def normalize_limit_rate(limit_rate):	3✔
375	if limit_rate is None or limit_rate <= 0:	3✔
376	limit_rate = MAX_REQUEST_N	3✔
377
378	return limit_rate	3✔
379
380
381	if __name__ == '__main__':	3!
382	command()	×

rsocket / rsocket-py / 18095678244

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous