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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
|
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0.
from awscrt.io import *
from test import NativeResourceTest, TIMEOUT
import io
import os
import sys
import unittest
class EventLoopGroupTest(NativeResourceTest):
def test_init_defaults(self):
event_loop_group = EventLoopGroup()
def test_1_thread(self):
event_loop_group = EventLoopGroup(1)
def test_cpu_group(self):
event_loop_group = EventLoopGroup(cpu_group=0)
def test_shutdown_complete(self):
event_loop_group = EventLoopGroup()
shutdown_event = event_loop_group.shutdown_event
del event_loop_group
self.assertTrue(shutdown_event.wait(TIMEOUT))
def test_init_defaults_singleton(self):
event_loop_group = EventLoopGroup.get_or_create_static_default()
def test_init_defaults_singleton_is_singleton(self):
event_loop_group_one = EventLoopGroup.get_or_create_static_default()
event_loop_group_two = EventLoopGroup.get_or_create_static_default()
self.assertTrue(event_loop_group_one == event_loop_group_two)
def test_shutdown_complete_singleton(self):
event_loop_group = EventLoopGroup.get_or_create_static_default()
shutdown_event = event_loop_group.shutdown_event
del event_loop_group
EventLoopGroup.release_static_default()
self.assertTrue(shutdown_event.wait(TIMEOUT))
class DefaultHostResolverTest(NativeResourceTest):
def test_init(self):
event_loop_group = EventLoopGroup()
host_resolver = DefaultHostResolver(event_loop_group)
def test_init_singleton(self):
host_resolver = DefaultHostResolver.get_or_create_static_default()
def test_init_singleton_is_singleton(self):
host_resolver_one = DefaultHostResolver.get_or_create_static_default()
host_resolver_two = DefaultHostResolver.get_or_create_static_default()
self.assertTrue(host_resolver_one == host_resolver_two)
class ClientBootstrapTest(NativeResourceTest):
def test_create_destroy(self):
event_loop_group = EventLoopGroup()
host_resolver = DefaultHostResolver(event_loop_group)
bootstrap = ClientBootstrap(event_loop_group, host_resolver)
# ensure shutdown_event fires
bootstrap_shutdown_event = bootstrap.shutdown_event
del bootstrap
self.assertTrue(bootstrap_shutdown_event.wait(TIMEOUT))
def test_create_destroy_singleton(self):
bootstrap = ClientBootstrap.get_or_create_static_default()
# ensure shutdown_event fires
bootstrap_shutdown_event = bootstrap.shutdown_event
del bootstrap
ClientBootstrap.release_static_default()
self.assertTrue(bootstrap_shutdown_event.wait(TIMEOUT))
def test_init_singleton_is_singleton(self):
client_one = ClientBootstrap.get_or_create_static_default()
client_two = ClientBootstrap.get_or_create_static_default()
self.assertTrue(client_one == client_two)
class ClientTlsContextTest(NativeResourceTest):
def test_init_defaults(self):
opt = TlsContextOptions()
ctx = ClientTlsContext(opt)
def test_with_mtls_from_path(self):
opt = TlsContextOptions.create_client_with_mtls_from_path(
'test/resources/unittest.crt', 'test/resources/unittest.key')
ctx = ClientTlsContext(opt)
def test_with_mtls_pkcs12(self):
try:
opt = TlsContextOptions.create_client_with_mtls_pkcs12(
'test/resources/unittest.p12', '1234')
ctx = ClientTlsContext(opt)
except Exception as e:
if 'PLATFORM_NOT_SUPPORTED' in str(e):
raise unittest.SkipTest(f'PKCS#12 not supported on this platform ({sys.platform})')
else:
# well then this is a real error
raise e
def test_override_default_trust_store_dir(self):
opt = TlsContextOptions()
opt.override_default_trust_store_from_path('test/resources', None)
ctx = ClientTlsContext(opt)
def test_override_default_trust_store_file(self):
opt = TlsContextOptions()
opt.override_default_trust_store_from_path(None, 'test/resources/ca.crt')
ctx = ClientTlsContext(opt)
class TlsConnectionOptionsTest(NativeResourceTest):
def test_init(self):
opt = TlsContextOptions()
ctx = ClientTlsContext(opt)
conn_opt = TlsConnectionOptions(ctx)
def test_alpn_list(self):
opt = TlsContextOptions()
ctx = ClientTlsContext(opt)
conn_opt = TlsConnectionOptions(ctx)
conn_opt.set_alpn_list(['h2', 'http/1.1'])
def test_server_name(self):
opt = TlsContextOptions()
ctx = ClientTlsContext(opt)
conn_opt = TlsConnectionOptions(ctx)
conn_opt.set_server_name('localhost')
class MockPythonStream:
"""For testing duck-typed stream classes.
Doesn't inherit from io.IOBase. Doesn't implement readinto()"""
def __init__(self, src_data):
self.data = bytes(src_data)
self.len = len(src_data)
self.pos = 0
def seek(self, where):
self.pos = where
def tell(self):
return self.pos
def read(self, amount=None):
if amount is None:
amount = self.len - self.pos
else:
amount = min(amount, self.len - self.pos)
prev_pos = self.pos
self.pos += amount
return self.data[prev_pos: self.pos]
class InputStreamTest(NativeResourceTest):
def _test(self, python_stream, expected):
input_stream = InputStream(python_stream)
result = bytearray()
fixed_mv_len = 4
fixed_mv = memoryview(bytearray(fixed_mv_len))
while True:
read_len = input_stream._read_into_memoryview(fixed_mv)
if read_len is None:
continue
if read_len == 0:
break
if read_len > 0:
self.assertLessEqual(read_len, fixed_mv_len)
result += fixed_mv[0:read_len]
self.assertEqual(expected, result)
def test_read_official_io(self):
# Read from a class defined in the io module
src_data = b'a long string here'
python_stream = io.BytesIO(src_data)
self._test(python_stream, src_data)
def test_read_duck_typed_io(self):
# Read from a class defined in the io module
src_data = b'a man a can a planal canada'
python_stream = MockPythonStream(src_data)
self._test(python_stream, src_data)
class Pkcs11LibTest(NativeResourceTest):
def _lib_path(self):
val = os.environ.get("AWS_TEST_PKCS11_LIB")
if not val:
raise unittest.SkipTest(f"test requires env var: AWS_TEST_PKCS11_LIB")
return val
def test_init(self):
# sanity check that we can create/destroy
lib_path = self._lib_path()
pcks11_lib = Pkcs11Lib(file=lib_path, behavior=Pkcs11Lib.InitializeFinalizeBehavior.STRICT)
def test_exceptions(self):
# check that initialization errors bubble up as exceptions
with self.assertRaises(Exception):
pkcs11_lib = Pkcs11Lib(file='obviously-invalid-path.so')
with self.assertRaises(Exception):
with open(self._lib_path()) as literal_open_file:
# a filepath str should passed, not a literal open file
pkcs11_lib = Pkcs11Lib(file=literal_open_file)
def test_strict_behavior(self):
lib_path = self._lib_path()
lib1 = Pkcs11Lib(file=lib_path, behavior=Pkcs11Lib.InitializeFinalizeBehavior.STRICT)
# InitializeFinalizeBehavior.STRICT behavior should fail if the PKCS#11 lib is already loaded
with self.assertRaises(Exception):
lib2 = Pkcs11Lib(file=lib_path, behavior=Pkcs11Lib.InitializeFinalizeBehavior.STRICT)
def test_omit_behavior(self):
lib_path = self._lib_path()
# InitializeFinalizeBehavior.OMIT should fail unless another instance of the PKCS#11 lib is already loaded
with self.assertRaises(Exception):
lib = Pkcs11Lib(file=lib_path, behavior=Pkcs11Lib.InitializeFinalizeBehavior.OMIT)
# InitializeFinalizeBehavior.OMIT behavior should be fine when another
# instance of the PKCS#11 lib is already loaded
strict_lib = Pkcs11Lib(file=lib_path, behavior=Pkcs11Lib.InitializeFinalizeBehavior.STRICT)
omit_lib = Pkcs11Lib(file=lib_path, behavior=Pkcs11Lib.InitializeFinalizeBehavior.OMIT)
def test_is_tls_cipher_supported(self):
self.assertEqual(True, TlsCipherPref.DEFAULT.is_supported())
if __name__ == '__main__':
unittest.main()
|
