Spaces:
Sleeping
Sleeping
File size: 3,143 Bytes
ebab0e2 | 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 | from unittest.mock import AsyncMock, MagicMock
import pytest
from playwright.async_api import TimeoutError as PlaywrightTimeoutError
from app.errors import ScrapeError
from app.scraper import PlaywrightScraper
def make_mock_browser(page: MagicMock) -> MagicMock:
context = MagicMock()
context.new_page = AsyncMock(return_value=page)
context.close = AsyncMock()
browser = MagicMock()
browser.new_context = AsyncMock(return_value=context)
return browser, context
def make_mock_page(content: str = "<html>ok</html>") -> MagicMock:
page = MagicMock()
page.goto = AsyncMock()
page.content = AsyncMock(return_value=content)
return page
async def test_scrape_returns_page_content_on_success():
page = make_mock_page("<html>hello</html>")
browser, context = make_mock_browser(page)
stealth = AsyncMock()
scraper = PlaywrightScraper(
browser_provider=lambda: browser, user_agent="UA", timeout_ms=5000, stealth=stealth
)
result = await scraper.scrape("https://example.com/")
assert result == "<html>hello</html>"
stealth.assert_awaited_once_with(page)
context.close.assert_awaited_once()
async def test_scrape_raises_503_when_browser_not_ready():
scraper = PlaywrightScraper(browser_provider=lambda: None, user_agent="UA", timeout_ms=5000)
with pytest.raises(ScrapeError) as exc_info:
await scraper.scrape("https://example.com/")
assert exc_info.value.status_code == 503
async def test_scrape_maps_playwright_timeout_to_504_and_still_closes_context():
page = make_mock_page()
page.goto = AsyncMock(side_effect=PlaywrightTimeoutError("timed out"))
browser, context = make_mock_browser(page)
scraper = PlaywrightScraper(
browser_provider=lambda: browser, user_agent="UA", timeout_ms=5000, stealth=AsyncMock()
)
with pytest.raises(ScrapeError) as exc_info:
await scraper.scrape("https://example.com/")
assert exc_info.value.status_code == 504
context.close.assert_awaited_once()
async def test_context_is_closed_even_when_page_setup_fails():
page = make_mock_page()
browser, context = make_mock_browser(page)
failing_stealth = AsyncMock(side_effect=RuntimeError("boom"))
scraper = PlaywrightScraper(
browser_provider=lambda: browser, user_agent="UA", timeout_ms=5000, stealth=failing_stealth
)
with pytest.raises(RuntimeError):
await scraper.scrape("https://example.com/")
context.close.assert_awaited_once()
async def test_scrape_configures_context_and_navigation():
page = make_mock_page()
browser, context = make_mock_browser(page)
scraper = PlaywrightScraper(
browser_provider=lambda: browser, user_agent="Custom UA", timeout_ms=12345, stealth=AsyncMock()
)
await scraper.scrape("https://example.com/")
_args, kwargs = browser.new_context.call_args
assert kwargs["user_agent"] == "Custom UA"
assert kwargs["viewport"] == {"width": 1920, "height": 1080}
_args, kwargs = page.goto.call_args
assert kwargs["wait_until"] == "domcontentloaded"
assert kwargs["timeout"] == 12345
|