Files
superset2/tests/unit_tests/utils/webdriver_test.py
Elizabeth Thompson 199435e5c5 fix(reports): poll for spinner absence instead of snapshotting loading elements
The Playwright screenshot path waited for `.loading` elements to detach
by calling `page.locator(".loading").all()` once and iterating the
snapshot. Spinners that appeared after the snapshot was taken (common on
long dashboards where charts load lazily) were never waited for, causing
PDF reports to capture charts that were still fetching data.

Replace the snapshot loop with `page.wait_for_function` which
continuously polls `document.querySelectorAll('.loading').length === 0`
until the page is genuinely clear of all spinners, then use
`self._screenshot_load_wait` (already stored on the base class) for the
timeout to stay consistent with the Selenium path.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-23 02:01:59 +00:00

813 lines
33 KiB
Python

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
from unittest.mock import MagicMock, patch, PropertyMock
import pytest
from superset.utils.webdriver import (
check_playwright_availability,
PLAYWRIGHT_AVAILABLE,
PLAYWRIGHT_INSTALL_MESSAGE,
validate_webdriver_config,
WebDriverPlaywright,
WebDriverSelenium,
)
@pytest.fixture
def mock_app():
"""Mock Flask app with webdriver configuration."""
app = MagicMock()
app.config = {
"WEBDRIVER_TYPE": "chrome",
"WEBDRIVER_OPTION_ARGS": [],
"WEBDRIVER_CONFIGURATION": {},
"SCREENSHOT_LOCATE_WAIT": 10,
"SCREENSHOT_LOAD_WAIT": 10,
}
return app
class TestWebDriverSelenium:
"""Test WebDriverSelenium timeout handling for urllib3 2.x compatibility."""
@patch("superset.utils.webdriver.app")
@patch("superset.utils.webdriver.firefox")
@patch("superset.utils.webdriver.chrome")
def test_timeout_conversion_to_float(
self, mock_chrome, mock_firefox, mock_app_patch, mock_app
):
"""Test that timeout values are properly converted to float."""
# Set up app mock to be used throughout
mock_app_patch.config = {
"WEBDRIVER_TYPE": "chrome",
"WEBDRIVER_OPTION_ARGS": [],
"SCREENSHOT_LOCATE_WAIT": 10,
"SCREENSHOT_LOAD_WAIT": 10,
"WEBDRIVER_WINDOW": {},
"WEBDRIVER_CONFIGURATION": {
"timeout": "30",
"connect_timeout": "10.5",
"socket_timeout": 20,
"read_timeout": "15.0",
"command_executor_timeout": "25",
},
}
mock_driver_class = MagicMock()
mock_chrome.webdriver.WebDriver = mock_driver_class
mock_chrome.service.Service = MagicMock()
mock_options = MagicMock()
mock_options.add_argument = MagicMock()
mock_chrome.options.Options = MagicMock(return_value=mock_options)
driver = WebDriverSelenium(driver_type="chrome")
driver.create()
# Check that the driver was called with float timeout values
mock_driver_class.assert_called_once()
call_kwargs = mock_driver_class.call_args.kwargs
assert call_kwargs["timeout"] == 30.0
assert call_kwargs["connect_timeout"] == 10.5
assert call_kwargs["socket_timeout"] == 20.0
assert call_kwargs["read_timeout"] == 15.0
assert call_kwargs["command_executor_timeout"] == 25.0
@patch("superset.utils.webdriver.app")
@patch("superset.utils.webdriver.chrome")
def test_timeout_none_handling(self, mock_chrome, mock_app_patch, mock_app):
"""Test that None, 'None', and 'null' timeout values are set to None."""
mock_app_patch.config = {
"WEBDRIVER_TYPE": "chrome",
"WEBDRIVER_OPTION_ARGS": [],
"SCREENSHOT_LOCATE_WAIT": 10,
"SCREENSHOT_LOAD_WAIT": 10,
"WEBDRIVER_WINDOW": {},
"WEBDRIVER_CONFIGURATION": {
"timeout": None,
"connect_timeout": "None",
"socket_timeout": "null",
},
}
mock_driver_class = MagicMock()
mock_chrome.webdriver.WebDriver = mock_driver_class
mock_chrome.service.Service = MagicMock()
mock_options = MagicMock()
mock_options.add_argument = MagicMock()
mock_chrome.options.Options = MagicMock(return_value=mock_options)
driver = WebDriverSelenium(driver_type="chrome")
driver.create()
# Check that None values are preserved
mock_driver_class.assert_called_once()
call_kwargs = mock_driver_class.call_args.kwargs
assert call_kwargs["timeout"] is None
assert call_kwargs["connect_timeout"] is None
assert call_kwargs["socket_timeout"] is None
@patch("superset.utils.webdriver.app")
@patch("superset.utils.webdriver.chrome")
@patch("superset.utils.webdriver.logger")
def test_invalid_timeout_warning(
self, mock_logger, mock_chrome, mock_app_patch, mock_app
):
"""Test that invalid timeout values log warnings and are set to None."""
mock_app_patch.config = {
"WEBDRIVER_TYPE": "chrome",
"WEBDRIVER_OPTION_ARGS": [],
"SCREENSHOT_LOCATE_WAIT": 10,
"SCREENSHOT_LOAD_WAIT": 10,
"WEBDRIVER_WINDOW": {},
"WEBDRIVER_CONFIGURATION": {
"timeout": "invalid",
"connect_timeout": "not_a_number",
"Page_Load_Timeout": "abc123", # Test case-insensitive matching
},
}
mock_driver_class = MagicMock()
mock_chrome.webdriver.WebDriver = mock_driver_class
mock_chrome.service.Service = MagicMock()
mock_options = MagicMock()
mock_options.add_argument = MagicMock()
mock_chrome.options.Options = MagicMock(return_value=mock_options)
driver = WebDriverSelenium(driver_type="chrome")
driver.create()
# Check that invalid values are set to None
mock_driver_class.assert_called_once()
call_kwargs = mock_driver_class.call_args.kwargs
assert call_kwargs["timeout"] is None
assert call_kwargs["connect_timeout"] is None
assert call_kwargs["Page_Load_Timeout"] is None
# Check that warnings were logged with lazy logging format
assert mock_logger.warning.call_count == 3
mock_logger.warning.assert_any_call(
"Invalid timeout value for %s: %s, setting to None", "timeout", "invalid"
)
mock_logger.warning.assert_any_call(
"Invalid timeout value for %s: %s, setting to None",
"connect_timeout",
"not_a_number",
)
mock_logger.warning.assert_any_call(
"Invalid timeout value for %s: %s, setting to None",
"Page_Load_Timeout",
"abc123",
)
@patch("superset.utils.webdriver.app")
@patch("superset.utils.webdriver.chrome")
def test_non_timeout_config_preserved(self, mock_chrome, mock_app_patch, mock_app):
"""Test that non-timeout configuration values are preserved."""
mock_app_patch.config = {
"WEBDRIVER_TYPE": "chrome",
"WEBDRIVER_OPTION_ARGS": [],
"SCREENSHOT_LOCATE_WAIT": 10,
"SCREENSHOT_LOAD_WAIT": 10,
"WEBDRIVER_WINDOW": {},
"WEBDRIVER_CONFIGURATION": {
"timeout": "30",
"some_other_option": "value",
"another_option": 123,
"boolean_option": True,
},
}
mock_driver_class = MagicMock()
mock_chrome.webdriver.WebDriver = mock_driver_class
mock_chrome.service.Service = MagicMock()
mock_options = MagicMock()
mock_options.add_argument = MagicMock()
mock_chrome.options.Options = MagicMock(return_value=mock_options)
driver = WebDriverSelenium(driver_type="chrome")
driver.create()
# Check that all config values are passed through
mock_driver_class.assert_called_once()
call_kwargs = mock_driver_class.call_args.kwargs
assert call_kwargs["timeout"] == 30.0
assert call_kwargs["some_other_option"] == "value"
assert call_kwargs["another_option"] == 123
assert call_kwargs["boolean_option"] is True
@patch("superset.utils.webdriver.app")
@patch("superset.utils.webdriver.chrome")
def test_timeout_key_case_insensitive(self, mock_chrome, mock_app_patch, mock_app):
"""Test that timeout detection is case-insensitive."""
mock_app_patch.config = {
"WEBDRIVER_TYPE": "chrome",
"WEBDRIVER_OPTION_ARGS": [],
"SCREENSHOT_LOCATE_WAIT": 10,
"SCREENSHOT_LOAD_WAIT": 10,
"WEBDRIVER_WINDOW": {},
"WEBDRIVER_CONFIGURATION": {
"TIMEOUT": "10",
"Connect_Timeout": "20",
"SOCKET_TIMEOUT": "30",
"connection_timeout_ms": "5000", # Contains 'connection_timeout'
},
}
mock_driver_class = MagicMock()
mock_chrome.webdriver.WebDriver = mock_driver_class
mock_chrome.service.Service = MagicMock()
mock_options = MagicMock()
mock_options.add_argument = MagicMock()
mock_chrome.options.Options = MagicMock(return_value=mock_options)
driver = WebDriverSelenium(driver_type="chrome")
driver.create()
# Check that all timeout values are converted to float
mock_driver_class.assert_called_once()
call_kwargs = mock_driver_class.call_args.kwargs
assert call_kwargs["TIMEOUT"] == 10.0
assert call_kwargs["Connect_Timeout"] == 20.0
assert call_kwargs["SOCKET_TIMEOUT"] == 30.0
assert call_kwargs["connection_timeout_ms"] == 5000.0
@patch("superset.utils.webdriver.app")
@patch("superset.utils.webdriver.chrome")
def test_empty_webdriver_config(self, mock_chrome, mock_app_patch, mock_app):
"""Test handling of empty webdriver configuration."""
mock_app_patch.config = {
"WEBDRIVER_TYPE": "chrome",
"WEBDRIVER_OPTION_ARGS": [],
"SCREENSHOT_LOCATE_WAIT": 10,
"SCREENSHOT_LOAD_WAIT": 10,
"WEBDRIVER_WINDOW": {},
"WEBDRIVER_CONFIGURATION": {},
}
mock_driver_class = MagicMock()
mock_chrome.webdriver.WebDriver = mock_driver_class
mock_chrome.service.Service = MagicMock()
mock_options = MagicMock()
mock_options.add_argument = MagicMock()
mock_chrome.options.Options = MagicMock(return_value=mock_options)
driver = WebDriverSelenium(driver_type="chrome")
driver.create()
# Should create driver without errors
mock_driver_class.assert_called_once()
class TestPlaywrightAvailabilityCheck:
"""Test comprehensive Playwright availability checking."""
@patch("superset.utils.webdriver.sync_playwright", None)
def test_check_playwright_availability_returns_false_when_module_not_available(
self,
):
"""Test check_playwright_availability returns False when no module."""
result = check_playwright_availability()
assert result is False
@patch("superset.utils.webdriver.sync_playwright")
@patch("superset.utils.webdriver.logger")
def test_check_playwright_availability_uses_lightweight_check(
self, mock_logger, mock_sync_playwright
):
"""Test check_playwright_availability uses executable_path first."""
# Setup mocks for successful executable path check
mock_playwright_instance = MagicMock()
mock_sync_playwright.return_value.__enter__.return_value = (
mock_playwright_instance
)
mock_playwright_instance.chromium.executable_path = "/path/to/chromium"
result = check_playwright_availability()
assert result is True
# Should not launch browser if executable_path works
mock_playwright_instance.chromium.launch.assert_not_called()
@patch("superset.utils.webdriver.sync_playwright")
@patch("superset.utils.webdriver.logger")
def test_check_playwright_availability_falls_back_to_launch(
self, mock_logger, mock_sync_playwright
):
"""Test check_playwright_availability falls back to browser launch."""
# Setup mocks where executable_path fails but launch succeeds
mock_playwright_instance = MagicMock()
mock_browser = MagicMock()
mock_sync_playwright.return_value.__enter__.return_value = (
mock_playwright_instance
)
# Make executable_path raise exception
type(mock_playwright_instance.chromium).executable_path = PropertyMock(
side_effect=Exception("executable_path failed")
)
mock_playwright_instance.chromium.launch.return_value = mock_browser
result = check_playwright_availability()
assert result is True
# Should fall back to browser launch
mock_playwright_instance.chromium.launch.assert_called_once_with(headless=True)
mock_browser.close.assert_called_once()
@patch("superset.utils.webdriver.sync_playwright")
@patch("superset.utils.webdriver.logger")
def test_check_playwright_availability_handles_browser_launch_failure(
self, mock_logger, mock_sync_playwright
):
"""Test check_playwright_availability handles browser launch failures."""
# Setup mocks to raise exception on browser launch
mock_playwright_instance = MagicMock()
mock_sync_playwright.return_value.__enter__.return_value = (
mock_playwright_instance
)
# Mock executable_path to raise exception to force fallback to launch test
type(mock_playwright_instance.chromium).executable_path = PropertyMock(
side_effect=Exception("Executable path check failed")
)
mock_playwright_instance.chromium.launch.side_effect = Exception(
"Browser binaries not installed"
)
result = check_playwright_availability()
assert result is False
mock_logger.warning.assert_called_once()
warning_call = mock_logger.warning.call_args[0][0]
assert (
"Playwright module is installed but browser launch failed" in warning_call
)
assert "playwright install chromium" in warning_call
@patch("superset.utils.webdriver.sync_playwright")
@patch("superset.utils.webdriver.logger")
def test_check_playwright_availability_handles_context_manager_error(
self, mock_logger, mock_sync_playwright
):
"""Test check_playwright_availability handles context manager errors."""
# Setup mock to raise exception when entering context
mock_sync_playwright.return_value.__enter__.side_effect = Exception(
"Context error"
)
result = check_playwright_availability()
assert result is False
mock_logger.warning.assert_called_once()
class TestPlaywrightMigrationSupport:
"""Test Playwright migration and fallback functionality."""
@patch("superset.extensions.feature_flag_manager.is_feature_enabled")
def test_validate_webdriver_config_all_available(self, mock_feature_flag):
"""Test validate_webdriver_config when all dependencies available."""
mock_feature_flag.return_value = True
result = validate_webdriver_config()
assert result["selenium_available"] is True
assert isinstance(result["playwright_available"], bool)
assert isinstance(result["playwright_feature_enabled"], bool)
if result["playwright_available"]:
assert result["recommended_action"] is None
else:
assert result["recommended_action"] == PLAYWRIGHT_INSTALL_MESSAGE
@patch("superset.extensions.feature_flag_manager.is_feature_enabled")
def test_validate_webdriver_config_feature_flag_disabled(self, mock_feature_flag):
"""Test validate_webdriver_config when feature flag is disabled."""
mock_feature_flag.return_value = False
result = validate_webdriver_config()
assert result["selenium_available"] is True
assert result["playwright_feature_enabled"] is False
@patch("superset.extensions.feature_flag_manager.is_feature_enabled")
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", False)
def test_validate_webdriver_config_playwright_unavailable(self, mock_feature_flag):
"""Test validate_webdriver_config when Playwright not available."""
mock_feature_flag.return_value = True
result = validate_webdriver_config()
assert result["selenium_available"] is True
assert result["playwright_available"] is False
assert result["playwright_feature_enabled"] is True
assert result["recommended_action"] == PLAYWRIGHT_INSTALL_MESSAGE
class TestWebDriverPlaywrightFallback:
"""Test WebDriverPlaywright fallback behavior when unavailable."""
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", False)
@patch("superset.utils.webdriver.logger")
def test_get_screenshot_returns_none_when_unavailable(self, mock_logger, mock_app):
"""Test WebDriverPlaywright.get_screenshot returns None when unavailable."""
mock_user = MagicMock()
mock_user.username = "test_user"
driver = WebDriverPlaywright("chrome")
result = driver.get_screenshot("http://example.com", "test-element", mock_user)
assert result is None
# Verify warning log was called with correct message
mock_logger.info.assert_called_once()
log_call = mock_logger.info.call_args[0][0]
assert "Playwright not available" in log_call
assert "falling back to Selenium" in log_call
assert "WebGL/Canvas charts may not render correctly" in log_call
# Check the substituted parameter
assert mock_logger.info.call_args[0][1] == PLAYWRIGHT_INSTALL_MESSAGE
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver.sync_playwright")
@patch("superset.utils.webdriver.app")
def test_get_screenshot_works_when_available(self, mock_app, mock_sync_playwright):
"""Test WebDriverPlaywright.get_screenshot works when Playwright available."""
# Setup mocks
mock_user = MagicMock()
mock_user.username = "test_user"
mock_app.config = {
"WEBDRIVER_OPTION_ARGS": [],
"WEBDRIVER_WINDOW": {"pixel_density": 1},
"SCREENSHOT_PLAYWRIGHT_DEFAULT_TIMEOUT": 30000,
"SCREENSHOT_PLAYWRIGHT_WAIT_EVENT": "networkidle",
"SCREENSHOT_SELENIUM_HEADSTART": 5,
"SCREENSHOT_SELENIUM_ANIMATION_WAIT": 1,
"SCREENSHOT_REPLACE_UNEXPECTED_ERRORS": False,
"SCREENSHOT_TILED_ENABLED": False,
"SCREENSHOT_LOCATE_WAIT": 10,
"SCREENSHOT_LOAD_WAIT": 10,
"SCREENSHOT_WAIT_FOR_ERROR_MODAL_VISIBLE": 10,
"SCREENSHOT_WAIT_FOR_ERROR_MODAL_INVISIBLE": 10,
}
# Setup playwright mocks
mock_playwright_instance = MagicMock()
mock_browser = MagicMock()
mock_context = MagicMock()
mock_page = MagicMock()
mock_element = MagicMock()
mock_sync_playwright.return_value.__enter__.return_value = (
mock_playwright_instance
)
mock_playwright_instance.chromium.launch.return_value = mock_browser
mock_browser.new_context.return_value = mock_context
mock_context.new_page.return_value = mock_page
mock_page.locator.return_value = mock_element
mock_element.screenshot.return_value = b"fake_screenshot"
# Mock the auth method
with patch.object(WebDriverPlaywright, "auth") as mock_auth:
mock_auth.return_value = mock_context
driver = WebDriverPlaywright("chrome")
result = driver.get_screenshot(
"http://example.com", "test-element", mock_user
)
assert result == b"fake_screenshot"
mock_page.goto.assert_called_once_with(
"http://example.com", wait_until="networkidle"
)
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver.sync_playwright")
@patch("superset.utils.webdriver.logger")
def test_get_screenshot_handles_playwright_timeout(
self, mock_logger, mock_sync_playwright
):
"""Test WebDriverPlaywright handles PlaywrightTimeout gracefully."""
from superset.utils.webdriver import PlaywrightTimeout
mock_user = MagicMock()
mock_user.username = "test_user"
# Setup playwright mocks to raise timeout
mock_playwright_instance = MagicMock()
mock_browser = MagicMock()
mock_context = MagicMock()
mock_page = MagicMock()
mock_sync_playwright.return_value.__enter__.return_value = (
mock_playwright_instance
)
mock_playwright_instance.chromium.launch.return_value = mock_browser
mock_browser.new_context.return_value = mock_context
mock_context.new_page.return_value = mock_page
mock_page.goto.side_effect = PlaywrightTimeout()
with patch("superset.utils.webdriver.app") as mock_app:
mock_app.config = {
"WEBDRIVER_OPTION_ARGS": [],
"WEBDRIVER_WINDOW": {"pixel_density": 1},
"SCREENSHOT_PLAYWRIGHT_DEFAULT_TIMEOUT": 30000,
"SCREENSHOT_PLAYWRIGHT_WAIT_EVENT": "networkidle",
"SCREENSHOT_SELENIUM_HEADSTART": 5,
"SCREENSHOT_SELENIUM_ANIMATION_WAIT": 1,
"SCREENSHOT_LOCATE_WAIT": 10,
"SCREENSHOT_LOAD_WAIT": 10,
"SCREENSHOT_WAIT_FOR_ERROR_MODAL_VISIBLE": 10,
"SCREENSHOT_WAIT_FOR_ERROR_MODAL_INVISIBLE": 10,
"SCREENSHOT_REPLACE_UNEXPECTED_ERRORS": True,
"SCREENSHOT_TILED_ENABLED": False,
}
with patch.object(WebDriverPlaywright, "auth") as mock_auth:
mock_auth.return_value = mock_context
driver = WebDriverPlaywright("chrome")
result = driver.get_screenshot(
"http://example.com", "test-element", mock_user
)
# page.goto() timeout is caught and logged without aborting; execution
# continues to the element waits, which succeed here, so a screenshot
# is taken and returned (not None).
assert result is not None
mock_logger.exception.assert_called()
exception_call = mock_logger.exception.call_args[0][0]
assert "Web event %s not detected" in exception_call
class TestWebDriverConstantsWithImportError:
"""Test module-level constants behavior with import errors."""
def test_playwright_constants_defined_when_import_fails(self):
"""Test constants are properly defined even when Playwright import fails."""
# These should be available even when playwright is not installed
assert PLAYWRIGHT_INSTALL_MESSAGE is not None
assert isinstance(PLAYWRIGHT_INSTALL_MESSAGE, str)
# PLAYWRIGHT_AVAILABLE should be boolean regardless of installation
assert isinstance(PLAYWRIGHT_AVAILABLE, bool)
@patch("superset.utils.webdriver.sync_playwright", None)
def test_dummy_classes_when_playwright_unavailable(self):
"""Test that dummy classes are defined when Playwright unavailable."""
# Force reimport to test ImportError path
from importlib import reload
import superset.utils.webdriver as webdriver_module
# Mock the import to fail
with patch.dict("sys.modules", {"playwright.sync_api": None}):
reload(webdriver_module)
# Should have dummy classes defined
assert hasattr(webdriver_module, "BrowserContext")
assert hasattr(webdriver_module, "PlaywrightError")
assert hasattr(webdriver_module, "PlaywrightTimeout")
class TestWebDriverPlaywrightErrorHandling:
"""Test error handling in WebDriverPlaywright methods."""
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver.sync_playwright")
@patch("superset.utils.webdriver.logger")
def test_find_unexpected_errors_handles_playwright_error(
self, mock_logger, mock_sync_playwright
):
"""Test find_unexpected_errors handles PlaywrightError gracefully."""
from superset.utils.webdriver import PlaywrightError
mock_page = MagicMock()
mock_page.get_by_role.side_effect = PlaywrightError("Test error")
result = WebDriverPlaywright.find_unexpected_errors(mock_page)
assert result == []
mock_logger.exception.assert_called_once_with(
"Failed to capture unexpected errors"
)
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver.sync_playwright")
@patch("superset.utils.webdriver.logger")
def test_find_unexpected_errors_processes_alerts(
self, mock_logger, mock_sync_playwright
):
"""Test find_unexpected_errors processes alert elements correctly."""
mock_page = MagicMock()
mock_alert_div = MagicMock()
mock_button = MagicMock()
mock_modal_content = MagicMock()
mock_modal_body = MagicMock()
mock_close_button = MagicMock()
# Setup the mock chain
mock_page.get_by_role.return_value.all.return_value = [mock_alert_div]
mock_alert_div.get_by_role.return_value = mock_button
mock_page.locator.side_effect = [
mock_modal_content,
mock_modal_body,
mock_close_button,
mock_modal_content,
]
mock_modal_body.text_content.return_value = "Error message"
mock_modal_body.inner_html.return_value = "Error message"
result = WebDriverPlaywright.find_unexpected_errors(mock_page)
assert result == ["Error message"]
mock_button.click.assert_called_once()
mock_close_button.click.assert_called_once()
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver.sync_playwright")
@patch("superset.utils.webdriver.app")
def test_uses_wait_for_function_to_detect_spinners(
self, mock_app, mock_sync_playwright
):
"""wait_for_function polls for spinner absence rather than snapshotting."""
mock_user = MagicMock()
mock_user.username = "test_user"
mock_app.config = {
"WEBDRIVER_OPTION_ARGS": [],
"WEBDRIVER_WINDOW": {"pixel_density": 1},
"SCREENSHOT_PLAYWRIGHT_DEFAULT_TIMEOUT": 30000,
"SCREENSHOT_PLAYWRIGHT_WAIT_EVENT": "networkidle",
"SCREENSHOT_SELENIUM_HEADSTART": 0,
"SCREENSHOT_SELENIUM_ANIMATION_WAIT": 0,
"SCREENSHOT_REPLACE_UNEXPECTED_ERRORS": False,
"SCREENSHOT_TILED_ENABLED": False,
"SCREENSHOT_LOCATE_WAIT": 10,
"SCREENSHOT_LOAD_WAIT": 60,
"SCREENSHOT_WAIT_FOR_ERROR_MODAL_VISIBLE": 10,
"SCREENSHOT_WAIT_FOR_ERROR_MODAL_INVISIBLE": 10,
}
mock_playwright_instance = MagicMock()
mock_browser = MagicMock()
mock_context = MagicMock()
mock_page = MagicMock()
mock_element = MagicMock()
mock_sync_playwright.return_value.__enter__.return_value = (
mock_playwright_instance
)
mock_playwright_instance.chromium.launch.return_value = mock_browser
mock_browser.new_context.return_value = mock_context
mock_context.new_page.return_value = mock_page
mock_page.locator.return_value = mock_element
mock_element.screenshot.return_value = b"screenshot"
with patch.object(WebDriverPlaywright, "auth", return_value=mock_context):
driver = WebDriverPlaywright("chrome")
driver.get_screenshot("http://example.com", "test-element", mock_user)
mock_page.wait_for_function.assert_called_once_with(
"() => document.querySelectorAll('.loading').length === 0",
timeout=60 * 1000,
)
# Guard against reintroducing the old snapshot-based approach
loading_locator_calls = [
c for c in mock_page.locator.call_args_list if c.args == (".loading",)
]
assert loading_locator_calls == []
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver.sync_playwright")
@patch("superset.utils.webdriver.logger")
@patch("superset.utils.webdriver.app")
def test_spinner_timeout_logs_warning_and_raises(
self, mock_app, mock_logger, mock_sync_playwright
):
"""Spinner timeout is logged as a warning and re-raised."""
from superset.utils.webdriver import PlaywrightTimeout
mock_user = MagicMock()
mock_user.username = "test_user"
mock_app.config = {
"WEBDRIVER_OPTION_ARGS": [],
"WEBDRIVER_WINDOW": {"pixel_density": 1},
"SCREENSHOT_PLAYWRIGHT_DEFAULT_TIMEOUT": 30000,
"SCREENSHOT_PLAYWRIGHT_WAIT_EVENT": "networkidle",
"SCREENSHOT_SELENIUM_HEADSTART": 0,
"SCREENSHOT_SELENIUM_ANIMATION_WAIT": 0,
"SCREENSHOT_REPLACE_UNEXPECTED_ERRORS": False,
"SCREENSHOT_TILED_ENABLED": False,
"SCREENSHOT_LOCATE_WAIT": 10,
"SCREENSHOT_LOAD_WAIT": 60,
"SCREENSHOT_WAIT_FOR_ERROR_MODAL_VISIBLE": 10,
"SCREENSHOT_WAIT_FOR_ERROR_MODAL_INVISIBLE": 10,
}
mock_playwright_instance = MagicMock()
mock_browser = MagicMock()
mock_context = MagicMock()
mock_page = MagicMock()
mock_element = MagicMock()
mock_sync_playwright.return_value.__enter__.return_value = (
mock_playwright_instance
)
mock_playwright_instance.chromium.launch.return_value = mock_browser
mock_browser.new_context.return_value = mock_context
mock_context.new_page.return_value = mock_page
mock_page.locator.return_value = mock_element
timeout = PlaywrightTimeout()
mock_page.wait_for_function.side_effect = timeout
with patch.object(WebDriverPlaywright, "auth", return_value=mock_context):
driver = WebDriverPlaywright("chrome")
with pytest.raises(PlaywrightTimeout) as exc_info:
driver.get_screenshot("http://example.com", "test-element", mock_user)
assert exc_info.value is timeout
mock_logger.warning.assert_any_call(
"Timed out waiting for charts to load at url %s",
"http://example.com",
)
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver.sync_playwright")
@patch("superset.utils.webdriver.logger")
def test_get_screenshot_raises_on_element_wait_timeout(
self, mock_logger, mock_sync_playwright
):
"""Test that PlaywrightTimeout propagates when waiting for page elements."""
from superset.utils.webdriver import PlaywrightTimeout
mock_user = MagicMock()
mock_user.username = "test_user"
# Setup mocks
mock_playwright_instance = MagicMock()
mock_browser = MagicMock()
mock_context = MagicMock()
mock_page = MagicMock()
mock_element = MagicMock()
mock_sync_playwright.return_value.__enter__.return_value = (
mock_playwright_instance
)
mock_playwright_instance.chromium.launch.return_value = mock_browser
mock_browser.new_context.return_value = mock_context
mock_context.new_page.return_value = mock_page
# Keep a reference to the exact instance so we can verify identity below.
timeout = PlaywrightTimeout()
mock_page.locator.return_value = mock_element
mock_element.wait_for.side_effect = timeout
with patch("superset.utils.webdriver.app") as mock_app:
mock_app.config = {
"WEBDRIVER_OPTION_ARGS": [],
"WEBDRIVER_WINDOW": {"pixel_density": 1},
"SCREENSHOT_PLAYWRIGHT_DEFAULT_TIMEOUT": 30000,
"SCREENSHOT_PLAYWRIGHT_WAIT_EVENT": "networkidle",
"SCREENSHOT_SELENIUM_HEADSTART": 5,
"SCREENSHOT_LOCATE_WAIT": 10,
"SCREENSHOT_LOAD_WAIT": 10,
"SCREENSHOT_WAIT_FOR_ERROR_MODAL_VISIBLE": 10,
"SCREENSHOT_WAIT_FOR_ERROR_MODAL_INVISIBLE": 10,
"SCREENSHOT_REPLACE_UNEXPECTED_ERRORS": True,
"SCREENSHOT_TILED_ENABLED": False,
}
with patch.object(WebDriverPlaywright, "auth") as mock_auth:
mock_auth.return_value = mock_context
driver = WebDriverPlaywright("chrome")
with pytest.raises(PlaywrightTimeout) as exc_info:
driver.get_screenshot(
"http://example.com", "test-element", mock_user
)
# The exact injected instance must propagate — guards against the
# fallback alias (PlaywrightTimeout = Exception when playwright is
# not installed) accepting unrelated exceptions.
assert exc_info.value is timeout
mock_logger.exception.assert_any_call(
"Timed out requesting url %s", "http://example.com"
)