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
|