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
|
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0.
from test import NativeResourceTest
from awscrt import checksums
import unittest
import sys
class TestChecksums(NativeResourceTest):
def test_crc32_zeros_one_shot(self):
output = checksums.crc32(bytes(32))
expected = 0x190A55AD
self.assertEqual(expected, output)
def test_crc32_zeros_iterated(self):
output = 0
for i in range(32):
output = checksums.crc32(bytes(1), output)
expected = 0x190A55AD
self.assertEqual(expected, output)
def test_crc32_values_one_shot(self):
output = checksums.crc32(''.join(chr(i) for i in range(32)))
expected = 0x91267E8A
self.assertEqual(expected, output)
def test_crc32_values_iterated(self):
output = 0
for i in range(32):
output = checksums.crc32(chr(i), output)
expected = 0x91267E8A
self.assertEqual(expected, output)
def test_crc32_large_buffer(self):
# stress test gil optimization for 32 bit architecture which cannot handle huge buffer
large_buffer = bytes(25 * 2**20)
val = checksums.crc32(large_buffer)
self.assertEqual(0x72103906, val)
def test_crc32_huge_buffer(self):
# stress the internal logic that handles buffers larger than C's INT_MAX
if sys.platform.startswith('freebsd'):
# Skip this test for freebsd, as it simply crashes instead of raising exception in this case
raise unittest.SkipTest('Skip this test for freebsd')
try:
INT_MAX = 2**32 - 1
huge_buffer = bytes(INT_MAX + 5)
except (MemoryError, OverflowError):
raise unittest.SkipTest('Machine cant allocate giant buffer for giant buffer test')
val = checksums.crc32(huge_buffer)
self.assertEqual(0xc622f71d, val)
def test_crc32c_zeros_one_shot(self):
output = checksums.crc32c(bytes(32))
expected = 0x8A9136AA
self.assertEqual(expected, output)
def test_crc32c_zeros_iterated(self):
output = 0
for i in range(32):
output = checksums.crc32c(bytes(1), output)
expected = 0x8A9136AA
self.assertEqual(expected, output)
def test_crc32c_values_one_shot(self):
output = checksums.crc32c(''.join(chr(i) for i in range(32)))
expected = 0x46DD794E
self.assertEqual(expected, output)
def test_crc32c_values_iterated(self):
output = 0
for i in range(32):
output = checksums.crc32c(chr(i), output)
expected = 0x46DD794E
self.assertEqual(expected, output)
def test_crc32c_large_buffer(self):
# stress test gil optimization for 32 bit architecture which cannot handle huge buffer
large_buffer = bytes(25 * 2**20)
val = checksums.crc32c(large_buffer)
self.assertEqual(0xfb5b991d, val)
def test_crc32c_huge_buffer(self):
if sys.platform.startswith('freebsd'):
# Skip this test for freebsd, as it simply crashes instead of raising exception in this case
raise unittest.SkipTest('Skip this test for freebsd')
try:
INT_MAX = 2**32 - 1
huge_buffer = bytes(INT_MAX + 5)
except BaseException:
raise unittest.SkipTest('Machine cant allocate giant buffer for giant buffer test')
val = checksums.crc32c(huge_buffer)
self.assertEqual(0x572a7c8a, val)
def test_crc64nvme_zeros_one_shot(self):
output = checksums.crc64nvme(bytes(32))
expected = 0xcf3473434d4ecf3b
self.assertEqual(expected, output)
def test_crc64nvme_zeros_iterated(self):
output = 0
for i in range(32):
output = checksums.crc64nvme(bytes(1), output)
expected = 0xcf3473434d4ecf3b
self.assertEqual(expected, output)
def test_crc64nvme_values_one_shot(self):
output = checksums.crc64nvme(''.join(chr(i) for i in range(32)))
expected = 0xb9d9d4a8492cbd7f
self.assertEqual(expected, output)
def test_crc64nvme_values_iterated(self):
output = 0
for i in range(32):
output = checksums.crc64nvme(chr(i), output)
expected = 0xb9d9d4a8492cbd7f
self.assertEqual(expected, output)
def test_crc64nvme_large_buffer(self):
# stress test gil optimization for 32 bit architecture which cannot handle huge buffer
large_buffer = bytes(25 * 2**20)
val = checksums.crc64nvme(large_buffer)
self.assertEqual(0x5b6f5045463ca45e, val)
def test_crc64nvme_huge_buffer(self):
if sys.platform.startswith('freebsd'):
# Skip this test for freebsd, as it simply crashes instead of raising exception in this case
raise unittest.SkipTest('Skip this test for freebsd')
try:
INT_MAX = 2**32 - 1
huge_buffer = bytes(INT_MAX + 5)
except BaseException:
raise unittest.SkipTest('Machine cant allocate giant buffer for giant buffer test')
val = checksums.crc64nvme(huge_buffer)
self.assertEqual(0x2645c28052b1fbb0, val)
if __name__ == '__main__':
unittest.main()
|
