File: test_eventstream.py

package info (click to toggle)
aws-crt-python 0.20.4%2Bdfsg-1~bpo12%2B1
  • links: PTS, VCS
  • area: main
  • in suites: bookworm-backports
  • size: 72,656 kB
  • sloc: ansic: 381,805; python: 23,008; makefile: 6,251; sh: 4,536; cpp: 699; ruby: 208; java: 77; perl: 73; javascript: 46; xml: 11
file content (127 lines) | stat: -rw-r--r-- 4,797 bytes parent folder | download | duplicates (3)
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
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0.

from awscrt.eventstream import *
from test import NativeResourceTest
import time
from uuid import UUID, uuid4


class TestHeaders(NativeResourceTest):
    def test_bool_true(self):
        name = 'truthy'
        value = True
        h = Header.from_bool(name, value)
        self.assertIs(HeaderType.BOOL_TRUE, h.type)
        self.assertEqual(value, h.value_as_bool())
        self.assertEqual(value, h.value)
        self.assertEqual(name, h.name)

    def test_bool_false(self):
        name = 'falsey'
        value = False
        h = Header.from_bool(name, value)
        self.assertIs(HeaderType.BOOL_FALSE, h.type)
        self.assertEqual(value, h.value_as_bool())
        self.assertEqual(value, h.value)
        self.assertEqual(name, h.name)

    def test_byte(self):
        name = 'bytey'
        value = 127
        h = Header.from_byte(name, value)
        self.assertIs(HeaderType.BYTE, h.type)
        self.assertEqual(value, h.value_as_byte())
        self.assertEqual(value, h.value)
        self.assertEqual(name, h.name)

        self.assertRaises(ValueError, Header.from_byte, 'too-big', 128)
        self.assertRaises(ValueError, Header.from_byte, 'too-small', -129)

    def test_int16(self):
        name = 'sweet16'
        value = 32000
        h = Header.from_int16(name, value)
        self.assertIs(HeaderType.INT16, h.type)
        self.assertEqual(value, h.value_as_int16())
        self.assertEqual(value, h.value)
        self.assertEqual(name, h.name)

        self.assertRaises(ValueError, Header.from_int16, 'too-big', 64000)
        self.assertRaises(ValueError, Header.from_int16, 'too-small', -64000)

    def test_int32(self):
        name = 'win32'
        value = 2000000000
        h = Header.from_int32(name, value)
        self.assertIs(HeaderType.INT32, h.type)
        self.assertEqual(value, h.value_as_int32())
        self.assertEqual(value, h.value)
        self.assertEqual(name, h.name)

        self.assertRaises(ValueError, Header.from_int32, 'too-big', 4000000000)
        self.assertRaises(ValueError, Header.from_int32, 'too-small', -4000000000)

    def test_int64(self):
        name = 'N64'
        value = 9223372036854775807
        h = Header.from_int64(name, value)
        self.assertIs(HeaderType.INT64, h.type)
        self.assertEqual(value, h.value_as_int64())
        self.assertEqual(value, h.value)
        self.assertEqual(name, h.name)

        self.assertRaises(ValueError, Header.from_int32, 'too-big', 18000000000000000000)
        self.assertRaises(ValueError, Header.from_int32, 'too-small', -18000000000000000000)

    def test_byte_buf(self):
        name = 'buffy'
        value = bytes(range(0, 256)) * 100
        h = Header.from_byte_buf(name, value)
        self.assertIs(HeaderType.BYTE_BUF, h.type)
        self.assertEqual(value, h.value_as_byte_buf())
        self.assertEqual(value, h.value)
        self.assertEqual(name, h.name)

    def test_string(self):
        name = 'stringy'
        value = 'abcdefghijklmnopqrstuvwxyz' * 100
        h = Header.from_string(name, value)
        self.assertIs(HeaderType.STRING, h.type)
        self.assertEqual(value, h.value_as_string())
        self.assertEqual(value, h.value)
        self.assertEqual(name, h.name)

    def test_timestamp(self):
        name = 'timeywimey'
        value = time.time()
        h = Header.from_timestamp(name, value)
        self.assertIs(HeaderType.TIMESTAMP, h.type)
        # compare with delta, since protocol uses int instead of float
        self.assertAlmostEqual(value, h.value_as_timestamp(), delta=1.0)
        self.assertAlmostEqual(value, h.value, delta=1.0)
        self.assertEqual(name, h.name)

    def test_uuid(self):
        name = 'davuuid'
        value = UUID('01234567-89ab-cdef-0123-456789abcdef')
        h = Header.from_uuid(name, value)
        self.assertIs(HeaderType.UUID, h.type)
        self.assertEqual(value, h.value_as_uuid())
        self.assertEqual(value, h.value)
        self.assertEqual(name, h.name)

    def test_wrong_type(self):
        h = Header.from_bool('truthy', True)
        self.assertRaises(TypeError, h.value_as_byte)
        self.assertRaises(TypeError, h.value_as_int16)
        self.assertRaises(TypeError, h.value_as_int32)
        self.assertRaises(TypeError, h.value_as_int64)
        self.assertRaises(TypeError, h.value_as_byte_buf)
        self.assertRaises(TypeError, h.value_as_string)
        self.assertRaises(TypeError, h.value_as_timestamp)
        self.assertRaises(TypeError, h.value_as_uuid)

        h = Header.from_int32('32', 32)
        self.assertRaises(TypeError, h.value_as_bool)
        self.assertRaises(TypeError, h.value_as_int64)