File: test_init.py

package info (click to toggle)
aiohttp-fast-zlib 0.3.0-1
  • links: PTS, VCS
  • area: main
  • in suites: sid, trixie
  • size: 304 kB
  • sloc: python: 186; makefile: 3
file content (153 lines) | stat: -rw-r--r-- 5,197 bytes parent folder | download
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