File size: 5,009 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
134
135
136
137
from typing import List
from unittest import TestCase
from uuid import uuid4, UUID

import pytest
from botframework.streaming.payloads import HeaderSerializer
from botframework.streaming.payloads.models import Header, PayloadTypes
from botframework.streaming.transport import TransportConstants


class TestHeaderSerializer(TestCase):
    def test_can_round_trip(self):
        header = Header()
        header.type = PayloadTypes.REQUEST
        header.payload_length = 168
        header.id = uuid4()
        header.end = True

        buffer: List[int] = [None] * TransportConstants.MAX_PAYLOAD_LENGTH
        offset: int = 0

        length = HeaderSerializer.serialize(header, buffer, offset)
        result = HeaderSerializer.deserialize(buffer, 0, length)

        self.assertEqual(header.type, result.type)
        self.assertEqual(header.payload_length, result.payload_length)
        self.assertEqual(header.id, result.id)
        self.assertEqual(header.end, result.end)

    def test_serializes_to_ascii(self):
        header = Header()
        header.type = PayloadTypes.REQUEST
        header.payload_length = 168
        header.id = uuid4()
        header.end = True

        buffer: List[int] = [None] * TransportConstants.MAX_PAYLOAD_LENGTH
        offset: int = 0

        length = HeaderSerializer.serialize(header, buffer, offset)
        decoded = bytes(buffer[offset:length]).decode("ascii")

        self.assertEqual(f"A.000168.{str(header.id)}.1\n", decoded)

    def test_deserializes_from_ascii(self):
        header_id: UUID = uuid4()
        header: str = f"A.000168.{str(header_id)}.1\n"
        buffer: List[int] = list(bytes(header, "ascii"))

        result = HeaderSerializer.deserialize(buffer, 0, len(buffer))

        self.assertEqual("A", result.type)
        self.assertEqual(168, result.payload_length)
        self.assertEqual(header_id, result.id)
        self.assertTrue(result.end)

    def test_deserialize_unknown_type(self):
        header_id: UUID = uuid4()
        header: str = f"Z.000168.{str(header_id)}.1\n"
        buffer: List[int] = list(bytes(header, "ascii"))

        result = HeaderSerializer.deserialize(buffer, 0, len(buffer))

        self.assertEqual("Z", result.type)
        self.assertEqual(168, result.payload_length)

    def test_deserialize_length_too_short_throws(self):
        header_id: UUID = uuid4()
        header: str = f"A.000168.{str(header_id)}.1\n"
        buffer: List[int] = list(bytes(header, "ascii"))

        with pytest.raises(ValueError):
            HeaderSerializer.deserialize(buffer, 0, 5)

    def test_deserialize_length_too_long_throws(self):
        header_id: UUID = uuid4()
        header: str = f"A.000168.{str(header_id)}.1\n"
        buffer: List[int] = list(bytes(header, "ascii"))

        with pytest.raises(ValueError):
            HeaderSerializer.deserialize(buffer, 0, 55)

    def test_deserialize_bad_type_delimiter_throws(self):
        header_id: UUID = uuid4()
        header: str = f"Ax000168.{str(header_id)}.1\n"
        buffer: List[int] = list(bytes(header, "ascii"))

        with pytest.raises(ValueError):
            HeaderSerializer.deserialize(buffer, 0, len(buffer))

    def test_deserialize_bad_length_delimiter_throws(self):
        header_id: UUID = uuid4()
        header: str = f"A.000168x{str(header_id)}.1\n"
        buffer: List[int] = list(bytes(header, "ascii"))

        with pytest.raises(ValueError):
            HeaderSerializer.deserialize(buffer, 0, len(buffer))

    def test_deserialize_bad_id_delimiter_throws(self):
        header_id: UUID = uuid4()
        header: str = f"A.000168.{str(header_id)}x1\n"
        buffer: List[int] = list(bytes(header, "ascii"))

        with pytest.raises(ValueError):
            HeaderSerializer.deserialize(buffer, 0, len(buffer))

    def test_deserialize_bad_terminator_throws(self):
        header_id: UUID = uuid4()
        header: str = f"A.000168.{str(header_id)}.1c"
        buffer: List[int] = list(bytes(header, "ascii"))

        with pytest.raises(ValueError):
            HeaderSerializer.deserialize(buffer, 0, len(buffer))

    def test_deserialize_bad_length_throws(self):
        header_id: UUID = uuid4()
        header: str = f"A.00p168.{str(header_id)}.1\n"
        buffer: List[int] = list(bytes(header, "ascii"))

        with pytest.raises(ValueError):
            HeaderSerializer.deserialize(buffer, 0, len(buffer))

    def test_deserialize_bad_id_throws(self):
        header: str = "A.000168.68e9p9ca-a651-40f4-ad8f-3aaf781862b4.1\n"
        buffer: List[int] = list(bytes(header, "ascii"))

        with pytest.raises(ValueError):
            HeaderSerializer.deserialize(buffer, 0, len(buffer))

    def test_deserialize_bad_end_throws(self):
        header_id: UUID = uuid4()
        header: str = f"A.000168.{str(header_id)}.z\n"
        buffer: List[int] = list(bytes(header, "ascii"))

        with pytest.raises(ValueError):
            HeaderSerializer.deserialize(buffer, 0, len(buffer))