Spaces:
Build error
Build error
File size: 4,405 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 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 |
import json
from http import HTTPStatus
import aiounittest
from botbuilder.schema import Activity
from botframework.streaming import ReceiveResponse, StreamingResponse
from botframework.streaming.payloads import ResponseMessageStream
class TestResponses(aiounittest.AsyncTestCase):
async def test_receive_response_empty_streams(self):
sut = ReceiveResponse()
self.assertIsNotNone(sut.streams)
self.assertEqual(0, len(sut.streams))
async def test_receive_response_none_properties(self):
sut = ReceiveResponse()
self.assertEqual(0, sut.status_code)
async def test_streaming_response_null_properties(self):
sut = StreamingResponse()
self.assertEqual(0, sut.status_code)
self.assertIsNone(sut.streams)
async def test_streaming_response_add_stream_none_throws(self):
sut = StreamingResponse()
with self.assertRaises(TypeError):
sut.add_stream(None)
async def test_streaming_response_add_stream_success(self):
sut = StreamingResponse()
content = "hi"
sut.add_stream(content)
self.assertIsNotNone(sut.streams)
self.assertEqual(1, len(sut.streams))
self.assertEqual(content, sut.streams[0].content)
async def test_streaming_response_add_stream_existing_list_success(self):
sut = StreamingResponse()
content = "hi"
content_2 = "hello"
sut.streams = [ResponseMessageStream(content=content_2)]
sut.add_stream(content)
self.assertIsNotNone(sut.streams)
self.assertEqual(2, len(sut.streams))
self.assertEqual(content_2, sut.streams[0].content)
self.assertEqual(content, sut.streams[1].content)
async def test_streaming_response_not_found_success(self):
sut = StreamingResponse.not_found()
self.assertEqual(HTTPStatus.NOT_FOUND, sut.status_code)
self.assertIsNone(sut.streams)
async def test_streaming_response_forbidden_success(self):
sut = StreamingResponse.forbidden()
self.assertEqual(HTTPStatus.FORBIDDEN, sut.status_code)
self.assertIsNone(sut.streams)
async def test_streaming_response_ok_success(self):
sut = StreamingResponse.ok()
self.assertEqual(HTTPStatus.OK, sut.status_code)
self.assertIsNone(sut.streams)
async def test_streaming_response_internal_server_error_success(self):
sut = StreamingResponse.internal_server_error()
self.assertEqual(HTTPStatus.INTERNAL_SERVER_ERROR, sut.status_code)
self.assertIsNone(sut.streams)
async def test_streaming_response_create_with_body_success(self):
content = "hi"
sut = StreamingResponse.create_response(HTTPStatus.OK, content)
self.assertEqual(HTTPStatus.OK, sut.status_code)
self.assertIsNotNone(sut.streams)
self.assertEqual(1, len(sut.streams))
self.assertEqual(content, sut.streams[0].content)
async def test_streaming_response_set_body_string_success(self):
sut = StreamingResponse()
sut.set_body("123")
self.assertIsNotNone(sut.streams)
self.assertEqual(1, len(sut.streams))
self.assertIsInstance(sut.streams[0].content, list)
self.assertIsInstance(sut.streams[0].content[0], int)
self.assertEqual("123", bytes(sut.streams[0].content).decode("utf-8-sig"))
async def test_streaming_response_set_body_none_does_not_throw(self):
sut = StreamingResponse()
sut.set_body(None)
async def test_streaming_response_set_body_success(self):
sut = StreamingResponse()
activity = Activity(text="hi", type="message")
sut.set_body(activity)
self.assertIsNotNone(sut.streams)
self.assertEqual(1, len(sut.streams))
self.assertIsInstance(sut.streams[0].content, list)
self.assertIsInstance(sut.streams[0].content[0], int)
assert_activity = Activity.deserialize(
json.loads(bytes(sut.streams[0].content).decode("utf-8-sig"))
)
self.assertEqual(activity.text, assert_activity.text)
self.assertEqual(activity.type, assert_activity.type)
async def test_receive_base_read_body_as_string_no_content_empty_string(self):
sut = ReceiveResponse()
sut.streams = []
result = sut.read_body_as_str()
self.assertEqual("", result)
|