test: adding new tests and updating existing ones

tests/cli/test_cli.py

@@ -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

tests/cli/test_shell_functionality.py

@@ -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

tests/fetchers/async/test_camoufox_session.py

@@ -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")

tests/fetchers/async/test_dynamic_session.py

@@ -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")

tests/fetchers/async/test_requests_session.py

@@ -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

tests/fetchers/sync/test_requests_session.py

@@ -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