test: adding new tests and updating existing ones
This commit is contained in:
Karim shoair
@@ -0,0 +1,199 @@
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
from unittest.mock import patch, MagicMock
|
||||
import pytest_httpbin
|
||||
|
||||
from scrapling.cli import (
|
||||
install, shell, mcp,
|
||||
get, post, put, delete, fetch, stealthy_fetch
|
||||
)
|
||||
|
||||
|
||||
@pytest_httpbin.use_class_based_httpbin
|
||||
class TestCLI:
|
||||
"""Test CLI functionality"""
|
||||
|
||||
@pytest.fixture
|
||||
def html_url(self, httpbin):
|
||||
return f"{httpbin.url}/html"
|
||||
|
||||
@pytest.fixture
|
||||
def runner(self):
|
||||
return CliRunner()
|
||||
|
||||
def test_install_command(self, runner):
|
||||
"""Test install command"""
|
||||
result = runner.invoke(install)
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_shell_command(self, runner):
|
||||
"""Test shell command"""
|
||||
with patch('scrapling.core.shell.CustomShell') as mock_shell:
|
||||
mock_instance = MagicMock()
|
||||
mock_shell.return_value = mock_instance
|
||||
|
||||
result = runner.invoke(shell)
|
||||
assert result.exit_code == 0
|
||||
mock_instance.start.assert_called_once()
|
||||
|
||||
def test_mcp_command(self, runner):
|
||||
"""Test MCP command"""
|
||||
with patch('scrapling.core.ai.ScraplingMCPServer') as mock_server:
|
||||
mock_instance = MagicMock()
|
||||
mock_server.return_value = mock_instance
|
||||
|
||||
result = runner.invoke(mcp)
|
||||
assert result.exit_code == 0
|
||||
mock_instance.serve.assert_called_once()
|
||||
|
||||
def test_extract_get_command(self, runner, tmp_path, html_url):
|
||||
"""Test extract `get` command"""
|
||||
output_file = tmp_path / "output.md"
|
||||
|
||||
with patch('scrapling.fetchers.Fetcher.get') as mock_get:
|
||||
mock_response = MagicMock()
|
||||
mock_response.status = 200
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
with patch('scrapling.cli.Convertor.write_content_to_file'):
|
||||
result = runner.invoke(
|
||||
get,
|
||||
[html_url, str(output_file)]
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
|
||||
# Test with various options
|
||||
with patch('scrapling.fetchers.Fetcher.get') as mock_get:
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
with patch('scrapling.cli.Convertor.write_content_to_file'):
|
||||
result = runner.invoke(
|
||||
get,
|
||||
[
|
||||
html_url,
|
||||
str(output_file),
|
||||
'-H', 'User-Agent: Test',
|
||||
'--cookies', 'session=abc123',
|
||||
'--timeout', '60',
|
||||
'--proxy', 'http://proxy:8080',
|
||||
'-s', '.content',
|
||||
'-p', 'page=1'
|
||||
]
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_extract_post_command(self, runner, tmp_path, html_url):
|
||||
"""Test extract `post` command"""
|
||||
output_file = tmp_path / "output.html"
|
||||
|
||||
with patch('scrapling.fetchers.Fetcher.post') as mock_post:
|
||||
mock_response = MagicMock()
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
with patch('scrapling.cli.Convertor.write_content_to_file'):
|
||||
result = runner.invoke(
|
||||
post,
|
||||
[
|
||||
html_url,
|
||||
str(output_file),
|
||||
'-d', 'key=value',
|
||||
'-j', '{"data": "test"}'
|
||||
]
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_extract_put_command(self, runner, tmp_path, html_url):
|
||||
"""Test extract `put` command"""
|
||||
output_file = tmp_path / "output.html"
|
||||
|
||||
with patch('scrapling.fetchers.Fetcher.put') as mock_put:
|
||||
mock_response = MagicMock()
|
||||
mock_put.return_value = mock_response
|
||||
|
||||
with patch('scrapling.cli.Convertor.write_content_to_file'):
|
||||
result = runner.invoke(
|
||||
put,
|
||||
[
|
||||
html_url,
|
||||
str(output_file),
|
||||
'-d', 'key=value',
|
||||
'-j', '{"data": "test"}'
|
||||
]
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_extract_delete_command(self, runner, tmp_path, html_url):
|
||||
"""Test extract `delete` command"""
|
||||
output_file = tmp_path / "output.html"
|
||||
|
||||
with patch('scrapling.fetchers.Fetcher.delete') as mock_delete:
|
||||
mock_response = MagicMock()
|
||||
mock_delete.return_value = mock_response
|
||||
|
||||
with patch('scrapling.cli.Convertor.write_content_to_file'):
|
||||
result = runner.invoke(
|
||||
delete,
|
||||
[
|
||||
html_url,
|
||||
str(output_file)
|
||||
]
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_extract_fetch_command(self, runner, tmp_path, html_url):
|
||||
"""Test extract fetch command"""
|
||||
output_file = tmp_path / "output.txt"
|
||||
|
||||
with patch('scrapling.fetchers.DynamicFetcher.fetch') as mock_fetch:
|
||||
mock_response = MagicMock()
|
||||
mock_fetch.return_value = mock_response
|
||||
|
||||
with patch('scrapling.cli.Convertor.write_content_to_file'):
|
||||
result = runner.invoke(
|
||||
fetch,
|
||||
[
|
||||
html_url,
|
||||
str(output_file),
|
||||
'--headless',
|
||||
'--stealth',
|
||||
'--timeout', '60000'
|
||||
]
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_extract_stealthy_fetch_command(self, runner, tmp_path, html_url):
|
||||
"""Test extract fetch command"""
|
||||
output_file = tmp_path / "output.md"
|
||||
|
||||
with patch('scrapling.fetchers.StealthyFetcher.fetch') as mock_fetch:
|
||||
mock_response = MagicMock()
|
||||
mock_fetch.return_value = mock_response
|
||||
|
||||
with patch('scrapling.cli.Convertor.write_content_to_file'):
|
||||
result = runner.invoke(
|
||||
stealthy_fetch,
|
||||
[
|
||||
html_url,
|
||||
str(output_file),
|
||||
'--headless',
|
||||
'--css-selector', 'body',
|
||||
'--timeout', '60000'
|
||||
]
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_invalid_arguments(self, runner, html_url):
|
||||
"""Test invalid arguments handling"""
|
||||
# Missing required arguments
|
||||
result = runner.invoke(get)
|
||||
assert result.exit_code != 0
|
||||
|
||||
# Invalid output file extension
|
||||
with patch('scrapling.cli.Convertor.write_content_to_file') as mock_write:
|
||||
mock_write.side_effect = ValueError("Unknown file type")
|
||||
|
||||
_ = runner.invoke(
|
||||
get,
|
||||
[html_url, 'output.invalid']
|
||||
)
|
||||
# Should handle the error gracefully
|
||||
@@ -0,0 +1,200 @@
|
||||
import pytest
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
from scrapling.parser import Selector
|
||||
from scrapling.core.shell import CustomShell, CurlParser, Convertor
|
||||
|
||||
|
||||
class TestCurlParser:
|
||||
"""Test curl command parsing"""
|
||||
|
||||
@pytest.fixture
|
||||
def parser(self):
|
||||
return CurlParser()
|
||||
|
||||
def test_basic_curl_parse(self, parser):
|
||||
"""Test parsing basic curl commands"""
|
||||
# Simple GET
|
||||
curl_cmd = 'curl https://example.com'
|
||||
request = parser.parse(curl_cmd)
|
||||
|
||||
assert request.url == 'https://example.com'
|
||||
assert request.method == 'get'
|
||||
assert request.data is None
|
||||
|
||||
def test_curl_with_headers(self, parser):
|
||||
"""Test parsing curl with headers"""
|
||||
curl_cmd = '''curl https://example.com \
|
||||
-H "User-Agent: Mozilla/5.0" \
|
||||
-H "Accept: application/json"'''
|
||||
|
||||
request = parser.parse(curl_cmd)
|
||||
|
||||
assert request.headers['User-Agent'] == 'Mozilla/5.0'
|
||||
assert request.headers['Accept'] == 'application/json'
|
||||
|
||||
def test_curl_with_data(self, parser):
|
||||
"""Test parsing curl with data"""
|
||||
# Form data
|
||||
curl_cmd = 'curl https://example.com -X POST -d "key=value&foo=bar"'
|
||||
request = parser.parse(curl_cmd)
|
||||
|
||||
assert request.method == 'post'
|
||||
assert request.data == 'key=value&foo=bar'
|
||||
|
||||
# JSON data
|
||||
curl_cmd = """curl https://example.com -X POST --data-raw '{"key": "value"}'"""
|
||||
request = parser.parse(curl_cmd)
|
||||
|
||||
assert request.json_data == {"key": "value"}
|
||||
|
||||
def test_curl_with_cookies(self, parser):
|
||||
"""Test parsing curl with cookies"""
|
||||
curl_cmd = '''curl https://example.com \
|
||||
-H "Cookie: session=abc123; user=john" \
|
||||
-b "extra=cookie"'''
|
||||
|
||||
request = parser.parse(curl_cmd)
|
||||
|
||||
assert request.cookies['session'] == 'abc123'
|
||||
assert request.cookies['user'] == 'john'
|
||||
assert request.cookies['extra'] == 'cookie'
|
||||
|
||||
def test_curl_with_proxy(self, parser):
|
||||
"""Test parsing curl with proxy"""
|
||||
curl_cmd = 'curl https://example.com -x http://proxy:8080 -U user:pass'
|
||||
request = parser.parse(curl_cmd)
|
||||
|
||||
assert 'http://user:pass@proxy:8080' in request.proxy['http']
|
||||
|
||||
def test_curl2fetcher(self, parser):
|
||||
"""Test converting curl to fetcher request"""
|
||||
with patch('scrapling.fetchers.Fetcher.get') as mock_get:
|
||||
mock_response = MagicMock()
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
curl_cmd = 'curl https://example.com'
|
||||
_ = parser.convert2fetcher(curl_cmd)
|
||||
|
||||
mock_get.assert_called_once()
|
||||
|
||||
def test_invalid_curl_commands(self, parser):
|
||||
"""Test handling invalid curl commands"""
|
||||
# Invalid format
|
||||
with pytest.raises(AttributeError):
|
||||
parser.parse('not a curl command')
|
||||
|
||||
|
||||
class TestConvertor:
|
||||
"""Test content conversion functionality"""
|
||||
|
||||
@pytest.fixture
|
||||
def sample_html(self):
|
||||
return """
|
||||
<html>
|
||||
<body>
|
||||
<div class="content">
|
||||
<h1>Title</h1>
|
||||
<p>Some text content</p>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
def test_extract_markdown(self, sample_html):
|
||||
"""Test extracting content as Markdown"""
|
||||
page = Selector(sample_html)
|
||||
content = list(Convertor._extract_content(page, "markdown"))
|
||||
|
||||
assert len(content) > 0
|
||||
assert "Title\n=====" in content[0] # Markdown conversion
|
||||
|
||||
def test_extract_html(self, sample_html):
|
||||
"""Test extracting content as HTML"""
|
||||
page = Selector(sample_html)
|
||||
content = list(Convertor._extract_content(page, "html"))
|
||||
|
||||
assert len(content) > 0
|
||||
assert "<h1>Title</h1>" in content[0]
|
||||
|
||||
def test_extract_text(self, sample_html):
|
||||
"""Test extracting content as plain text"""
|
||||
page = Selector(sample_html)
|
||||
content = list(Convertor._extract_content(page, "text"))
|
||||
|
||||
assert len(content) > 0
|
||||
assert "Title" in content[0]
|
||||
assert "Some text content" in content[0]
|
||||
|
||||
def test_extract_with_selector(self, sample_html):
|
||||
"""Test extracting with CSS selector"""
|
||||
page = Selector(sample_html)
|
||||
content = list(Convertor._extract_content(
|
||||
page,
|
||||
"text",
|
||||
css_selector=".content"
|
||||
))
|
||||
|
||||
assert len(content) > 0
|
||||
|
||||
def test_write_to_file(self, sample_html, tmp_path):
|
||||
"""Test writing content to files"""
|
||||
page = Selector(sample_html)
|
||||
|
||||
# Test markdown
|
||||
md_file = tmp_path / "output.md"
|
||||
Convertor.write_content_to_file(page, str(md_file))
|
||||
assert md_file.exists()
|
||||
|
||||
# Test HTML
|
||||
html_file = tmp_path / "output.html"
|
||||
Convertor.write_content_to_file(page, str(html_file))
|
||||
assert html_file.exists()
|
||||
|
||||
# Test text
|
||||
txt_file = tmp_path / "output.txt"
|
||||
Convertor.write_content_to_file(page, str(txt_file))
|
||||
assert txt_file.exists()
|
||||
|
||||
def test_invalid_operations(self, sample_html):
|
||||
"""Test error handling in convertor"""
|
||||
page = Selector(sample_html)
|
||||
|
||||
# Invalid extraction type
|
||||
with pytest.raises(ValueError):
|
||||
list(Convertor._extract_content(page, "invalid"))
|
||||
|
||||
# Invalid filename
|
||||
with pytest.raises(ValueError):
|
||||
Convertor.write_content_to_file(page, "")
|
||||
|
||||
# Unknown file extension
|
||||
with pytest.raises(ValueError):
|
||||
Convertor.write_content_to_file(page, "output.xyz")
|
||||
|
||||
|
||||
class TestCustomShell:
|
||||
"""Test interactive shell functionality"""
|
||||
|
||||
def test_shell_initialization(self):
|
||||
"""Test shell initialization"""
|
||||
with patch('scrapling.core.shell.InteractiveShellEmbed'):
|
||||
shell = CustomShell(code="", log_level="debug")
|
||||
|
||||
assert shell.log_level == 10 # DEBUG level
|
||||
assert shell.page is None
|
||||
assert len(shell.pages) == 0
|
||||
|
||||
def test_shell_namespace(self):
|
||||
"""Test shell namespace creation"""
|
||||
with patch('scrapling.core.shell.InteractiveShellEmbed'):
|
||||
shell = CustomShell(code="")
|
||||
namespace = shell.get_namespace()
|
||||
|
||||
# Check all expected functions/classes are available
|
||||
assert 'get' in namespace
|
||||
assert 'post' in namespace
|
||||
assert 'Fetcher' in namespace
|
||||
assert 'DynamicFetcher' in namespace
|
||||
assert 'view' in namespace
|
||||
assert 'uncurl' in namespace
|
||||
@@ -0,0 +1,85 @@
|
||||
|
||||
import pytest
|
||||
import asyncio
|
||||
|
||||
import pytest_httpbin
|
||||
|
||||
from scrapling.engines import AsyncStealthySession
|
||||
|
||||
|
||||
@pytest_httpbin.use_class_based_httpbin
|
||||
@pytest.mark.asyncio
|
||||
class TestAsyncStealthySession:
|
||||
"""Test AsyncStealthySession"""
|
||||
|
||||
# The `AsyncStealthySession` is inheriting from `StealthySession` class so no need to repeat all the tests
|
||||
@pytest.fixture
|
||||
def urls(self, httpbin):
|
||||
return {
|
||||
"basic": f"{httpbin.url}/get",
|
||||
"html": f"{httpbin.url}/html",
|
||||
}
|
||||
|
||||
async def test_concurrent_async_requests(self, urls):
|
||||
"""Test concurrent requests with async session"""
|
||||
async with AsyncStealthySession(max_pages=3) as session:
|
||||
# Launch multiple concurrent requests
|
||||
tasks = [
|
||||
session.fetch(urls["basic"]),
|
||||
session.fetch(urls["html"]),
|
||||
session.fetch(urls["basic"])
|
||||
]
|
||||
|
||||
assert session.max_pages == 3
|
||||
assert session.page_pool.max_pages == 3
|
||||
assert session.context is not None
|
||||
|
||||
responses = await asyncio.gather(*tasks)
|
||||
|
||||
# All should succeed
|
||||
assert all(r.status == 200 for r in responses)
|
||||
|
||||
# Check pool stats
|
||||
stats = session.get_pool_stats()
|
||||
assert stats["total_pages"] <= 3
|
||||
|
||||
# After exit, should be closed
|
||||
assert session._closed is True
|
||||
|
||||
# Should raise RuntimeError when used after closing
|
||||
with pytest.raises(RuntimeError):
|
||||
await session.fetch(urls["basic"])
|
||||
|
||||
async def test_page_pool_management(self, urls):
|
||||
"""Test page pool creation and reuse"""
|
||||
async with AsyncStealthySession() as session:
|
||||
# The first request creates a page
|
||||
_ = await session.fetch(urls["basic"])
|
||||
assert session.page_pool.pages_count == 1
|
||||
|
||||
# The second request should reuse the page
|
||||
_ = await session.fetch(urls["html"])
|
||||
assert session.page_pool.pages_count == 1
|
||||
|
||||
# Check pool stats
|
||||
stats = session.get_pool_stats()
|
||||
assert stats["total_pages"] == 1
|
||||
assert stats["max_pages"] == 1
|
||||
|
||||
async def test_stealthy_session_with_options(self, urls):
|
||||
"""Test AsyncStealthySession with various options"""
|
||||
async with AsyncStealthySession(
|
||||
max_pages=1,
|
||||
block_images=True,
|
||||
disable_ads=True,
|
||||
humanize=True
|
||||
) as session:
|
||||
response = await session.fetch(urls["html"])
|
||||
assert response.status == 200
|
||||
|
||||
async def test_error_handling_in_fetch(self, urls):
|
||||
"""Test error handling during fetch"""
|
||||
async with AsyncStealthySession() as session:
|
||||
# Test with invalid URL
|
||||
with pytest.raises(Exception):
|
||||
await session.fetch("invalid://url")
|
||||
@@ -0,0 +1,84 @@
|
||||
import pytest
|
||||
import asyncio
|
||||
|
||||
import pytest_httpbin
|
||||
|
||||
from scrapling.engines import AsyncDynamicSession
|
||||
|
||||
|
||||
@pytest_httpbin.use_class_based_httpbin
|
||||
@pytest.mark.asyncio
|
||||
class TestAsyncDynamicSession:
|
||||
"""Test AsyncDynamicSession"""
|
||||
|
||||
# The `AsyncDynamicSession` is inheriting from `DynamicSession` class so no need to repeat all the tests
|
||||
@pytest.fixture
|
||||
def urls(self, httpbin):
|
||||
return {
|
||||
"basic": f"{httpbin.url}/get",
|
||||
"html": f"{httpbin.url}/html",
|
||||
}
|
||||
|
||||
async def test_concurrent_async_requests(self, urls):
|
||||
"""Test concurrent requests with async session"""
|
||||
async with AsyncDynamicSession(max_pages=3) as session:
|
||||
# Launch multiple concurrent requests
|
||||
tasks = [
|
||||
session.fetch(urls["basic"]),
|
||||
session.fetch(urls["html"]),
|
||||
session.fetch(urls["basic"])
|
||||
]
|
||||
|
||||
assert session.max_pages == 3
|
||||
assert session.page_pool.max_pages == 3
|
||||
assert session.context is not None
|
||||
|
||||
responses = await asyncio.gather(*tasks)
|
||||
|
||||
# All should succeed
|
||||
assert all(r.status == 200 for r in responses)
|
||||
|
||||
# Check pool stats
|
||||
stats = session.get_pool_stats()
|
||||
assert stats["total_pages"] <= 3
|
||||
|
||||
# After exit, should be closed
|
||||
assert session._closed is True
|
||||
|
||||
# Should raise RuntimeError when used after closing
|
||||
with pytest.raises(RuntimeError):
|
||||
await session.fetch(urls["basic"])
|
||||
|
||||
async def test_page_pool_management(self, urls):
|
||||
"""Test page pool creation and reuse"""
|
||||
async with AsyncDynamicSession() as session:
|
||||
# The first request creates a page
|
||||
_ = await session.fetch(urls["basic"])
|
||||
assert session.page_pool.pages_count == 1
|
||||
|
||||
# The second request should reuse the page
|
||||
_ = await session.fetch(urls["html"])
|
||||
assert session.page_pool.pages_count == 1
|
||||
|
||||
# Check pool stats
|
||||
stats = session.get_pool_stats()
|
||||
assert stats["total_pages"] == 1
|
||||
assert stats["max_pages"] == 1
|
||||
|
||||
async def test_dynamic_session_with_options(self, urls):
|
||||
"""Test AsyncDynamicSession with various options"""
|
||||
async with AsyncDynamicSession(
|
||||
headless=False,
|
||||
stealth=True,
|
||||
disable_resources=True,
|
||||
extra_headers={"X-Test": "value"}
|
||||
) as session:
|
||||
response = await session.fetch(urls["html"])
|
||||
assert response.status == 200
|
||||
|
||||
async def test_error_handling_in_fetch(self, urls):
|
||||
"""Test error handling during fetch"""
|
||||
async with AsyncDynamicSession() as session:
|
||||
# Test with invalid URL
|
||||
with pytest.raises(Exception):
|
||||
await session.fetch("invalid://url")
|
||||
@@ -0,0 +1,17 @@
|
||||
import pytest
|
||||
|
||||
|
||||
from scrapling.engines.static import AsyncFetcherClient
|
||||
|
||||
|
||||
class TestFetcherSession:
|
||||
"""Test FetcherSession functionality"""
|
||||
|
||||
def test_async_fetcher_client_creation(self):
|
||||
"""Test AsyncFetcherClient creation"""
|
||||
client = AsyncFetcherClient()
|
||||
|
||||
# Should not have context manager methods
|
||||
assert client.__aenter__ is None
|
||||
assert client.__aexit__ is None
|
||||
assert client._async_curl_session is True # Special marker
|
||||
@@ -1,7 +1,7 @@
|
||||
import pytest
|
||||
|
||||
|
||||
from scrapling.engines.static import FetcherSession, FetcherClient, AsyncFetcherClient
|
||||
from scrapling.engines.static import FetcherSession, FetcherClient
|
||||
|
||||
|
||||
class TestFetcherSession:
|
||||
@@ -45,12 +45,3 @@ class TestFetcherSession:
|
||||
assert client.__enter__ is None
|
||||
assert client.__exit__ is None
|
||||
assert client._curl_session is True # Special marker
|
||||
|
||||
def test_async_fetcher_client_creation(self):
|
||||
"""Test AsyncFetcherClient creation"""
|
||||
client = AsyncFetcherClient()
|
||||
|
||||
# Should not have context manager methods
|
||||
assert client.__aenter__ is None
|
||||
assert client.__aexit__ is None
|
||||
assert client._async_curl_session is True # Special marker
|
||||
|
||||
Reference in New Issue
Block a user