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
|
import zlib as zlib_original
from unittest.mock import patch
import aiohttp.compression_utils
import aiohttp.http_websocket
import pytest
import aiohttp_fast_zlib
try:
from isal import (
isal_zlib as expected_zlib,
)
except ImportError:
from zlib_ng import zlib_ng as expected_zlib
@pytest.mark.skipif(
aiohttp_fast_zlib._AIOHTTP_VERSION >= (3, 11),
reason="Only works with aiohttp less than 3.11+",
)
def test_enable_disable_pre_311():
"""Test enable/disable."""
assert aiohttp.http_websocket.zlib is zlib_original
aiohttp_fast_zlib.enable()
assert aiohttp.http_websocket.zlib is expected_zlib
aiohttp_fast_zlib.disable()
assert aiohttp.http_websocket.zlib is zlib_original
aiohttp_fast_zlib.enable()
assert aiohttp.http_websocket.zlib is expected_zlib
aiohttp_fast_zlib.disable()
@pytest.mark.skipif(
aiohttp_fast_zlib._AIOHTTP_VERSION < (3, 11)
or aiohttp_fast_zlib._AIOHTTP_VERSION >= (3, 12),
reason="Only works with aiohttp 3.11.x",
)
def test_enable_disable_311():
"""Test enable/disable for aiohttp 3.11.x."""
from aiohttp._websocket import writer
assert writer.zlib is zlib_original
aiohttp_fast_zlib.enable()
assert writer.zlib is expected_zlib
aiohttp_fast_zlib.disable()
assert writer.zlib is zlib_original
aiohttp_fast_zlib.enable()
assert writer.zlib is expected_zlib
aiohttp_fast_zlib.disable()
@pytest.mark.skipif(
aiohttp_fast_zlib._AIOHTTP_VERSION < (3, 11)
or aiohttp_fast_zlib._AIOHTTP_VERSION >= (3, 12),
reason="Only works with aiohttp 3.11.x",
)
def test_enable_disable_when_all_missing_311():
"""Test enable/disable for aiohttp 3.11.x when all fast libs are missing."""
from aiohttp._websocket import writer
with patch.object(aiohttp_fast_zlib, "best_zlib", zlib_original):
assert writer.zlib is zlib_original
aiohttp_fast_zlib.enable()
assert writer.zlib is zlib_original
aiohttp_fast_zlib.disable()
assert writer.zlib is zlib_original
aiohttp_fast_zlib.enable()
assert writer.zlib is zlib_original
aiohttp_fast_zlib.disable()
assert writer.zlib is zlib_original
@pytest.mark.skipif(
aiohttp_fast_zlib._AIOHTTP_VERSION >= (3, 11),
reason="Only works with aiohttp less than 3.11+",
)
def test_enable_disable_when_all_missing_pre_311():
"""Test enable/disable."""
with patch.object(aiohttp_fast_zlib, "best_zlib", zlib_original):
assert aiohttp.http_websocket.zlib is zlib_original
aiohttp_fast_zlib.enable()
assert aiohttp.http_websocket.zlib is zlib_original
aiohttp_fast_zlib.disable()
assert aiohttp.http_websocket.zlib is zlib_original
aiohttp_fast_zlib.enable()
assert aiohttp.http_websocket.zlib is zlib_original
aiohttp_fast_zlib.disable()
assert aiohttp.http_websocket.zlib is zlib_original
@pytest.mark.skipif(
aiohttp_fast_zlib._AIOHTTP_VERSION >= (3, 12),
reason="Only works with aiohttp < 3.12",
)
def test_enable_disable_when_all_missing():
"""Test enable/disable when all fast libs are missing."""
with patch.object(aiohttp_fast_zlib, "best_zlib", zlib_original):
assert aiohttp.compression_utils.zlib is zlib_original
aiohttp_fast_zlib.enable()
assert aiohttp.compression_utils.zlib is zlib_original
aiohttp_fast_zlib.disable()
assert aiohttp.compression_utils.zlib is zlib_original
aiohttp_fast_zlib.enable()
assert aiohttp.compression_utils.zlib is zlib_original
aiohttp_fast_zlib.disable()
assert aiohttp.compression_utils.zlib is zlib_original
@pytest.mark.skipif(
aiohttp_fast_zlib._AIOHTTP_VERSION < (3, 12),
reason="Only works with aiohttp >= 3.12",
)
def test_enable_disable_312_plus():
"""Test enable/disable for aiohttp 3.12+ with native set_zlib_backend."""
from aiohttp.compression_utils import ZLibBackend
# Test enable
aiohttp_fast_zlib.enable()
assert ZLibBackend._zlib_backend is expected_zlib
# Test disable
aiohttp_fast_zlib.disable()
assert ZLibBackend._zlib_backend is zlib_original
# Test enable again
aiohttp_fast_zlib.enable()
assert ZLibBackend._zlib_backend is expected_zlib
# Clean up
aiohttp_fast_zlib.disable()
assert ZLibBackend._zlib_backend is zlib_original
@pytest.mark.skipif(
aiohttp_fast_zlib._AIOHTTP_VERSION < (3, 12),
reason="Only works with aiohttp >= 3.12",
)
def test_enable_disable_when_all_missing_312_plus():
"""Test enable/disable for aiohttp 3.12+ when all fast libs are missing."""
from aiohttp.compression_utils import ZLibBackend
# Store the original backend
original_backend = ZLibBackend._zlib_backend
with patch.object(aiohttp_fast_zlib, "best_zlib", zlib_original):
# Test enable - should not change backend when best_zlib is zlib_original
aiohttp_fast_zlib.enable()
assert ZLibBackend._zlib_backend is original_backend
# Test disable - should not change backend either
aiohttp_fast_zlib.disable()
assert ZLibBackend._zlib_backend is original_backend
|
