102 lines
3.1 KiB
Python
102 lines
3.1 KiB
Python
"""CORS error monitoring for browser automation."""
|
|
import logging
|
|
from typing import List, Dict, Any
|
|
from playwright.sync_api import Page
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class CORSMonitor:
|
|
"""Monitor browser console for CORS errors."""
|
|
|
|
def __init__(self, page: Page):
|
|
"""
|
|
Initialize the CORS monitor.
|
|
|
|
Args:
|
|
page: Playwright page object
|
|
"""
|
|
self.page = page
|
|
self.cors_errors: List[Dict[str, Any]] = []
|
|
self._listener = None
|
|
|
|
def start_monitoring(self) -> None:
|
|
"""Start listening for console messages."""
|
|
logger.info("Starting CORS monitoring")
|
|
self.cors_errors = []
|
|
self._listener = lambda msg: self._handle_console_message(msg)
|
|
self.page.on("console", self._listener)
|
|
|
|
def stop_monitoring(self) -> None:
|
|
"""Stop listening for console messages."""
|
|
logger.info("Stopping CORS monitoring")
|
|
if self._listener:
|
|
self.page.remove_listener("console", self._listener)
|
|
self._listener = None
|
|
|
|
def _handle_console_message(self, msg) -> None:
|
|
"""Handle a console message and check for CORS errors."""
|
|
text = msg.text.lower()
|
|
msg_type = msg.type
|
|
|
|
# Check for CORS-related errors
|
|
cors_keywords = [
|
|
"cors",
|
|
"cross-origin",
|
|
"access-control-allow-origin",
|
|
"blocked by cors",
|
|
"has been blocked by",
|
|
]
|
|
|
|
is_cors_error = any(keyword in text for keyword in cors_keywords)
|
|
|
|
if is_cors_error or (msg_type == "error" and "cors" in text):
|
|
error_info = {
|
|
"type": msg_type,
|
|
"text": msg.text,
|
|
"location": msg.location if hasattr(msg, "location") else None,
|
|
"timestamp": None, # Could add timestamp if needed
|
|
}
|
|
self.cors_errors.append(error_info)
|
|
logger.warning(f"CORS error detected: {msg.text}")
|
|
|
|
def get_cors_errors(self) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get all CORS errors captured.
|
|
|
|
Returns:
|
|
List of CORS error dictionaries
|
|
"""
|
|
return self.cors_errors.copy()
|
|
|
|
def clear_errors(self) -> None:
|
|
"""Clear all captured CORS errors."""
|
|
logger.info("Clearing CORS errors")
|
|
self.cors_errors = []
|
|
|
|
def has_errors(self) -> bool:
|
|
"""Check if any CORS errors were captured."""
|
|
return len(self.cors_errors) > 0
|
|
|
|
def get_error_summary(self) -> str:
|
|
"""
|
|
Get a summary of CORS errors for reporting.
|
|
|
|
Returns:
|
|
Markdown-formatted error summary
|
|
"""
|
|
if not self.cors_errors:
|
|
return "No CORS errors detected."
|
|
|
|
lines = [f"### CORS Errors ({len(self.cors_errors)} found)", ""]
|
|
|
|
for i, error in enumerate(self.cors_errors, 1):
|
|
lines.append(f"**Error {i}:**")
|
|
lines.append(f"- Type: {error['type']}")
|
|
lines.append(f"- Message: `{error['text']}`")
|
|
if error.get("location"):
|
|
lines.append(f"- Location: {error['location']}")
|
|
lines.append("")
|
|
|
|
return "\n".join(lines)
|