Spaces:
Build error
Build error
File size: 1,879 Bytes
0827183 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 |
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
from typing import List
from uuid import uuid4
import aiounittest
from botframework.streaming import PayloadStream, StreamingRequest
from botframework.streaming.payloads import SendOperations
from botframework.streaming.payloads.assemblers import PayloadStreamAssembler
from botframework.streaming.payload_transport import PayloadSender
from botframework.streaming.transport import TransportSenderBase
class MockTransportSender(TransportSenderBase):
# pylint: disable=unused-argument
def __init__(self):
super().__init__()
self.is_connected = True
self.buffers = []
async def send(self, buffer: List[int], offset: int, count: int) -> int:
self.buffers.append(buffer.copy())
return count
class TestSendOperations(aiounittest.AsyncTestCase):
async def test_request_dissasembler_with_variable_stream_send(self):
sender = PayloadSender()
transport = MockTransportSender()
sender.connect(transport)
sut = SendOperations(sender)
request = StreamingRequest.create_post("/a/b")
stream = PayloadStream(PayloadStreamAssembler(None, uuid4(), "blah", 100))
stream.write([0] * 100, 0, 100)
request.add_stream(await stream.read_until_end())
await sut.send_request(uuid4(), request)
self.assertEqual(4, len(transport.buffers))
async def test_request_dissasembler_with_json_stream_send(self):
sender = PayloadSender()
transport = MockTransportSender()
sender.connect(transport)
sut = SendOperations(sender)
request = StreamingRequest.create_post("/a/b")
request.add_stream(bytes("abc", "ascii"))
await sut.send_request(uuid4(), request)
self.assertEqual(4, len(transport.buffers))
|