Spaces:
Build error
Build error
File size: 1,842 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 58 |
from typing import List
from uuid import UUID, uuid4
import aiounittest
from botframework.streaming import ReceiveRequest
from botframework.streaming.payloads import StreamManager
from botframework.streaming.payloads.assemblers import (
ReceiveRequestAssembler,
PayloadStreamAssembler,
)
from botframework.streaming.payloads.models import (
Header,
RequestPayload,
StreamDescription,
)
class MockStreamManager(StreamManager):
def get_payload_assembler(self, identifier: UUID) -> PayloadStreamAssembler:
return PayloadStreamAssembler(self, identifier)
class TestPayloadProcessor(aiounittest.AsyncTestCase):
async def test_process_request(self):
# Arrange
header_id: UUID = uuid4()
header = Header(type="A", id=header_id, end=True)
header.payload_length = 3
stream_manager = MockStreamManager()
on_completed_called = False
async def mock_on_completed(identifier: UUID, request: ReceiveRequest):
nonlocal on_completed_called
assert identifier == header_id
assert request.verb == "POST"
assert request.path == "/api/messages"
assert len(request.streams) == 1
on_completed_called = True
sut = ReceiveRequestAssembler(
header, stream_manager, on_completed=mock_on_completed
)
# Act
stream_id: UUID = uuid4()
streams: List[StreamDescription] = [
StreamDescription(id=str(stream_id), content_type="json", length=100)
]
payload = RequestPayload(
verb="POST", path="/api/messages", streams=streams
).to_json()
payload_stream: List[int] = list(bytes(payload, "utf-8"))
await sut.process_request(payload_stream)
# Assert
assert on_completed_called
|