diff --git a/tests/ai/test_ai_mcp.py b/tests/ai/test_ai_mcp.py index 92f5886..d7f8b84 100644 --- a/tests/ai/test_ai_mcp.py +++ b/tests/ai/test_ai_mcp.py @@ -1,12 +1,18 @@ import pytest +import pytest_httpbin from unittest.mock import Mock, patch from scrapling.core.ai import ScraplingMCPServer, ResponseModel +@pytest_httpbin.use_class_based_httpbin class TestMCPServer: """Test MCP server functionality""" + @pytest.fixture(scope="class") + def test_url(self, httpbin): + return f"{httpbin.url}/html" + @pytest.fixture def server(self): return ScraplingMCPServer() @@ -16,71 +22,46 @@ class TestMCPServer: assert server._server is not None assert server._server.name == "Scrapling" - def test_get_tool(self): + def test_get_tool(self, server, test_url): """Test the get tool method""" - with patch('scrapling.fetchers.Fetcher.get') as mock_get: - mock_response = Mock() - mock_response.status = 200 - mock_response.url = "https://example.com" - mock_get.return_value = mock_response - - with patch('scrapling.core.ai.Convertor._extract_content') as mock_extract: - mock_extract.return_value = iter(["Content"]) - - result = ScraplingMCPServer.get( - url="https://example.com", - extraction_type="markdown" - ) - - assert isinstance(result, ResponseModel) - assert result.status == 200 - assert result.url == "https://example.com" + result = server.get(url=test_url, extraction_type="markdown") + assert isinstance(result, ResponseModel) + assert result.status == 200 + assert result.url == test_url @pytest.mark.asyncio - async def test_bulk_get_tool(self): + async def test_bulk_get_tool(self, server, test_url): """Test the bulk_get tool method""" - with patch('scrapling.engines.FetcherSession') as mock_session: - mock_instance = Mock() - mock_session.return_value.__aenter__.return_value = mock_instance + results = await server.bulk_get(urls=(test_url, test_url), extraction_type="html") - # Mock async get method - async def mock_async_get(*args, **kwargs): - mock_resp = Mock() - mock_resp.status = 200 - mock_resp.url = args[0] - return mock_resp - - mock_instance.get = mock_async_get - - with patch('scrapling.core.ai.Convertor._extract_content') as mock_extract: - mock_extract.return_value = iter(["Content"]) - - results = await ScraplingMCPServer.bulk_get( - urls=("https://example1.com", "https://example2.com"), - extraction_type="html" - ) - - assert len(results) == 2 - assert all(isinstance(r, ResponseModel) for r in results) + assert len(results) == 2 + assert all(isinstance(r, ResponseModel) for r in results) @pytest.mark.asyncio - async def test_fetch_tool(self): + async def test_fetch_tool(self, server, test_url): """Test the fetch tool method""" - with patch('scrapling.fetchers.DynamicFetcher.async_fetch') as mock_fetch: - mock_response = Mock() - mock_response.status = 200 - mock_response.url = "https://example.com" - mock_fetch.return_value = mock_response + result = await server.fetch(url=test_url, headless=True) + assert isinstance(result, ResponseModel) + assert result.status == 200 - with patch('scrapling.core.ai.Convertor._extract_content') as mock_extract: - mock_extract.return_value = iter(["Content"]) + @pytest.mark.asyncio + async def test_bulk_fetch_tool(self, server, test_url): + """Test the bulk_fetch tool method""" + result = await server.bulk_fetch(urls=(test_url, test_url), headless=True) + assert all(isinstance(r, ResponseModel) for r in result) - result = await ScraplingMCPServer.fetch( - url="https://example.com", - headless=True - ) + @pytest.mark.asyncio + async def test_stealthy_fetch_tool(self, server, test_url): + """Test the stealthy_fetch tool method""" + result = await server.stealthy_fetch(url=test_url, headless=True) + assert isinstance(result, ResponseModel) + assert result.status == 200 - assert isinstance(result, ResponseModel) + @pytest.mark.asyncio + async def test_bulk_stealthy_fetch_tool(self, server, test_url): + """Test the bulk_stealthy_fetch tool method""" + result = await server.bulk_stealthy_fetch(urls=(test_url, test_url), headless=True) + assert all(isinstance(r, ResponseModel) for r in result) def test_serve_method(self, server): """Test the serve method""" diff --git a/tests/core/__init__.py b/tests/core/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/core/test_shell_core.py b/tests/core/test_shell_core.py new file mode 100644 index 0000000..4578406 --- /dev/null +++ b/tests/core/test_shell_core.py @@ -0,0 +1,243 @@ +import pytest + +from scrapling.core.shell import ( + _CookieParser, + _ParseHeaders, + Request, + _known_logging_levels, +) + + +class TestCookieParser: + """Test cookie parsing functionality""" + + def test_simple_cookie_parsing(self): + """Test parsing a simple cookie""" + cookie_string = "session_id=abc123" + cookies = list(_CookieParser(cookie_string)) + assert len(cookies) == 1 + assert cookies[0] == ("session_id", "abc123") + + def test_multiple_cookies_parsing(self): + """Test parsing multiple cookies""" + cookie_string = "session_id=abc123; theme=dark; lang=en" + cookies = list(_CookieParser(cookie_string)) + assert len(cookies) == 3 + cookie_dict = dict(cookies) + assert cookie_dict["session_id"] == "abc123" + assert cookie_dict["theme"] == "dark" + assert cookie_dict["lang"] == "en" + + def test_cookie_with_attributes(self): + """Test parsing cookies with attributes""" + cookie_string = "session_id=abc123; Path=/; HttpOnly; Secure" + cookies = list(_CookieParser(cookie_string)) + assert len(cookies) == 1 + assert cookies[0] == ("session_id", "abc123") + + def test_empty_cookie_string(self): + """Test parsing empty cookie string""" + cookies = list(_CookieParser("")) + assert len(cookies) == 0 + + def test_malformed_cookie_handling(self): + """Test handling of malformed cookies""" + # Should not raise exception but may return an empty list + cookies = list(_CookieParser("invalid_cookie_format")) + assert isinstance(cookies, list) + + +class TestParseHeaders: + """Test header parsing functionality""" + + def test_simple_headers(self): + """Test parsing simple headers""" + header_lines = [ + "Content-Type: text/html", + "Content-Length: 1234", + "User-Agent: TestAgent/1.0" + ] + headers, cookies = _ParseHeaders(header_lines) + + assert headers["Content-Type"] == "text/html" + assert headers["Content-Length"] == "1234" + assert headers["User-Agent"] == "TestAgent/1.0" + assert len(cookies) == 0 + + def test_headers_with_cookies(self): + """Test parsing headers with cookie headers""" + header_lines = [ + "Content-Type: text/html", + "Set-Cookie: session_id=abc123", + "Set-Cookie: theme=dark; Path=/", + ] + headers, cookies = _ParseHeaders(header_lines) + + assert headers["Content-Type"] == "text/html" + assert "Set-Cookie" in headers # Should contain the first Set-Cookie + # Cookie parsing behavior depends on implementation + + def test_headers_without_colons(self): + """Test headers without colons""" + header_lines = [ + "Content-Type: text/html", + "InvalidHeader;", # Header ending with semicolon + ] + headers, cookies = _ParseHeaders(header_lines) + + assert headers["Content-Type"] == "text/html" + assert "InvalidHeader" in headers + assert headers["InvalidHeader"] == "" + + def test_invalid_header_format(self): + """Test invalid header format raises error""" + header_lines = [ + "Content-Type: text/html", + "InvalidHeaderWithoutColon", # No colon, no semicolon + ] + + with pytest.raises(ValueError, match="Could not parse header without colon"): + _ParseHeaders(header_lines) + + def test_headers_with_multiple_colons(self): + """Test headers with multiple colons""" + header_lines = [ + "Authorization: Bearer: token123", + "X-Custom: value:with:colons", + ] + headers, cookies = _ParseHeaders(header_lines) + + assert headers["Authorization"] == "Bearer: token123" + assert headers["X-Custom"] == "value:with:colons" + + def test_headers_with_whitespace(self): + """Test headers with extra whitespace""" + header_lines = [ + " Content-Type : text/html ", + "\tUser-Agent\t:\tTestAgent/1.0\t", + ] + headers, cookies = _ParseHeaders(header_lines) + + # Should handle whitespace correctly + assert "Content-Type" in headers or " Content-Type " in headers + assert "text/html" in str(headers.values()) or " text/html " in str(headers.values()) + + def test_parse_cookies_disabled(self): + """Test parsing with cookies disabled""" + header_lines = [ + "Content-Type: text/html", + "Set-Cookie: session_id=abc123", + ] + headers, cookies = _ParseHeaders(header_lines, parse_cookies=False) + + assert headers["Content-Type"] == "text/html" + # Cookie parsing behavior when disabled + assert len(cookies) == 0 or "Set-Cookie" in headers + + def test_empty_header_lines(self): + """Test parsing empty header lines""" + headers, cookies = _ParseHeaders([]) + assert len(headers) == 0 + assert len(cookies) == 0 + + +class TestRequestNamedTuple: + """Test Request namedtuple functionality""" + + def test_request_creation(self): + """Test creating Request namedtuple""" + request = Request( + method="GET", + url="https://example.com", + params={"q": "test"}, + data=None, + json_data=None, + headers={"User-Agent": "Test"}, + cookies={"session": "abc123"}, + proxy=None, + follow_redirects=True + ) + + assert request.method == "GET" + assert request.url == "https://example.com" + assert request.params == {"q": "test"} + assert request.headers == {"User-Agent": "Test"} + assert request.follow_redirects is True + + def test_request_defaults(self): + """Test Request with default/None values""" + request = Request( + method="POST", + url="https://api.example.com", + params=None, + data='{"key": "value"}', + json_data={"key": "value"}, + headers={}, + cookies={}, + proxy="http://proxy:8080", + follow_redirects=False + ) + + assert request.method == "POST" + assert request.data == '{"key": "value"}' + assert request.json_data == {"key": "value"} + assert request.proxy == "http://proxy:8080" + assert request.follow_redirects is False + + def test_request_field_access(self): + """Test accessing Request fields""" + request = Request( + "GET", "https://example.com", {}, None, None, {}, {}, None, True + ) + + # Test field access by name + assert hasattr(request, 'method') + assert hasattr(request, 'url') + assert hasattr(request, 'params') + assert hasattr(request, 'data') + assert hasattr(request, 'json_data') + assert hasattr(request, 'headers') + assert hasattr(request, 'cookies') + assert hasattr(request, 'proxy') + assert hasattr(request, 'follow_redirects') + + # Test field access by index + assert request[0] == "GET" + assert request[1] == "https://example.com" + + +class TestLoggingLevels: + """Test logging level constants""" + + def test_known_logging_levels(self): + """Test that all known logging levels are defined""" + expected_levels = ["debug", "info", "warning", "error", "critical", "fatal"] + + for level in expected_levels: + assert level in _known_logging_levels + assert isinstance(_known_logging_levels[level], int) + + def test_logging_level_values(self): + """Test logging level values are correct""" + from logging import DEBUG, INFO, WARNING, ERROR, CRITICAL, FATAL + + assert _known_logging_levels["debug"] == DEBUG + assert _known_logging_levels["info"] == INFO + assert _known_logging_levels["warning"] == WARNING + assert _known_logging_levels["error"] == ERROR + assert _known_logging_levels["critical"] == CRITICAL + assert _known_logging_levels["fatal"] == FATAL + + def test_level_hierarchy(self): + """Test that logging levels have correct hierarchy""" + levels = [ + _known_logging_levels["debug"], + _known_logging_levels["info"], + _known_logging_levels["warning"], + _known_logging_levels["error"], + _known_logging_levels["critical"], + ] + + # Levels should be in ascending order + for i in range(len(levels) - 1): + assert levels[i] < levels[i + 1] diff --git a/tests/core/test_storage_core.py b/tests/core/test_storage_core.py new file mode 100644 index 0000000..241ac0b --- /dev/null +++ b/tests/core/test_storage_core.py @@ -0,0 +1,37 @@ +import tempfile +import os + +from scrapling.core.storage import SQLiteStorageSystem + + +class TestSQLiteStorageSystem: + """Test SQLiteStorageSystem functionality""" + + def test_sqlite_storage_creation(self): + """Test SQLite storage system creation""" + # Use an in-memory database for testing + storage = SQLiteStorageSystem(storage_file=":memory:") + assert storage is not None + + def test_sqlite_storage_with_file(self): + """Test SQLite storage with an actual file""" + with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp_file: + db_path = tmp_file.name + + try: + storage = SQLiteStorageSystem(storage_file=db_path) + assert storage is not None + assert os.path.exists(db_path) + finally: + if os.path.exists(db_path): + os.unlink(db_path) + + def test_sqlite_storage_initialization_args(self): + """Test SQLite storage with various initialization arguments""" + # Test with URL parameter + storage = SQLiteStorageSystem( + storage_file=":memory:", + url="https://example.com" + ) + assert storage is not None + assert storage.url == "https://example.com" diff --git a/tests/fetchers/async/test_camoufox.py b/tests/fetchers/async/test_camoufox.py index 3e5ba5c..35c26cc 100644 --- a/tests/fetchers/async/test_camoufox.py +++ b/tests/fetchers/async/test_camoufox.py @@ -24,8 +24,13 @@ class TestStealthyFetcher: "html_url": f"{url}/html", "delayed_url": f"{url}/delay/10", # 10 Seconds delay response "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): """Test doing a basic fetch request with multiple statuses""" assert (await fetcher.async_fetch(urls["status_200"])).status == 200 @@ -63,7 +68,7 @@ class TestStealthyFetcher: { "network_idle": True, "wait": 10, - "cookies": [], + "cookies": [{"name": "test", "value": "123", "domain": "example.com", "path": "/"}], "google_search": True, "extra_headers": {"ayo": ""}, "os_randomize": True, diff --git a/tests/fetchers/async/test_dynamic.py b/tests/fetchers/async/test_dynamic.py index 70716e3..0d171ce 100644 --- a/tests/fetchers/async/test_dynamic.py +++ b/tests/fetchers/async/test_dynamic.py @@ -65,7 +65,7 @@ class TestDynamicFetcherAsync: "locale": "en-US", "extra_headers": {"ayo": ""}, "useragent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:131.0) Gecko/20100101 Firefox/131.0", - "cookies": [], + "cookies": [{"name": "test", "value": "123", "domain": "example.com", "path": "/"}], "network_idle": True, "custom_config": {"keep_comments": False, "keep_cdata": False}, }, diff --git a/tests/fetchers/sync/test_camoufox.py b/tests/fetchers/sync/test_camoufox.py index 9594a9c..5590cff 100644 --- a/tests/fetchers/sync/test_camoufox.py +++ b/tests/fetchers/sync/test_camoufox.py @@ -2,7 +2,6 @@ import pytest import pytest_httpbin from scrapling import StealthyFetcher - StealthyFetcher.adaptive = True @@ -23,6 +22,11 @@ class TestStealthyFetcher: self.html_url = f"{httpbin.url}/html" self.delayed_url = f"{httpbin.url}/delay/10" # 10 Seconds delay response self.cookies_url = f"{httpbin.url}/cookies/set/test/value" + self.cloudflare_url = "https://nopecha.com/demo/cloudflare" # Interactive turnstile page + + def test_cloudflare_fetch(self, fetcher): + """Test if Cloudflare bypass is working""" + assert fetcher.fetch(self.cloudflare_url, solve_cloudflare=True).status == 200 def test_basic_fetch(self, fetcher): """Test doing a basic fetch request with multiple statuses""" @@ -60,7 +64,7 @@ class TestStealthyFetcher: "network_idle": True, "wait": 10, "timeout": 30_000, - "cookies": [], + "cookies": [{"name": "test", "value": "123", "domain": "example.com", "path": "/"}], "google_search": True, "extra_headers": {"ayo": ""}, "os_randomize": True, diff --git a/tests/fetchers/sync/test_camoufox_session.py b/tests/fetchers/sync/test_camoufox_session.py new file mode 100644 index 0000000..c062708 --- /dev/null +++ b/tests/fetchers/sync/test_camoufox_session.py @@ -0,0 +1,97 @@ +import re +import pytest +import pytest_httpbin + +from scrapling.engines._browsers._camoufox import StealthySession, __CF_PATTERN__ + + +class TestCamoufoxConstants: + """Test Camoufox constants and patterns""" + + def test_cf_pattern_regex(self): + """Test __CF_PATTERN__ regex compilation""" + + assert isinstance(__CF_PATTERN__, re.Pattern) + + # Test matching URLs + test_urls = [ + "https://challenges.cloudflare.com/cdn-cgi/challenge-platform/h/123456", + "https://challenges.cloudflare.com/cdn-cgi/challenge-platform/orchestrate/jsch/v1", + "http://challenges.cloudflare.com/cdn-cgi/challenge-platform/scripts/abc" + ] + + for url in test_urls: + assert __CF_PATTERN__.search(url) is not None + + # Test non-matching URLs + non_matching_urls = [ + "https://example.com/challenge", + "https://cloudflare.com/something", + "https://challenges.cloudflare.com/other-path" + ] + + for url in non_matching_urls: + assert __CF_PATTERN__.search(url) is None + + +@pytest_httpbin.use_class_based_httpbin +class TestStealthySession: + + """All the code is tested in the async version tests, so no need to repeat it here. The async class inherits from this one.""" + @pytest.fixture(autouse=True) + def setup_urls(self, httpbin): + """Fixture to set up URLs for testing""" + self.status_200 = f"{httpbin.url}/status/200" + self.status_404 = f"{httpbin.url}/status/404" + self.status_501 = f"{httpbin.url}/status/501" + self.basic_url = f"{httpbin.url}/get" + self.html_url = f"{httpbin.url}/html" + self.delayed_url = f"{httpbin.url}/delay/10" # 10 Seconds delay response + self.cookies_url = f"{httpbin.url}/cookies/set/test/value" + + def test_session_creation(self): + """Test if the session is created correctly""" + + with StealthySession( + max_pages=3, + headless=True, + block_images=True, + disable_resources=True, + solve_cloudflare=True, + wait=1000, + timeout=60000, + cookies=[{"name": "test", "value": "123", "domain": "example.com", "path": "/"}], + ) as session: + + assert session.max_pages == 3 + assert session.headless is True + assert session.block_images is True + assert session.disable_resources is True + assert session.solve_cloudflare is True + assert session.wait == 1000 + assert session.timeout == 60000 + assert session.context is not None + + # Test Cloudflare detection + for cloudflare_type in ('managed', 'interactive', 'non-interactive'): + page_content = f""" + + + + """ + result = session._detect_cloudflare(page_content) + assert result == cloudflare_type + + page_content = """ + + +

Regular page content

+ + + """ + + result = StealthySession._detect_cloudflare(page_content) + assert result is None + assert session.fetch(self.status_200).status == 200 diff --git a/tests/fetchers/sync/test_dynamic.py b/tests/fetchers/sync/test_dynamic.py index 2079647..4804acc 100644 --- a/tests/fetchers/sync/test_dynamic.py +++ b/tests/fetchers/sync/test_dynamic.py @@ -63,7 +63,7 @@ class TestDynamicFetcher: "locale": "en-US", "extra_headers": {"ayo": ""}, "useragent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:131.0) Gecko/20100101 Firefox/131.0", - "cookies": [], + "cookies": [{"name": "test", "value": "123", "domain": "example.com", "path": "/"}], "network_idle": True, "custom_config": {"keep_comments": False, "keep_cdata": False}, }, diff --git a/tests/parser/test_general.py b/tests/parser/test_general.py index 6c17a61..e5f8d59 100644 --- a/tests/parser/test_general.py +++ b/tests/parser/test_general.py @@ -221,7 +221,7 @@ class TestElementNavigation: """Test parent and sibling navigation""" table = page.css(".product-list")[0] parent = table.parent - assert parent.attrib["id"] == "products" + assert parent["id"] == "products" parent_siblings = parent.siblings assert len(parent_siblings) == 1 @@ -267,7 +267,7 @@ class TestJSONAndAttributes: products = page.css(".product") product_ids = [product.attrib["data-id"] for product in products] assert product_ids == ["1", "2", "3"] - assert "data-id" in products[0].attrib + assert "data-id" in products[0] # Review rating calculations reviews = page.css(".review") @@ -316,7 +316,9 @@ def test_selectors_generation(page): def _traverse(element: Selector): assert isinstance(element.generate_css_selector, str) + assert isinstance(element.generate_full_css_selector, str) assert isinstance(element.generate_xpath_selector, str) + assert isinstance(element.generate_full_xpath_selector, str) for branch in element.children: _traverse(branch) diff --git a/tests/parser/test_parser_advanced.py b/tests/parser/test_parser_advanced.py index aa9ea7a..71552f9 100644 --- a/tests/parser/test_parser_advanced.py +++ b/tests/parser/test_parser_advanced.py @@ -1,8 +1,58 @@ import re import pytest +from unittest.mock import Mock from scrapling import Selector, Selectors from scrapling.core.custom_types import TextHandler, TextHandlers +from scrapling.core.storage import SQLiteStorageSystem + + +class TestSelectorAdvancedFeatures: + """Test advanced Selector features like adaptive matching""" + + def test_adaptive_initialization_with_storage(self): + """Test adaptive initialization with custom storage""" + html = "

Test

" + + # Use the actual SQLiteStorageSystem for this test + selector = Selector( + content=html, + adaptive=True, + storage=SQLiteStorageSystem, + storage_args={"storage_file": ":memory:", "url": "https://example.com"} + ) + + assert selector._Selector__adaptive_enabled is True + assert selector._storage is not None + + def test_adaptive_initialization_with_default_storage_args(self): + """Test adaptive initialization with default storage args""" + html = "

Test

" + url = "https://example.com" + + # Test that adaptive mode uses default storage when no explicit args provided + selector = Selector( + content=html, + url=url, + adaptive=True + ) + + # Should create storage with default args + assert selector._storage is not None + + def test_adaptive_with_existing_storage(self): + """Test adaptive initialization with existing storage object""" + html = "

Test

" + + mock_storage = Mock() + + selector = Selector( + content=html, + adaptive=True, + _storage=mock_storage + ) + + assert selector._storage is mock_storage class TestAdvancedSelectors: