File size: 8,512 Bytes
76f79da
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""Tests for the blocklist module."""
from unittest.mock import Mock, patch

import pytest
from urllib3.exceptions import HTTPError

from lib.blocklist import BlocklistData, _parse_block_list_from_url, OnlineBlocklist


def test_parse_block_list_from_url_success() -> None:
    """Test parsing blocklist from URL with successful response."""
    mock_response = Mock()
    mock_response.status_code = 200
    mock_response.text = "user1\nuser2\nuser3"
    mock_response.headers = {"ETag": "test-etag-123"}
    mock_response.raise_for_status = Mock()

    old_data = BlocklistData([], None)

    with patch("lib.blocklist.requests.get", return_value=mock_response):
        result = _parse_block_list_from_url("http://example.com/blocklist", old_data)

    assert result.users == ["user1", "user2", "user3"]
    assert result.etag == "test-etag-123"


def test_parse_block_list_from_url__several_line_breaks() -> None:
    """Test parsing blocklist from URL with several line-breaks."""
    mock_response = Mock()
    mock_response.status_code = 200
    mock_response.headers = {"ETag": "test-etag-123"}
    mock_response.raise_for_status = Mock()

    old_data = BlocklistData([], None)

    texts = [
        "user1\nuser2\nuser3",
        "user1\r\nuser2\r\n\r\nuser3",
        "user1\ruser2\r\ruser3",
        "user1\nuser2\nuser3",
    ]
    for text in texts:
        mock_response.text = text

        with patch("lib.blocklist.requests.get", return_value=mock_response):
            result = _parse_block_list_from_url("http://example.com/blocklist", old_data)

        assert result.users == ["user1", "user2", "user3"]
        assert result.etag == "test-etag-123"


def test_parse_block_list_from_url_not_modified() -> None:
    """Test parsing blocklist returns cached data when not modified (304)."""
    mock_response = Mock()
    mock_response.status_code = 304
    mock_response.raise_for_status = Mock()

    old_data = BlocklistData(["cached_user1", "cached_user2"], "old-etag")

    with patch("lib.blocklist.requests.get", return_value=mock_response):
        result = _parse_block_list_from_url("http://example.com/blocklist", old_data)

    assert result.users == ["cached_user1", "cached_user2"]
    assert result.etag == "old-etag"


def test_parse_block_list_from_url_with_etag_header() -> None:
    """Test that ETag is sent in request headers when available."""
    mock_response = Mock()
    mock_response.status_code = 200
    mock_response.text = "user1"
    mock_response.headers = {}
    mock_response.raise_for_status = Mock()

    old_data = BlocklistData([], "existing-etag")

    with patch("lib.blocklist.requests.get", return_value=mock_response) as mock_get:
        _parse_block_list_from_url("http://example.com/blocklist", old_data)
        mock_get.assert_called_once_with(
            "http://example.com/blocklist",
            headers={"If-None-Match": "existing-etag"},
            timeout=15,
        )


def test_parse_block_list_from_url_without_etag_header() -> None:
    """Test that no ETag header is sent when old_data has no ETag."""
    mock_response = Mock()
    mock_response.status_code = 200
    mock_response.text = "user1"
    mock_response.headers = {}
    mock_response.raise_for_status = Mock()

    old_data = BlocklistData([], None)

    with patch("lib.blocklist.requests.get", return_value=mock_response) as mock_get:
        _parse_block_list_from_url("http://example.com/blocklist", old_data)
        mock_get.assert_called_once_with(
            "http://example.com/blocklist",
            headers={},
            timeout=15
        )


def test_parse_block_list_from_url_strips_whitespace() -> None:
    """Test that usernames are stripped of leading/trailing whitespace."""
    mock_response = Mock()
    mock_response.status_code = 200
    mock_response.text = "  user1  \n  user2\t\nuser3  "
    mock_response.headers = {}
    mock_response.raise_for_status = Mock()

    old_data = BlocklistData([], None)

    with patch("lib.blocklist.requests.get", return_value=mock_response):
        result = _parse_block_list_from_url("http://example.com/blocklist", old_data)

    assert result.users == ["user1", "user2", "user3"]


def test_parse_block_list_from_url_empty_response() -> None:
    """Test parsing blocklist with empty response."""
    mock_response = Mock()
    mock_response.status_code = 200
    mock_response.text = ""
    mock_response.headers = {}
    mock_response.raise_for_status = Mock()

    old_data = BlocklistData([], None)

    with patch("lib.blocklist.requests.get", return_value=mock_response):
        result = _parse_block_list_from_url("http://example.com/blocklist", old_data)

    assert result.users == []


def test_parse_block_list_from_url_no_etag_in_response() -> None:
    """Test parsing blocklist when response has no ETag header."""
    mock_response = Mock()
    mock_response.status_code = 200
    mock_response.text = "user1\nuser2"
    mock_response.headers = {}
    mock_response.raise_for_status = Mock()

    old_data = BlocklistData([], None)

    with patch("lib.blocklist.requests.get", return_value=mock_response):
        result = _parse_block_list_from_url("http://example.com/blocklist", old_data)

    assert result.users == ["user1", "user2"]
    assert result.etag is None


def test_parse_block_list_from_url_multiline_with_blanks() -> None:
    """Test parsing blocklist that contains blank lines."""
    mock_response = Mock()
    mock_response.status_code = 200
    mock_response.text = "user1\n\nuser2\n\n\nuser3"
    mock_response.headers = {}
    mock_response.raise_for_status = Mock()

    old_data = BlocklistData([], None)

    with patch("lib.blocklist.requests.get", return_value=mock_response):
        result = _parse_block_list_from_url("http://example.com/blocklist", old_data)

    assert result.users == ["user1", "user2", "user3"]


def test_parse_block_list_from_url__exception() -> None:
    """Test parsing blocklist from URL with exception."""
    mock_response = Mock()
    mock_response.status_code = 500
    mock_response.raise_for_status.side_effect = HTTPError("500 Internal Server Error")

    old_data = BlocklistData([], None)

    with patch("lib.blocklist.requests.get", return_value=mock_response), pytest.raises(HTTPError):
        _parse_block_list_from_url("http://example.com/blocklist", old_data)


def test_online_blocklist_refresh_success() -> None:
    """Test successful refresh of all blocklists."""
    mock_response1 = Mock()
    mock_response1.status_code = 200
    mock_response1.text = "user1\nuser2"
    mock_response1.headers = {"ETag": "etag1"}
    mock_response1.raise_for_status = Mock()

    mock_response2 = Mock()
    mock_response2.status_code = 200
    mock_response2.text = "user3\nuser4\nuser5"
    mock_response2.headers = {"ETag": "etag2"}
    mock_response2.raise_for_status = Mock()

    with patch("lib.blocklist.requests.get") as mock_get:
        mock_get.side_effect = [mock_response1, mock_response2]
        blocklist = OnlineBlocklist(["http://example.com/list1", "http://example.com/list2"])

    assert blocklist.blocklist["http://example.com/list1"].users == ["user1", "user2"]
    assert blocklist.blocklist["http://example.com/list1"].etag == "etag1"
    assert blocklist.blocklist["http://example.com/list2"].users == ["user3", "user4", "user5"]
    assert blocklist.blocklist["http://example.com/list2"].etag == "etag2"


def test_online_blocklist_refresh_partial_failure() -> None:
    """Test refresh when some blocklists fail to load."""
    mock_response_success = Mock()
    mock_response_success.status_code = 200
    mock_response_success.text = "user1\nuser2"
    mock_response_success.headers = {"ETag": "etag1"}
    mock_response_success.raise_for_status = Mock()

    mock_response_fail = Mock()
    mock_response_fail.raise_for_status.side_effect = HTTPError("500 Server Error")

    with patch("lib.blocklist.requests.get") as mock_get:
        # First call succeeds, second call fails
        mock_get.side_effect = [mock_response_success, mock_response_fail]
        blocklist = OnlineBlocklist(["http://example.com/list1", "http://example.com/list2"])

    # First blocklist should be loaded
    assert blocklist.blocklist["http://example.com/list1"].users == ["user1", "user2"]

    # Second blocklist should remain empty (failed to load)
    assert blocklist.blocklist["http://example.com/list2"].users == []
    assert blocklist.blocklist["http://example.com/list2"].etag is None