QT/superset2
1
0
Fork 0
mirror of https://github.com/apache/superset.git synced 2026-09-05 15:01:44 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
439942729cd1a09d0a1a3daab0214403b838ede3
superset2/tests/unit_tests/utils/webdriver_test.py
T
Elizabeth ThompsonandClaude Opus 4.8 439942729c fix(reports): mount off-screen dashboard charts before standalone capture 
Scheduled reports could deliver a dashboard screenshot with blank regions
where off-screen ("virtualized") charts should be. The non-tiled standalone
capture path waits only for viewport-visible chart holders to reach a
terminal state (#42624), then takes a full_page screenshot that includes
below-the-fold content. DashboardVirtualization only mounts holders that
intersect the window viewport, so any holder that never scrolls into view is
declared ready yet captured blank -- with no timeout, since the gate believes
it is done.

Grow the viewport to the full dashboard height before the readiness wait so
every holder intersects the viewport and virtualization mounts it; the
existing readiness gate then requires them all to reach a terminal state
before capture.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-07 23:02:26 +00:00

2237 lines
92 KiB
Python
Raw Blame History

# 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 ANY, call, MagicMock, patch, PropertyMock
from uuid import UUID
import pytest
from superset.utils.report_execution import (
ReportExecutionContext,
ReportExecutionDeadline,
)
from superset.utils.webdriver import (
check_playwright_availability,
PLAYWRIGHT_AVAILABLE,
PLAYWRIGHT_INSTALL_MESSAGE,
validate_webdriver_config,
WebDriverPlaywright,
WebDriverSelenium,
)
def _report_context(
*,
dashboard_id: int | None = 805,
chart_id: int | None = None,
expected_chart_count: int = 52,
) -> ReportExecutionContext:
"""Return a deterministic scheduled-report context."""
return ReportExecutionContext(
execution_id=UUID("084e7ee6-5557-4ecd-9632-b7f39c9ec524"),
report_schedule_id=11,
dashboard_id=dashboard_id,
chart_id=chart_id,
expected_chart_count=expected_chart_count,
deadline=ReportExecutionDeadline(
total_seconds=900,
started_at=0,
_clock=lambda: 0,
),
capture_reserve_seconds=60,
delivery_reserve_seconds=120,
cleanup_reserve_seconds=30,
)
@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()
@patch("superset.utils.webdriver.app")
def test_driver_sets_page_load_timeout(self, mock_app_patch: MagicMock) -> None:
"""driver.get() must be bounded so it can't block forever (#40047)."""
mock_app_patch.config = {
"SCREENSHOT_LOCATE_WAIT": 10,
"SCREENSHOT_LOAD_WAIT": 10,
"SCREENSHOT_PAGE_LOAD_WAIT": 120,
}
mock_driver = MagicMock()
driver = WebDriverSelenium(driver_type="chrome", window=(800, 600))
with patch.object(driver, "_create", return_value=mock_driver):
assert driver.driver is mock_driver
mock_driver.set_page_load_timeout.assert_called_once_with(120)
@patch("superset.utils.webdriver.app")
def test_driver_skips_page_load_timeout_when_none(
self, mock_app_patch: MagicMock
) -> None:
"""Setting SCREENSHOT_PAGE_LOAD_WAIT to None disables the bound."""
mock_app_patch.config = {
"SCREENSHOT_LOCATE_WAIT": 10,
"SCREENSHOT_LOAD_WAIT": 10,
"SCREENSHOT_PAGE_LOAD_WAIT": None,
}
mock_driver = MagicMock()
driver = WebDriverSelenium(driver_type="chrome", window=(800, 600))
with patch.object(driver, "_create", return_value=mock_driver):
assert driver.driver is mock_driver
mock_driver.set_page_load_timeout.assert_not_called()
@patch("superset.utils.webdriver.WebDriverWait")
@patch("superset.utils.webdriver.app")
def test_report_chart_uses_chart_readiness_not_dashboard_holders(
self,
mock_app_patch: MagicMock,
mock_wait: MagicMock,
) -> None:
"""Selenium chart reports require their chart terminal marker."""
from selenium.common.exceptions import TimeoutException
mock_app_patch.config = {
"SCREENSHOT_LOCATE_WAIT": 10,
"SCREENSHOT_LOAD_WAIT": 60,
"SCREENSHOT_PAGE_LOAD_WAIT": 120,
"SCREENSHOT_SELENIUM_HEADSTART": 0,
"SCREENSHOT_SELENIUM_ANIMATION_WAIT": 0,
"SCREENSHOT_REPLACE_UNEXPECTED_ERRORS": False,
}
mock_driver = MagicMock()
element = MagicMock()
mount_wait = MagicMock()
mount_wait.until.return_value = element
readiness_wait = MagicMock()
readiness_wait.until.side_effect = TimeoutException()
mock_wait.side_effect = [mount_wait, readiness_wait]
screenshot = WebDriverSelenium(driver_type="chrome")
screenshot._driver = mock_driver
with (
patch("superset.utils.webdriver.sleep"),
pytest.raises(TimeoutException),
):
screenshot.get_screenshot(
"http://example.com/chart/7",
"chart-container",
report_execution_context=_report_context(
dashboard_id=None,
chart_id=7,
expected_chart_count=1,
),
)
predicate = readiness_wait.until.call_args.args[0]
predicate(mock_driver)
readiness_js = mock_driver.execute_script.call_args.args[0]
assert "document.querySelector('.chart-container')" in readiness_js
assert "dashboard-component-chart-holder" not in readiness_js
assert element.screenshot_as_png.call_count == 0
@patch("superset.utils.webdriver.sleep")
@patch("superset.utils.webdriver.WebDriverWait")
@patch("superset.utils.webdriver.app")
def test_report_dashboard_budget_wires_selenium_timeouts_in_seconds(
self,
mock_app_patch: MagicMock,
mock_wait: MagicMock,
mock_sleep: MagicMock,
) -> None:
"""Selenium navigation, readiness, animation, and capture share one clock."""
mock_app_patch.config = {
"SCREENSHOT_LOCATE_WAIT": 10,
"SCREENSHOT_LOAD_WAIT": 60,
"SCREENSHOT_PAGE_LOAD_WAIT": 120,
"SCREENSHOT_SELENIUM_HEADSTART": 700,
"SCREENSHOT_SELENIUM_ANIMATION_WAIT": 700,
"SCREENSHOT_REPLACE_UNEXPECTED_ERRORS": False,
}
mock_driver = MagicMock()
mock_driver.execute_script.return_value = [
{"chartId": "7", "state": "rendered"}
]
element = MagicMock()
element.screenshot_as_png = b"screenshot"
mount_wait = MagicMock()
mount_wait.until.return_value = element
readiness_wait = MagicMock()
readiness_wait.until.return_value = True
mock_wait.side_effect = [mount_wait, readiness_wait]
context = ReportExecutionContext(
execution_id=UUID("084e7ee6-5557-4ecd-9632-b7f39c9ec524"),
report_schedule_id=11,
dashboard_id=805,
expected_chart_count=52,
deadline=ReportExecutionDeadline(
total_seconds=900,
started_at=0,
_clock=lambda: 100,
),
capture_reserve_seconds=60,
delivery_reserve_seconds=120,
cleanup_reserve_seconds=30,
)
screenshot = WebDriverSelenium(driver_type="chrome")
screenshot._driver = mock_driver
assert (
screenshot.get_screenshot(
"http://example.com/dashboard/805",
"standalone",
report_execution_context=context,
)
== b"screenshot"
)
# 900 total - 100 elapsed - 210 reserved = 590 seconds. Selenium APIs
# take seconds (unlike Playwright's millisecond timeouts).
mock_driver.set_page_load_timeout.assert_called_once_with(590)
assert mock_wait.call_args_list == [
call(mock_driver, 10),
call(mock_driver, 590),
]
assert mock_sleep.call_args_list == [call(590), call(590)]
assert element.screenshot_as_png == b"screenshot"
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._browser_manager")
@patch("superset.utils.webdriver.app")
def test_get_screenshot_works_when_available(self, mock_app, mock_browser_manager):
"""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_browser = MagicMock()
mock_context = MagicMock()
mock_page = MagicMock()
mock_element = MagicMock()
mock_browser_manager.get_browser.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._browser_manager")
@patch("superset.utils.webdriver.logger")
def test_get_screenshot_handles_playwright_timeout(
self, mock_logger, mock_browser_manager
):
"""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_browser = MagicMock()
mock_context = MagicMock()
mock_page = MagicMock()
mock_browser_manager.get_browser.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%s", ""
)
@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._browser_manager")
@patch("superset.utils.webdriver.app")
def test_uses_wait_for_function_to_detect_spinners(
self, mock_app, mock_browser_manager
):
"""wait_for_function polls for chart-holder readiness, not 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_browser = MagicMock()
mock_context = MagicMock()
mock_page = MagicMock()
mock_element = MagicMock()
mock_browser_manager.get_browser.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()
call_args, call_kwargs = mock_page.wait_for_function.call_args
js = call_args[0]
# The old absence-of-`.loading` predicate passed vacuously when a
# chart holder hadn't mounted anything yet; the fix requires a
# positive terminal-state check instead (same predicate as the tiled
# path, #42119).
assert "dashboard-component-chart-holder" in js
assert ".slice_container" in js
assert "data-test" not in js
assert call_kwargs["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._browser_manager")
@patch("superset.utils.webdriver.logger")
@patch("superset.utils.webdriver.app")
def test_spinner_timeout_logs_warning_and_raises(
self, mock_app, mock_logger, mock_browser_manager
):
"""Readiness timeout is logged as a warning, with per-chart diagnostics,
and re-raised rather than silently capturing."""
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_browser = MagicMock()
mock_context = MagicMock()
mock_page = MagicMock()
mock_element = MagicMock()
mock_browser_manager.get_browser.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
mock_page.evaluate.return_value = [
{"chartId": "42", "state": "nothing_mounted"}
]
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.error.assert_not_called()
warning_call = mock_logger.warning.call_args
assert "terminal_reason=readiness_timeout" in warning_call.args[0]
assert warning_call.args[1] == "http://example.com"
assert warning_call.args[3] == 1 # mounted holders
assert warning_call.args[4] == 0 # ready holders
assert warning_call.args[7] == 60
assert warning_call.args[9] == [{"chartId": "42", "state": "nothing_mounted"}]
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.logger")
def test_get_screenshot_raises_on_element_wait_timeout(
self, mock_logger, mock_browser_manager
):
"""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_browser = MagicMock()
mock_context = MagicMock()
mock_page = MagicMock()
mock_element = MagicMock()
mock_browser_manager.get_browser.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_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")
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%s", "http://example.com", ""
)
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.logger")
def test_missing_element_for_dashboard_height_falls_back_without_crashing(
self, mock_logger, mock_browser_manager
):
"""Missing dashboard element should not crash height evaluation."""
mock_user = MagicMock()
mock_user.username = "test_user"
mock_browser = MagicMock()
mock_context = MagicMock()
mock_page = MagicMock()
mock_element = MagicMock()
mock_chart_container = MagicMock()
mock_browser_manager.get_browser.return_value = mock_browser
mock_browser.new_context.return_value = mock_context
mock_context.new_page.return_value = mock_page
def locator_side_effect(selector):
if selector == ".dashboard":
return mock_element
if selector == ".chart-container":
locator = MagicMock()
locator.all.return_value = [mock_chart_container]
return locator
if selector == ".loading":
locator = MagicMock()
locator.all.return_value = []
return locator
return MagicMock()
mock_page.locator.side_effect = locator_side_effect
mock_element.wait_for.return_value = None
mock_element.screenshot.return_value = b"fake_screenshot"
mock_chart_container.wait_for.return_value = None
mock_page.wait_for_timeout.return_value = None
def evaluate_side_effect(script):
if script == 'document.querySelectorAll(".chart-container").length':
return 1
if "const target = document.querySelector" in script:
return 0
if "dashboard-component-chart-holder" in script:
return []
return None
mock_page.evaluate.side_effect = evaluate_side_effect
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": False,
"SCREENSHOT_TILED_ENABLED": True,
"SCREENSHOT_TILED_CHART_THRESHOLD": 20,
"SCREENSHOT_TILED_HEIGHT_THRESHOLD": 5000,
"SCREENSHOT_TILED_VIEWPORT_HEIGHT": 600,
}
with patch.object(WebDriverPlaywright, "auth") as mock_auth:
mock_auth.return_value = mock_context
driver = WebDriverPlaywright("chrome")
result = driver.get_screenshot(
"http://example.com", "dashboard", mock_user
)
assert result == b"fake_screenshot"
# chart_count (1) is well below the tiling threshold (20), so this is
# the benign/expected case and must not be logged as a WARNING.
mock_logger.debug.assert_any_call(
"Could not determine dashboard height for element %s "
"at url %s (%s chart containers found); %s%s",
"dashboard",
"http://example.com",
1,
"falling back to standard screenshot behavior",
"",
)
assert not any(
call.args and "Could not determine dashboard height" in call.args[0]
for call in mock_logger.warning.call_args_list
)
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.logger")
@patch("superset.utils.webdriver.take_tiled_screenshot")
def test_unknown_height_does_not_veto_tiling_for_large_dashboard(
self, mock_take_tiled, mock_logger, mock_browser_manager
):
"""
A large dashboard (by chart_count) whose height can't be measured
must still attempt tiling instead of being silently downgraded to a
standard screenshot, since below-the-fold charts may not have
rendered without the scroll-driven tiling pass.
"""
mock_user = MagicMock()
mock_user.username = "test_user"
mock_browser = MagicMock()
mock_context = MagicMock()
mock_page = MagicMock()
mock_element = MagicMock()
mock_chart_container = MagicMock()
mock_browser_manager.get_browser.return_value = mock_browser
mock_browser.new_context.return_value = mock_context
mock_context.new_page.return_value = mock_page
def locator_side_effect(selector):
if selector == ".chart-container":
locator = MagicMock()
locator.all.return_value = [mock_chart_container]
return locator
return mock_element
mock_page.locator.side_effect = locator_side_effect
mock_element.wait_for.return_value = None
mock_chart_container.wait_for.return_value = None
mock_page.wait_for_timeout.return_value = None
mock_take_tiled.return_value = b"tiled_screenshot"
def evaluate_side_effect(script):
if script == 'document.querySelectorAll(".chart-container").length':
return 25 # chart_count >= threshold
if "const target = document.querySelector" in script:
return 0 # height could not be determined
return None
mock_page.evaluate.side_effect = evaluate_side_effect
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": False,
"SCREENSHOT_TILED_ENABLED": True,
"SCREENSHOT_TILED_CHART_THRESHOLD": 20,
"SCREENSHOT_TILED_HEIGHT_THRESHOLD": 5000,
"SCREENSHOT_TILED_VIEWPORT_HEIGHT": 600,
}
with patch.object(WebDriverPlaywright, "auth") as mock_auth:
mock_auth.return_value = mock_context
driver = WebDriverPlaywright("chrome")
result = driver.get_screenshot(
"http://example.com", "dashboard", mock_user
)
assert result == b"tiled_screenshot"
mock_take_tiled.assert_called_once()
mock_logger.warning.assert_any_call(
"Could not determine dashboard height for element %s "
"at url %s (%s chart containers found); %s%s",
"dashboard",
"http://example.com",
25,
"attempting tiled screenshot anyway",
"",
)
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.logger")
def test_chart_container_timeout_logs_warning_with_progress_and_raises(
self, mock_logger, mock_browser_manager
):
"""
Timing out while waiting for `.chart-container` elements to draw must
be logged as a WARNING (matching the other locate-wait timeouts in
this method, and the customer-side-slowness convention established
for these Playwright timeouts) with rendered/total progress, and must
still fail the screenshot by re-raising.
"""
from superset.utils.webdriver import PlaywrightTimeout
mock_user = MagicMock()
mock_user.username = "test_user"
mock_browser = MagicMock()
mock_context = MagicMock()
mock_page = MagicMock()
mock_element = MagicMock()
mock_browser_manager.get_browser.return_value = mock_browser
mock_browser.new_context.return_value = mock_context
mock_context.new_page.return_value = mock_page
timeout = PlaywrightTimeout()
rendered_ok = MagicMock()
rendered_ok.wait_for.return_value = None
never_renders = MagicMock()
never_renders.wait_for.side_effect = timeout
def locator_side_effect(selector):
if selector == ".chart-container":
locator = MagicMock()
locator.all.return_value = [rendered_ok, never_renders]
return locator
return mock_element
mock_page.locator.side_effect = locator_side_effect
mock_element.wait_for.return_value = None
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": False,
"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
)
assert exc_info.value is timeout
mock_logger.warning.assert_any_call(
"Timed out waiting for chart containers to draw at url %s "
"(%s of %s chart containers rendered before the timeout)%s",
"http://example.com",
1,
2,
"",
exc_info=True,
)
mock_logger.exception.assert_not_called()
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.logger")
@patch("superset.utils.webdriver.take_tiled_screenshot")
def test_tiled_screenshot_failure_raises_without_fallback(
self, mock_take_tiled, mock_logger, mock_browser_manager
) -> None:
"""When take_tiled_screenshot returns None, fail loudly instead of
falling back to an unguarded standard screenshot."""
from superset.utils.webdriver import PlaywrightTimeout
mock_user = MagicMock()
mock_user.username = "test_user"
mock_browser = MagicMock()
mock_context = MagicMock()
mock_page = MagicMock()
mock_element = MagicMock()
mock_browser_manager.get_browser.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.wait_for.return_value = None
# page.screenshot is used by _get_screenshot for the "standalone" element;
# it must never be reached by the failure path under test.
mock_page.screenshot.return_value = b"fallback_screenshot"
def evaluate_side_effect(script):
if "querySelectorAll" in script:
return 25 # chart_count >= threshold
if "const target" in script:
return 6000 # dashboard_height > height_threshold and > tile_height
return None
mock_page.evaluate.side_effect = evaluate_side_effect
mock_take_tiled.return_value = None # tiled screenshot returns None
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": 0,
"SCREENSHOT_SELENIUM_ANIMATION_WAIT": 0,
"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": False,
"SCREENSHOT_TILED_ENABLED": True,
"SCREENSHOT_TILED_CHART_THRESHOLD": 20,
"SCREENSHOT_TILED_HEIGHT_THRESHOLD": 5000,
"SCREENSHOT_TILED_VIEWPORT_HEIGHT": 600,
}
with patch.object(WebDriverPlaywright, "auth") as mock_auth:
mock_auth.return_value = mock_context
driver = WebDriverPlaywright("chrome")
# match= keeps this assertion meaningful even when playwright
# is not installed and PlaywrightTimeout aliases bare Exception.
with pytest.raises(
PlaywrightTimeout, match="Tiled screenshot failed for url"
):
driver.get_screenshot(
"http://example.com",
"standalone",
mock_user,
report_execution_context=_report_context(),
)
mock_take_tiled.assert_called_once()
mock_page.screenshot.assert_not_called()
mock_element.screenshot.assert_not_called()
assert any(
"no safe fallback exists" in call.args[0]
for call in mock_logger.warning.call_args_list
)
class TestWebDriverPlaywrightChartReadiness:
"""Regression tests for the non-tiled vacuous-pass fix.
The readiness predicate itself (`CHART_HOLDERS_READY_JS` /
`FIND_UNREADY_CHART_HOLDERS_JS` in screenshot_utils.py) is exercised
directly in test_screenshot_utils.py; these tests confirm the standard
(non-tiled) `get_screenshot` path wires up to that *same* shared
predicate instead of the old absence-of-`.loading` check, which passed
vacuously for a chart holder that hadn't mounted anything yet.
"""
_base_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": 5,
"SCREENSHOT_WAIT_FOR_ERROR_MODAL_VISIBLE": 10,
"SCREENSHOT_WAIT_FOR_ERROR_MODAL_INVISIBLE": 10,
}
def _make_pw_mocks(self, mock_browser_manager):
mock_browser = MagicMock()
mock_context = MagicMock()
mock_page = MagicMock()
mock_element = MagicMock()
mock_browser_manager.get_browser.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"
return mock_context, mock_page
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.app")
def test_chart_holder_with_nothing_mounted_does_not_satisfy_wait(
self, mock_app, mock_browser_manager
):
"""A chart holder present in the DOM but with nothing mounted yet (no
spinner, no rendered content -- e.g. the gap between page-load
completing and React/query bootstrap) must not satisfy the readiness
wait. The old `.loading`-absence check passed immediately in this
case, producing a silently blank screenshot.
"""
from superset.utils.webdriver import PlaywrightTimeout
mock_user = MagicMock()
mock_user.username = "test_user"
mock_app.config = {**self._base_config}
mock_context, mock_page = self._make_pw_mocks(mock_browser_manager)
def fake_wait_for_function(js, timeout=None):
# Confirm the predicate sent is the shared terminal-state check
# (not the old absence-of-`.loading` check), then simulate it
# correctly reporting "not ready" for a chart holder that hasn't
# mounted anything.
assert "dashboard-component-chart-holder" in js
raise PlaywrightTimeout("Timeout waiting for chart holders")
mock_page.wait_for_function.side_effect = fake_wait_for_function
mock_page.evaluate.return_value = [{"chartId": "7", "state": "nothing_mounted"}]
with patch.object(WebDriverPlaywright, "auth", return_value=mock_context):
driver = WebDriverPlaywright("chrome")
with pytest.raises(PlaywrightTimeout):
driver.get_screenshot("http://example.com", "test-element", mock_user)
# No screenshot should be captured -- fail loudly instead of
# silently returning a blank image.
mock_page.screenshot.assert_not_called()
mock_page.locator.return_value.screenshot.assert_not_called()
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.app")
def test_all_chart_holders_ready_passes(self, mock_app, mock_browser_manager):
"""All chart holders rendered or errored -> wait passes, screenshot taken."""
mock_user = MagicMock()
mock_user.username = "test_user"
mock_app.config = {**self._base_config}
mock_context, mock_page = self._make_pw_mocks(mock_browser_manager)
# mock_page.wait_for_function is a no-op MagicMock by default, i.e.
# the readiness predicate is satisfied immediately.
with patch.object(WebDriverPlaywright, "auth", return_value=mock_context):
driver = WebDriverPlaywright("chrome")
result = driver.get_screenshot(
"http://example.com", "test-element", mock_user
)
assert result == b"screenshot"
# Readiness diagnostics are emitted before polling so a task killed by
# an outer limit still leaves useful state in the logs.
assert mock_page.evaluate.call_count == 2
assert all(
"state: 'rendered'" in call.args[0]
for call in mock_page.evaluate.call_args_list
)
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.app")
def test_chart_capture_uses_positive_terminal_state_predicate(
self, mock_app, mock_browser_manager
):
"""Chart captures require a terminal marker and no loading marker."""
from superset.utils.webdriver import PlaywrightTimeout
mock_app.config = {**self._base_config}
mock_context, mock_page = self._make_pw_mocks(mock_browser_manager)
timeout = PlaywrightTimeout("chart not ready")
mock_page.wait_for_function.side_effect = timeout
with patch.object(WebDriverPlaywright, "auth", return_value=mock_context):
with pytest.raises(PlaywrightTimeout):
WebDriverPlaywright("chrome").get_screenshot(
"http://example.com",
"chart-container",
MagicMock(),
report_execution_context=_report_context(
dashboard_id=None,
chart_id=7,
expected_chart_count=1,
),
)
predicate = mock_page.wait_for_function.call_args.args[0]
assert "document.querySelector('.chart-container')" in predicate
assert ".slice_container" in predicate
assert ".loading" in predicate
assert '[role="alert"]' in predicate
assert ".ant-empty" in predicate
assert ".missing-chart-container" in predicate
mock_page.locator.return_value.screenshot.assert_not_called()
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.logger")
@patch("superset.utils.webdriver.app")
def test_standalone_zero_holders_remain_not_ready_and_skip_capture(
self, mock_app, mock_logger, mock_browser_manager
):
from superset.utils.webdriver import PlaywrightTimeout
mock_app.config = {**self._base_config}
mock_context, mock_page = self._make_pw_mocks(mock_browser_manager)
mock_page.evaluate.return_value = []
mock_page.wait_for_function.side_effect = PlaywrightTimeout("zero holders")
with patch.object(WebDriverPlaywright, "auth", return_value=mock_context):
with pytest.raises(PlaywrightTimeout):
WebDriverPlaywright("chrome").get_screenshot(
"http://example.com",
"standalone",
MagicMock(),
report_execution_context=_report_context(),
)
assert any(
"report_readiness_waiting_for_mount" in call.args[0]
for call in mock_logger.info.call_args_list
)
assert (
"terminal_reason=readiness_timeout" in mock_logger.warning.call_args.args[0]
)
mock_page.screenshot.assert_not_called()
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.app")
def test_thumbnail_zero_holders_preserves_existing_capture_behavior(
self,
mock_app,
mock_browser_manager,
):
mock_app.config = {**self._base_config}
mock_context, mock_page = self._make_pw_mocks(mock_browser_manager)
mock_page.evaluate.return_value = []
with patch.object(WebDriverPlaywright, "auth", return_value=mock_context):
result = WebDriverPlaywright("chrome").get_screenshot(
"http://example.com",
"standalone",
MagicMock(),
)
predicate = mock_page.wait_for_function.call_args.args[0]
assert "holders.length > 0" not in predicate
assert result == mock_page.screenshot.return_value
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.app")
def test_readiness_check_scoped_to_viewport_visible_holders(
self, mock_app, mock_browser_manager
):
"""The non-tiled readiness check only requires viewport-intersecting
chart holders to be ready, mirroring the tiled path's reasoning
(#42119). `get_screenshot`'s standard (non-tiled) branch never
resizes the browser viewport to the full dashboard height before
capturing -- only the tiled branch's `set_viewport_size` call does
that -- so a below-the-fold chart holder is a
DashboardVirtualization placeholder that hasn't mounted anything
real yet by design and must not block this wait.
"""
mock_user = MagicMock()
mock_user.username = "test_user"
mock_app.config = {**self._base_config}
mock_context, mock_page = self._make_pw_mocks(mock_browser_manager)
with patch.object(WebDriverPlaywright, "auth", return_value=mock_context):
driver = WebDriverPlaywright("chrome")
driver.get_screenshot("http://example.com", "test-element", mock_user)
js = mock_page.wait_for_function.call_args[0][0]
# The predicate skips any chart holder whose bounding rect doesn't
# intersect the current viewport -- an off-screen/below-fold holder
# is excluded from the readiness requirement rather than blocking it.
assert "getBoundingClientRect" in js
assert "window.innerHeight" in js
# The non-tiled branch resizes the viewport only for a "standalone"
# full-page dashboard capture (to mount off-screen holders); this test
# uses a non-standalone element, so no resize happens and the
# viewport-scoped predicate is what keeps below-the-fold holders from
# blocking the wait.
mock_page.set_viewport_size.assert_not_called()
def test_mount_offscreen_holders_resizes_tall_standalone_dashboard(self):
"""A dashboard taller than the viewport is grown to its full height so
DashboardVirtualization mounts every off-screen ("virtualized") chart
holder before the standalone full_page capture."""
page = MagicMock()
page.evaluate.return_value = 4000
WebDriverPlaywright._mount_offscreen_chart_holders(
page, "http://example.com/dashboard/21", "standalone", 800, 600
)
page.set_viewport_size.assert_called_once_with({"width": 800, "height": 4000})
def test_mount_offscreen_holders_skips_when_dashboard_fits_viewport(self):
"""Nothing is off-screen when the dashboard already fits the viewport,
so the viewport is left untouched."""
page = MagicMock()
page.evaluate.return_value = 500
WebDriverPlaywright._mount_offscreen_chart_holders(
page, "http://example.com/dashboard/21", "standalone", 800, 600
)
page.set_viewport_size.assert_not_called()
def test_mount_offscreen_holders_scoped_to_standalone(self):
"""Chart captures have no dashboard grid to virtualize, so the helper
no-ops (it never even measures the page) for non-standalone elements."""
page = MagicMock()
WebDriverPlaywright._mount_offscreen_chart_holders(
page, "http://example.com", "chart-container", 800, 600
)
page.evaluate.assert_not_called()
page.set_viewport_size.assert_not_called()
@patch("superset.utils.webdriver.logger")
def test_mount_offscreen_holders_caps_height_and_warns(self, mock_logger):
"""A pathologically tall dashboard is capped at the memory guard and a
warning recommends enabling tiled screenshots."""
from superset.utils.webdriver import MAX_STANDALONE_CAPTURE_VIEWPORT_HEIGHT
page = MagicMock()
page.evaluate.return_value = MAX_STANDALONE_CAPTURE_VIEWPORT_HEIGHT + 50000
WebDriverPlaywright._mount_offscreen_chart_holders(
page, "http://example.com/dashboard/21", "standalone", 800, 600
)
page.set_viewport_size.assert_called_once_with(
{"width": 800, "height": MAX_STANDALONE_CAPTURE_VIEWPORT_HEIGHT}
)
assert any(
"exceeds the standalone" in call.args[0]
for call in mock_logger.warning.call_args_list
)
@patch("superset.utils.webdriver.logger")
def test_mount_offscreen_holders_swallows_errors(self, mock_logger):
"""Mounting is best-effort: a failure must never abort the capture."""
page = MagicMock()
page.evaluate.side_effect = RuntimeError("boom")
# Must not raise.
WebDriverPlaywright._mount_offscreen_chart_holders(
page, "http://example.com/dashboard/21", "standalone", 800, 600
)
page.set_viewport_size.assert_not_called()
assert mock_logger.warning.called
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.app")
def test_standalone_report_expands_viewport_before_readiness(
self, mock_app, mock_browser_manager
):
"""Regression for blank scheduled reports: the non-tiled standalone
report path grows the viewport to the whole dashboard (mounting
off-screen holders) *before* the readiness wait, so virtualized
below-the-fold charts are required to render rather than captured
blank."""
from superset.utils.screenshot_utils import DASHBOARD_CONTENT_HEIGHT_JS
mock_app.config = {**self._base_config}
mock_context, mock_page = self._make_pw_mocks(mock_browser_manager)
def fake_eval(script, *args, **kwargs):
if script == DASHBOARD_CONTENT_HEIGHT_JS:
return 4000
if "scrollTo" in script:
return None
return [{"chartId": "7", "state": "rendered"}]
mock_page.evaluate.side_effect = fake_eval
with patch.object(WebDriverPlaywright, "auth", return_value=mock_context):
result = WebDriverPlaywright("chrome").get_screenshot(
"http://example.com/dashboard/21",
"standalone",
MagicMock(),
report_execution_context=_report_context(),
)
mock_page.set_viewport_size.assert_called_once_with(
{"width": 800, "height": 4000}
)
call_names = [c[0] for c in mock_page.mock_calls]
assert call_names.index("set_viewport_size") < call_names.index(
"wait_for_function"
)
assert result == mock_page.screenshot.return_value
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.logger")
@patch("superset.utils.webdriver.app")
def test_log_context_threaded_into_readiness_wait(
self, mock_app, mock_logger, mock_browser_manager
):
"""log_context (e.g. report execution id) is threaded through the
non-tiled readiness wait for correlation, matching #42119's
convention for the tiled path."""
from superset.utils.webdriver import PlaywrightTimeout
mock_user = MagicMock()
mock_user.username = "test_user"
mock_app.config = {**self._base_config}
mock_context, mock_page = self._make_pw_mocks(mock_browser_manager)
mock_page.wait_for_function.side_effect = PlaywrightTimeout("timed out")
mock_page.evaluate.return_value = [{"chartId": "7", "state": "nothing_mounted"}]
with patch.object(WebDriverPlaywright, "auth", return_value=mock_context):
driver = WebDriverPlaywright("chrome")
with pytest.raises(PlaywrightTimeout):
driver.get_screenshot(
"http://example.com",
"test-element",
mock_user,
log_context="execution_id=abc-123",
)
assert mock_logger.warning.call_args.args[8] == " [execution_id=abc-123]"
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.logger")
@patch("superset.utils.webdriver.app")
def test_wait_is_capped_to_remaining_runtime_task_budget(
self, mock_app, mock_logger, mock_browser_manager
):
"""Elapsed setup time is removed from the task-derived safe budget."""
mock_user = MagicMock()
mock_user.username = "test_user"
mock_app.config = {
**self._base_config,
"SCREENSHOT_LOAD_WAIT": 600,
}
mock_context, mock_page = self._make_pw_mocks(mock_browser_manager)
with (
patch.object(WebDriverPlaywright, "auth", return_value=mock_context),
patch(
"superset.utils.webdriver.resolve_screenshot_task_budget_seconds",
return_value=240,
),
patch(
"superset.utils.webdriver.time.monotonic",
side_effect=[100.0, 110.0],
),
):
WebDriverPlaywright("chrome").get_screenshot(
"http://example.com", "test-element", mock_user
)
assert mock_page.wait_for_function.call_args.kwargs["timeout"] == 230_000
def test_report_readiness_uses_shared_deadline_and_phase_reserves(self):
from uuid import UUID
from superset.utils.report_execution import (
ReportExecutionContext,
ReportExecutionDeadline,
)
page = MagicMock()
page.evaluate.return_value = [{"chartId": "7", "state": "rendered"}]
deadline = ReportExecutionDeadline(
total_seconds=900,
started_at=0,
_clock=lambda: 100,
)
report_context = ReportExecutionContext(
execution_id=UUID("084e7ee6-5557-4ecd-9632-b7f39c9ec524"),
report_schedule_id=11,
dashboard_id=805,
expected_chart_count=52,
deadline=deadline,
capture_reserve_seconds=60,
delivery_reserve_seconds=120,
cleanup_reserve_seconds=30,
)
WebDriverPlaywright._wait_for_charts_ready(
page,
"http://example.com/dashboard/805",
5,
"standalone",
report_execution_context=report_context,
)
assert page.wait_for_function.call_args.kwargs["timeout"] == 590_000
def test_report_readiness_budget_exhaustion_skips_poll_and_capture(self):
from uuid import UUID
from superset.utils.report_execution import (
ReportExecutionBudgetExceededError,
ReportExecutionContext,
ReportExecutionDeadline,
)
page = MagicMock()
page.evaluate.return_value = []
deadline = ReportExecutionDeadline(
total_seconds=900,
started_at=0,
_clock=lambda: 700,
)
report_context = ReportExecutionContext(
execution_id=UUID("084e7ee6-5557-4ecd-9632-b7f39c9ec524"),
report_schedule_id=11,
dashboard_id=805,
expected_chart_count=52,
deadline=deadline,
capture_reserve_seconds=60,
delivery_reserve_seconds=120,
cleanup_reserve_seconds=30,
)
with pytest.raises(ReportExecutionBudgetExceededError):
WebDriverPlaywright._wait_for_charts_ready(
page,
"http://example.com/dashboard/805",
600,
"standalone",
report_execution_context=report_context,
)
page.wait_for_function.assert_not_called()
page.screenshot.assert_not_called()
@patch("superset.utils.webdriver.logger")
def test_chart_capture_ready_logs_container_state_not_holder_counts(
self, mock_logger
):
"""Chart pages have no dashboard grid holders, so the ready line must
report the `.chart-container` state instead of vacuous zero counters
(which read as "no charts" in customer logs)."""
from superset.utils.screenshot_utils import CHART_CONTAINER_STATE_JS
page = MagicMock()
page.wait_for_function.return_value = None
page.evaluate.side_effect = lambda script: (
"terminal" if script == CHART_CONTAINER_STATE_JS else []
)
with patch(
"superset.utils.webdriver.resolve_screenshot_task_budget_seconds",
return_value=None,
):
WebDriverPlaywright._wait_for_charts_ready(
page,
"http://example.com",
10,
"chart-container",
log_context="capture_kind=alert execution_id=abc-123",
)
ready_call = next(
call
for call in mock_logger.info.call_args_list
if call.args and call.args[0].startswith("report_readiness_ready")
)
assert "target=chart-container" in ready_call.args[0]
assert "mounted_holders" not in ready_call.args[0]
assert "terminal" in ready_call.args
assert " [capture_kind=alert execution_id=abc-123]" in ready_call.args
@patch("superset.utils.webdriver.logger")
def test_chart_capture_timeout_logs_container_state(self, mock_logger):
from superset.utils.screenshot_utils import CHART_CONTAINER_STATE_JS
from superset.utils.webdriver import PlaywrightTimeout
page = MagicMock()
page.wait_for_function.side_effect = PlaywrightTimeout()
page.evaluate.side_effect = lambda script: (
"loading" if script == CHART_CONTAINER_STATE_JS else []
)
with (
patch(
"superset.utils.webdriver.resolve_screenshot_task_budget_seconds",
return_value=None,
),
pytest.raises(PlaywrightTimeout),
):
WebDriverPlaywright._wait_for_charts_ready(
page,
"http://example.com",
10,
"chart-container",
)
terminal_call = next(
call
for call in mock_logger.warning.call_args_list
if call.args and call.args[0].startswith("report_readiness_terminal")
)
assert "target=chart-container" in terminal_call.args[0]
assert "terminal_reason=readiness_timeout" in terminal_call.args[0]
assert "mounted_holders" not in terminal_call.args[0]
assert "loading" in terminal_call.args
def test_zero_load_wait_without_task_budget_preserves_playwright_no_timeout(self):
page = MagicMock()
page.evaluate.return_value = []
with patch(
"superset.utils.webdriver.resolve_screenshot_task_budget_seconds",
return_value=None,
):
WebDriverPlaywright._wait_for_charts_ready(
page,
"http://example.com",
0,
"chart-container",
)
page.wait_for_function.assert_called_once()
assert page.wait_for_function.call_args.kwargs["timeout"] == 0
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.logger")
@patch("superset.utils.webdriver.app")
def test_exhausted_task_budget_raises_before_capture(
self, mock_app, mock_logger, mock_browser_manager
):
"""An exhausted budget fails loudly while cleanup time remains."""
from superset.utils.webdriver import ScreenshotTaskBudgetExceededError
mock_user = MagicMock()
mock_app.config = {**self._base_config, "SCREENSHOT_LOAD_WAIT": 600}
mock_context, mock_page = self._make_pw_mocks(mock_browser_manager)
diagnostics = [{"chartId": "17", "state": "nothing_mounted"}]
mock_page.evaluate.return_value = diagnostics
with (
patch.object(WebDriverPlaywright, "auth", return_value=mock_context),
patch(
"superset.utils.webdriver.resolve_screenshot_task_budget_seconds",
return_value=240,
),
patch(
"superset.utils.webdriver.time.monotonic",
side_effect=[100.0, 341.0],
),
):
with pytest.raises(ScreenshotTaskBudgetExceededError):
WebDriverPlaywright("chrome").get_screenshot(
"http://example.com", "test-element", mock_user
)
mock_page.wait_for_function.assert_not_called()
mock_page.locator.return_value.screenshot.assert_not_called()
mock_context.close.assert_called_once()
warning_args = mock_logger.warning.call_args.args
assert "budget exhausted" in warning_args[0]
assert warning_args[5] == diagnostics
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.logger")
@patch("superset.utils.webdriver.app")
def test_unready_diagnostics_logged_early_and_at_failure(
self, mock_app, mock_logger, mock_browser_manager
):
"""Unready IDs/states are logged before polling and on timeout."""
from superset.utils.webdriver import PlaywrightTimeout
mock_user = MagicMock()
mock_user.username = "test_user"
mock_app.config = {**self._base_config}
mock_context, mock_page = self._make_pw_mocks(mock_browser_manager)
diagnostics = [{"chartId": "17", "state": "waiting_on_database"}]
mock_page.evaluate.return_value = diagnostics
mock_page.wait_for_function.side_effect = PlaywrightTimeout("timed out")
with patch.object(WebDriverPlaywright, "auth", return_value=mock_context):
driver = WebDriverPlaywright("chrome")
with pytest.raises(PlaywrightTimeout):
driver.get_screenshot("http://example.com", "test-element", mock_user)
mock_logger.info.assert_any_call(
"Chart holders not ready before polling at url %s%s: %s",
"http://example.com",
"",
diagnostics,
)
failure_args = mock_logger.warning.call_args.args
assert failure_args[9] == diagnostics
assert failure_args[10] == diagnostics
mock_page.locator.return_value.screenshot.assert_not_called()
class TestWebDriverPlaywrightAnimationWaitOrder:
"""Animation wait must run after the spinner wait, not before."""
_base_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": 2,
"SCREENSHOT_REPLACE_UNEXPECTED_ERRORS": False,
"SCREENSHOT_LOCATE_WAIT": 10,
"SCREENSHOT_LOAD_WAIT": 30,
"SCREENSHOT_WAIT_FOR_ERROR_MODAL_VISIBLE": 10,
"SCREENSHOT_WAIT_FOR_ERROR_MODAL_INVISIBLE": 10,
}
def _make_pw_mocks(self, mock_browser_manager):
mock_browser = MagicMock()
mock_context = MagicMock()
mock_page = MagicMock()
mock_element = MagicMock()
mock_browser_manager.get_browser.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"
return mock_context, mock_page
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.app")
def test_animation_wait_after_spinner_wait_tiled_disabled(
self, mock_app, mock_browser_manager
):
"""Non-tiled path: animation wait runs after spinner wait_for_function."""
mock_user = MagicMock()
mock_user.username = "test_user"
mock_app.config = {**self._base_config, "SCREENSHOT_TILED_ENABLED": False}
mock_context, mock_page = self._make_pw_mocks(mock_browser_manager)
call_order: list[str] = []
def record_wait_for_function(*args, **kwargs):
call_order.append("spinner_wait")
def record_wait_for_timeout(ms):
if ms == 2 * 1000:
call_order.append("animation_wait")
mock_page.wait_for_function.side_effect = record_wait_for_function
mock_page.wait_for_timeout.side_effect = record_wait_for_timeout
with patch.object(WebDriverPlaywright, "auth", return_value=mock_context):
WebDriverPlaywright("chrome").get_screenshot(
"http://example.com", "test-element", mock_user
)
assert "spinner_wait" in call_order
assert "animation_wait" in call_order
spinner_idx = call_order.index("spinner_wait")
anim_idx = call_order.index("animation_wait")
assert spinner_idx < anim_idx, (
"spinner wait must precede animation wait in non-tiled path"
)
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.app")
def test_animation_wait_after_spinner_wait_tiled_enabled_small_dashboard(
self, mock_app, mock_browser_manager
):
"""Non-tiled path (tiled on, small dashboard): animation after spinner."""
mock_user = MagicMock()
mock_user.username = "test_user"
mock_app.config = {
**self._base_config,
"SCREENSHOT_TILED_ENABLED": True,
"SCREENSHOT_TILED_CHART_THRESHOLD": 20,
"SCREENSHOT_TILED_HEIGHT_THRESHOLD": 5000,
"SCREENSHOT_TILED_VIEWPORT_HEIGHT": 600,
}
mock_context, mock_page = self._make_pw_mocks(mock_browser_manager)
# Small dashboard: 3 charts, 1000px height — below both thresholds
mock_page.evaluate.side_effect = [3, 1000, [], []]
call_order: list[str] = []
def record_wait_for_function(*args, **kwargs):
call_order.append("spinner_wait")
def record_wait_for_timeout(ms):
if ms == 2 * 1000:
call_order.append("animation_wait")
mock_page.wait_for_function.side_effect = record_wait_for_function
mock_page.wait_for_timeout.side_effect = record_wait_for_timeout
with patch.object(WebDriverPlaywright, "auth", return_value=mock_context):
WebDriverPlaywright("chrome").get_screenshot(
"http://example.com", "test-element", mock_user
)
assert "spinner_wait" in call_order
assert "animation_wait" in call_order
assert call_order.index("spinner_wait") < call_order.index("animation_wait")
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.take_tiled_screenshot")
@patch("superset.utils.webdriver.app")
def test_chart_threshold_does_not_tile_short_dashboard(
self, mock_app, mock_take_tiled, mock_browser_manager
):
"""Preserve the historical height guard for reports and thumbnails."""
mock_user = MagicMock()
mock_user.username = "test_user"
mock_app.config = {
**self._base_config,
"SCREENSHOT_TILED_ENABLED": True,
"SCREENSHOT_TILED_CHART_THRESHOLD": 20,
"SCREENSHOT_TILED_HEIGHT_THRESHOLD": 5000,
"SCREENSHOT_TILED_VIEWPORT_HEIGHT": 600,
}
mock_context, mock_page = self._make_pw_mocks(mock_browser_manager)
mock_page.evaluate.side_effect = [25, 500, [], []]
with patch.object(WebDriverPlaywright, "auth", return_value=mock_context):
result = WebDriverPlaywright("chrome").get_screenshot(
"http://example.com", "test-element", mock_user
)
assert result == b"screenshot"
mock_take_tiled.assert_not_called()
mock_page.set_viewport_size.assert_not_called()
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.take_tiled_screenshot")
@patch("superset.utils.webdriver.app")
def test_tiled_path_passes_animation_wait_per_tile_no_global_wait(
self, mock_app, mock_take_tiled, mock_browser_manager
):
"""Tiled path delegates animation_wait to take_tiled_screenshot; no global."""
mock_user = MagicMock()
mock_user.username = "test_user"
mock_app.config = {
**self._base_config,
"SCREENSHOT_TILED_ENABLED": True,
"SCREENSHOT_TILED_CHART_THRESHOLD": 20,
"SCREENSHOT_TILED_HEIGHT_THRESHOLD": 5000,
"SCREENSHOT_TILED_VIEWPORT_HEIGHT": 600,
}
mock_context, mock_page = self._make_pw_mocks(mock_browser_manager)
# Large dashboard: 25 charts, 6000px height
mock_page.evaluate.side_effect = [25, 6000]
mock_take_tiled.return_value = b"tiled_screenshot"
with patch.object(WebDriverPlaywright, "auth", return_value=mock_context):
result = WebDriverPlaywright("chrome").get_screenshot(
"http://example.com", "standalone", mock_user
)
assert result == b"tiled_screenshot"
mock_take_tiled.assert_called_once_with(
mock_page,
"standalone",
600,
load_wait=30,
animation_wait=2,
log_context=None,
report_execution_context=None,
url="http://example.com",
screenshot_started_at=ANY,
)
# The only wait_for_timeout call should be the 0ms headstart; no global
# animation wait should be issued (handled per-tile by take_tiled_screenshot)
animation_waits = [
call[0][0]
for call in mock_page.wait_for_timeout.call_args_list
if call[0][0] == 2 * 1000
]
assert animation_waits == [], (
"No global 2s animation wait_for_timeout should fire on the tiled path"
)
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.take_tiled_screenshot")
@patch("superset.utils.webdriver.app")
def test_tiled_empty_bytes_raises_without_fallback(
self, mock_app, mock_take_tiled, mock_browser_manager
):
"""Tiled failure raises when take_tiled_screenshot returns b"" (not None),
instead of silently falling through to an unguarded raw capture."""
from superset.utils.webdriver import PlaywrightTimeout
mock_user = MagicMock()
mock_user.username = "test_user"
mock_app.config = {
**self._base_config,
"SCREENSHOT_TILED_ENABLED": True,
"SCREENSHOT_TILED_CHART_THRESHOLD": 20,
"SCREENSHOT_TILED_HEIGHT_THRESHOLD": 5000,
"SCREENSHOT_TILED_VIEWPORT_HEIGHT": 600,
}
mock_context, mock_page = self._make_pw_mocks(mock_browser_manager)
mock_page.evaluate.side_effect = [25, 6000]
# Empty bytes — falsy but not None; was silently passed through before the fix
mock_take_tiled.return_value = b""
# _get_screenshot("standalone") calls page.screenshot(full_page=True); it
# must never be reached by the failure path under test.
mock_page.screenshot.return_value = b"fallback"
with patch.object(WebDriverPlaywright, "auth", return_value=mock_context):
# match= keeps this assertion meaningful even when playwright
# is not installed and PlaywrightTimeout aliases bare Exception.
with pytest.raises(
PlaywrightTimeout, match="Tiled screenshot failed for url"
):
WebDriverPlaywright("chrome").get_screenshot(
"http://example.com", "standalone", mock_user
)
# Tiled path was taken (take_tiled_screenshot was called)
mock_take_tiled.assert_called_once()
# Standard screenshot must never be called as a fallback
mock_page.screenshot.assert_not_called()
@patch("superset.utils.webdriver.PLAYWRIGHT_AVAILABLE", True)
@patch("superset.utils.webdriver._browser_manager")
@patch("superset.utils.webdriver.app")
def test_animation_wait_skipped_when_zero(self, mock_app, mock_browser_manager):
"""No extra wait_for_timeout call when SCREENSHOT_SELENIUM_ANIMATION_WAIT=0."""
mock_user = MagicMock()
mock_user.username = "test_user"
mock_app.config = {
**self._base_config,
"SCREENSHOT_SELENIUM_ANIMATION_WAIT": 0,
"SCREENSHOT_TILED_ENABLED": False,
}
mock_context, mock_page = self._make_pw_mocks(mock_browser_manager)
with patch.object(WebDriverPlaywright, "auth", return_value=mock_context):
WebDriverPlaywright("chrome").get_screenshot(
"http://example.com", "test-element", mock_user
)
# Only headstart (0ms) should be called; no animation wait call
timeout_values = [
call[0][0] for call in mock_page.wait_for_timeout.call_args_list
]
assert timeout_values == [0], (
f"Expected only [0] (headstart), got {timeout_values}"
)
Reference in New Issue View Git Blame Copy Permalink