tests: Multiple changes to tests
- Remove all tests for the old encoding logic - Stop using the `nopecha` test page to test Cloudflare solver - Remove useless tests like testing for infinite timeout - Fixes to make the code compatible with new changes
This commit is contained in:
@@ -14,6 +14,7 @@ def configure_selector_mock():
|
|||||||
"""Helper function to create a properly configured Selector mock"""
|
"""Helper function to create a properly configured Selector mock"""
|
||||||
mock_response = MagicMock(spec=Selector)
|
mock_response = MagicMock(spec=Selector)
|
||||||
mock_response.body = "<html><body>Test content</body></html>"
|
mock_response.body = "<html><body>Test content</body></html>"
|
||||||
|
mock_response.encoding = "utf-8"
|
||||||
mock_response.get_all_text.return_value = "Test content"
|
mock_response.get_all_text.return_value = "Test content"
|
||||||
mock_response.css_first.return_value = mock_response
|
mock_response.css_first.return_value = mock_response
|
||||||
mock_response.css.return_value = [mock_response]
|
mock_response.css.return_value = [mock_response]
|
||||||
|
|||||||
@@ -1,3 +1,4 @@
|
|||||||
|
from playwright._impl._errors import TimeoutError
|
||||||
import pytest
|
import pytest
|
||||||
import pytest_httpbin
|
import pytest_httpbin
|
||||||
|
|
||||||
@@ -23,14 +24,9 @@ class TestStealthyFetcher:
|
|||||||
"basic_url": f"{url}/get",
|
"basic_url": f"{url}/get",
|
||||||
"html_url": f"{url}/html",
|
"html_url": f"{url}/html",
|
||||||
"delayed_url": f"{url}/delay/10", # 10 Seconds delay response
|
"delayed_url": f"{url}/delay/10", # 10 Seconds delay response
|
||||||
"cookies_url": f"{url}/cookies/set/test/value",
|
"cookies_url": f"{url}/cookies/set/test/value"
|
||||||
"cloudflare_url": "https://nopecha.com/demo/cloudflare", # Interactive turnstile page
|
|
||||||
}
|
}
|
||||||
|
|
||||||
async def test_cloudflare_fetch(self, fetcher, urls):
|
|
||||||
"""Test if Cloudflare bypass is working"""
|
|
||||||
assert (await fetcher.async_fetch(urls["cloudflare_url"], solve_cloudflare=True)).status == 200
|
|
||||||
|
|
||||||
async def test_basic_fetch(self, fetcher, urls):
|
async def test_basic_fetch(self, fetcher, urls):
|
||||||
"""Test doing a basic fetch request with multiple statuses"""
|
"""Test doing a basic fetch request with multiple statuses"""
|
||||||
assert (await fetcher.async_fetch(urls["status_200"])).status == 200
|
assert (await fetcher.async_fetch(urls["status_200"])).status == 200
|
||||||
@@ -86,9 +82,3 @@ class TestStealthyFetcher:
|
|||||||
**kwargs
|
**kwargs
|
||||||
)
|
)
|
||||||
assert response.status == 200
|
assert response.status == 200
|
||||||
|
|
||||||
async def test_infinite_timeout(self, fetcher, urls):
|
|
||||||
"""Test if infinite timeout breaks the code or not"""
|
|
||||||
assert (
|
|
||||||
await fetcher.async_fetch(urls["delayed_url"], timeout=0)
|
|
||||||
).status == 200
|
|
||||||
|
|||||||
@@ -90,9 +90,3 @@ class TestDynamicFetcherAsync:
|
|||||||
|
|
||||||
with pytest.raises(Exception):
|
with pytest.raises(Exception):
|
||||||
await fetcher.async_fetch(urls["html_url"], cdp_url="ws://blahblah")
|
await fetcher.async_fetch(urls["html_url"], cdp_url="ws://blahblah")
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_infinite_timeout(self, fetcher, urls):
|
|
||||||
"""Test if infinite timeout breaks the code or not"""
|
|
||||||
response = await fetcher.async_fetch(urls["delayed_url"], timeout=0)
|
|
||||||
assert response.status == 200
|
|
||||||
|
|||||||
@@ -77,7 +77,3 @@ class TestStealthyFetcher:
|
|||||||
**kwargs
|
**kwargs
|
||||||
)
|
)
|
||||||
assert response.status == 200
|
assert response.status == 200
|
||||||
|
|
||||||
def test_infinite_timeout(self, fetcher):
|
|
||||||
"""Test if infinite timeout breaks the code or not"""
|
|
||||||
assert fetcher.fetch(self.delayed_url, timeout=0).status == 200
|
|
||||||
|
|||||||
@@ -1,7 +1,6 @@
|
|||||||
from unittest.mock import Mock
|
from unittest.mock import Mock
|
||||||
|
|
||||||
from scrapling.parser import Selector
|
from scrapling.parser import Selector
|
||||||
from scrapling.engines.toolbelt.custom import ResponseEncoding
|
|
||||||
from scrapling.engines.toolbelt.convertor import ResponseFactory, Response
|
from scrapling.engines.toolbelt.convertor import ResponseFactory, Response
|
||||||
|
|
||||||
|
|
||||||
@@ -32,20 +31,6 @@ class TestResponseFactory:
|
|||||||
assert response.url == "https://example.com"
|
assert response.url == "https://example.com"
|
||||||
assert isinstance(response, Response)
|
assert isinstance(response, Response)
|
||||||
|
|
||||||
def test_response_encoding_edge_cases(self):
|
|
||||||
"""Test response encoding handling"""
|
|
||||||
# Test various content types
|
|
||||||
test_cases = [
|
|
||||||
(None, "utf-8"),
|
|
||||||
("", "utf-8"),
|
|
||||||
("text/html; charset=invalid", "utf-8"),
|
|
||||||
("application/octet-stream", "utf-8"),
|
|
||||||
]
|
|
||||||
|
|
||||||
for content_type, expected in test_cases:
|
|
||||||
encoding = ResponseEncoding.get_value(content_type)
|
|
||||||
assert encoding == expected
|
|
||||||
|
|
||||||
def test_response_history_processing(self):
|
def test_response_history_processing(self):
|
||||||
"""Test processing response history"""
|
"""Test processing response history"""
|
||||||
# Mock responses with redirects
|
# Mock responses with redirects
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
import pytest
|
import pytest
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
from scrapling.engines.toolbelt.custom import ResponseEncoding, StatusText, Response
|
from scrapling.engines.toolbelt.custom import StatusText, Response
|
||||||
from scrapling.engines.toolbelt.navigation import (
|
from scrapling.engines.toolbelt.navigation import (
|
||||||
construct_proxy_dict,
|
construct_proxy_dict,
|
||||||
js_bypass_path
|
js_bypass_path
|
||||||
@@ -131,12 +131,6 @@ def status_map():
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
def test_parsing_content_type(content_type_map):
|
|
||||||
"""Test if parsing different types of 'content-type' returns the expected result"""
|
|
||||||
for header_value, expected_encoding in content_type_map.items():
|
|
||||||
assert ResponseEncoding.get_value(header_value) == expected_encoding
|
|
||||||
|
|
||||||
|
|
||||||
def test_parsing_response_status(status_map):
|
def test_parsing_response_status(status_map):
|
||||||
"""Test if using different http responses' status codes returns the expected result"""
|
"""Test if using different http responses' status codes returns the expected result"""
|
||||||
for status_code, expected_status_text in status_map.items():
|
for status_code, expected_status_text in status_map.items():
|
||||||
|
|||||||
@@ -99,7 +99,7 @@ class TestAdvancedSelectors:
|
|||||||
keep_comments=False,
|
keep_comments=False,
|
||||||
keep_cdata=False
|
keep_cdata=False
|
||||||
)
|
)
|
||||||
content = page.body
|
content = page.html_content
|
||||||
assert "Comment" not in content
|
assert "Comment" not in content
|
||||||
|
|
||||||
def test_advanced_xpath_variables(self, complex_html):
|
def test_advanced_xpath_variables(self, complex_html):
|
||||||
|
|||||||
Reference in New Issue
Block a user