diff --git a/config/config_run.yaml.example b/config/config_run.yaml.example index 74e7190..5c070bc 100644 --- a/config/config_run.yaml.example +++ b/config/config_run.yaml.example @@ -48,6 +48,43 @@ cases: - verify: Verify the page displays the Baidu logo and many Pictures display - action: Click on a random image to view + # Example: Fixture-like state management (pytest-style) + # Define fixtures by setting non-empty snapshot: + # + # 1. Define fixtures (snapshot non-empty): + # - name: Setup Admin Login + # snapshot: "admin_user" # Non-empty = auto-save after execution + # steps: + # - action: Enter username 'admin' + # - action: Enter password 'admin123' + # - action: Click login button + # + # - name: Setup Normal User Login + # snapshot: "normal_user" # Another fixture + # steps: + # - action: Enter username 'user' + # - action: Enter password 'user123' + # - action: Click login button + # + # 2. Use fixtures (use_snapshot): + # - name: Test Admin Dashboard + # use_snapshot: "admin_user" # Load admin_user fixture + # steps: + # - action: Navigate to admin dashboard + # - verify: Verify admin permissions + # + # - name: Test User Profile + # use_snapshot: "normal_user" # Load normal_user fixture + # steps: + # - action: View profile page + # - verify: Verify user information + # + # - name: Test Public Page + # # No use_snapshot = independent test (no pre-loaded state) + # steps: + # - action: Visit public homepage + # - verify: Verify page loads correctly + log: level: info diff --git a/webqa_agent/browser/context_manager.py b/webqa_agent/browser/context_manager.py new file mode 100644 index 0000000..ea169b2 --- /dev/null +++ b/webqa_agent/browser/context_manager.py @@ -0,0 +1,195 @@ +"""Persistent BrowserContext Management + +This module provides utilities for persisting Playwright BrowserContext state +to local filesystem using storage_state API. + +Key Features: +- Storage state persistence (cookies, localStorage, sessionStorage) +- File-based state management with atomic writes +- Thread-safe operations with file locking +- Cross-session state reuse +""" + +import json +import logging +from pathlib import Path +from typing import Optional + +from playwright.async_api import BrowserContext + +try: + import filelock + HAS_FILELOCK = True +except ImportError: + HAS_FILELOCK = False + logging.warning( + "filelock not installed. Concurrent context saves may have race conditions. " + "Install with: pip install filelock" + ) + + +class PersistentContextManager: + """Persistent context manager (stateless utility class). + + Manages storage_state files for Playwright BrowserContext persistence. + All methods are static for simplicity and thread-safety. + """ + + @staticmethod + def get_storage_path( + snapshot_id: str, + base_dir: str = 'webqa_agent/browser/browser_context' + ) -> Path: + """Calculate storage_state JSON file path. + + Args: + snapshot_id: Unique identifier for the snapshot + base_dir: Base directory for storage (default: webqa_agent/browser/browser_context) + + Returns: + Path object: {base_dir}/{snapshot_id}.json + + Raises: + ValueError: If snapshot_id contains invalid characters (path traversal attempt) + """ + # Security: Prevent path traversal attacks + if not snapshot_id or '/' in snapshot_id or '\\' in snapshot_id or '..' in snapshot_id: + raise ValueError( + f"Invalid snapshot_id: '{snapshot_id}'. " + "Must not contain path separators or '..' for security." + ) + + # Convert to Path object (no directory creation side effect) + base_path = Path(base_dir) + storage_path = base_path / f'{snapshot_id}.json' + return storage_path + + @staticmethod + async def get_storage_state_path( + snapshot_id: str, + base_dir: str = 'webqa_agent/browser/browser_context' + ) -> Optional[str]: + """Get storage_state file path if it exists and is valid. + + Args: + snapshot_id: Unique identifier for the snapshot + base_dir: Base directory for storage + + Returns: + str: File path if exists and valid, None otherwise + """ + try: + storage_path = PersistentContextManager.get_storage_path(snapshot_id, base_dir) + + # Check if file exists + if not storage_path.exists(): + logging.debug(f"[PersistentContext] No saved state found for snapshot_id: {snapshot_id}") + return None + + # Validate JSON format + try: + with open(storage_path, 'r', encoding='utf-8') as f: + json.load(f) # Validate JSON + except json.JSONDecodeError as e: + logging.warning( + f"[PersistentContext] Corrupted storage_state file for {snapshot_id}: {e}. " + "Will create new context." + ) + return None + + logging.info(f"[PersistentContext] Found saved state for snapshot_id: {snapshot_id}") + return str(storage_path) + + except Exception as e: + logging.error(f"[PersistentContext] Failed to get storage_state path for {snapshot_id}: {e}") + return None + + @staticmethod + async def save_storage_state( + context: BrowserContext, + snapshot_id: str, + base_dir: str = 'webqa_agent/browser/browser_context' + ) -> None: + """Save context storage_state to file with atomic write and file locking. + + Args: + context: Playwright BrowserContext to save + snapshot_id: Unique identifier for the snapshot + base_dir: Base directory for storage + + Raises: + Exception: If save operation fails + """ + storage_path = PersistentContextManager.get_storage_path(snapshot_id, base_dir) + tmp_path = storage_path.with_suffix('.json.tmp') + + try: + # Ensure directory exists before writing + storage_path.parent.mkdir(parents=True, exist_ok=True) + + # Step 1: Write to temporary file + await context.storage_state(path=str(tmp_path)) + + # Step 2: Atomic rename with file lock (if available) + if HAS_FILELOCK: + lock_path = storage_path.with_suffix('.lock') + lock = filelock.FileLock(str(lock_path), timeout=10) + + try: + with lock: + tmp_path.replace(storage_path) # Atomic operation: rename tmp -> final + try: + lock_path.unlink(missing_ok=True) # Cleanup lock file + except OSError: + pass # Lock file cleanup is best-effort + + except filelock.Timeout: + logging.warning( + f"[PersistentContext] Lock timeout while saving {snapshot_id}. " + "Proceeding without lock." + ) + # Fallback: rename without lock + tmp_path.replace(storage_path) + else: + # No filelock available: direct rename (not fully atomic in concurrent scenarios) + tmp_path.replace(storage_path) + + logging.info(f"[PersistentContext] Saved storage_state for snapshot_id: {snapshot_id}") + + except Exception as e: + logging.error(f"[PersistentContext] Failed to save storage_state for {snapshot_id}: {e}") + try: + if tmp_path.exists(): + tmp_path.unlink() # Cleanup temporary file on failure + except OSError: + pass + raise + + @staticmethod + def delete_storage_state( + snapshot_id: str, + base_dir: str = 'webqa_agent/browser/browser_context' + ) -> bool: + """Delete storage_state file for specified snapshot_id. + + Args: + snapshot_id: Unique identifier for the snapshot + base_dir: Base directory for storage + + Returns: + bool: True if deleted successfully, False if file not found or error + """ + try: + storage_path = PersistentContextManager.get_storage_path(snapshot_id, base_dir) + + if storage_path.exists(): + storage_path.unlink() + logging.info(f"[PersistentContext] Cleaned up snapshot_id: {snapshot_id}") + return True + else: + logging.debug(f"[PersistentContext] No file to cleanup for snapshot_id: {snapshot_id}") + return False + + except Exception as e: + logging.error(f"[PersistentContext] Failed to cleanup snapshot_id {snapshot_id}: {e}") + return False diff --git a/webqa_agent/data/case_structures.py b/webqa_agent/data/case_structures.py index d1d0434..11f9b00 100644 --- a/webqa_agent/data/case_structures.py +++ b/webqa_agent/data/case_structures.py @@ -134,11 +134,20 @@ def parse_yaml_format(cls, data: Any) -> Dict[str, Any]: class Case(BaseModel): - """Test case with name and steps.""" + """Test case with fixture support. + + Fields: + name: Test case name + steps: List of action/verify steps + snapshot: If non-empty, this case is a fixture and will auto-save after execution + use_snapshot: Snapshot ID to load before running this case + """ model_config = ConfigDict(extra='allow') name: str = 'Unnamed Case' steps: List[CaseStep] = [] + snapshot: Optional[str] = None # Non-empty = auto-save fixture + use_snapshot: Optional[str] = None # Load specific fixture @classmethod def from_yaml_list(cls, cases_list: List[Dict[str, Any]]) -> List['Case']: diff --git a/webqa_agent/executor/case_executor.py b/webqa_agent/executor/case_executor.py index b579603..6ece01b 100644 --- a/webqa_agent/executor/case_executor.py +++ b/webqa_agent/executor/case_executor.py @@ -57,6 +57,10 @@ async def execute_cases( """Execute all cases using worker pool pattern (unified serial/parallel). + Execution flow: + 1. Cases with non-empty snapshot run serially first and auto-save browser state + 2. Remaining cases run concurrently after pre-setup-cases complete + Args: cases: List of case configurations from YAML workers: Number of parallel workers (1 = serial, >1 = parallel) @@ -68,17 +72,105 @@ async def execute_cases( mode_str = f'parallel ({workers} workers)' if workers > 1 else 'serial' logging.info(f"{icon['rocket']} Starting {mode_str} execution: {total_cases} cases") + # Phase 1: Separate cases by snapshot + fixture_cases = [ + (idx, case) for idx, case in enumerate(cases, 1) + if case.get('snapshot') # Non-empty snapshot = fixture case + ] + normal_cases = [ + (idx, case) for idx, case in enumerate(cases, 1) + if not case.get('snapshot') # No snapshot = normal case + ] + + if fixture_cases: + logging.info(f"{icon['lock']} Found {len(fixture_cases)} fixture cases (will run serially first)") + if normal_cases: + logging.info(f"{icon['rocket']} Found {len(normal_cases)} normal cases (will run concurrently after)") + # Create session pool (auto-initialized) session_pool = BrowserSessionPool(pool_size=workers, browser_config=self.browser_config) # Shared state - case_queue: asyncio.Queue = asyncio.Queue() results: List[SubTestResult] = [] results_lock = asyncio.Lock() completed_count = 0 - # Fill queue - for idx, case in enumerate(cases, 1): + # Phase 2: Execute fixture cases serially + for idx, case in fixture_cases: + case_name = case.get('name', f'Case {idx}') + case_id = case.get('case_id', f'case_{idx}') + browser_cfg = case.get('_config', {}).get('browser_config', self.browser_config) + # Set test_id context for logging + log_context = f'Run Case Test | {case_id} | {case_name}' + token = test_id_var.set(log_context) + session = None + case_result = None + + try: + logging.info(f"{icon['lock']} Starting fixture case: '{case_name}' ({completed_count + 1}/{total_cases})") + session = await session_pool.acquire(browser_config=browser_cfg, timeout=120.0) + + with Display.display(case_name): # pylint: disable=not-callable + case_result = await self.execute_single_case(session=session, case=case, case_index=idx) + + async with results_lock: + results.append(case_result) + completed_count += 1 + + status_icon = icon['check'] if case_result.status == TestStatus.PASSED else icon['cross'] + logging.info(f"{status_icon} Fixture case '{case_name}' - {case_result.status} ({completed_count}/{total_cases})") + + # Auto-save snapshot after execution (using PersistentContextManager directly) + snapshot_id = case.get('snapshot') + if snapshot_id: + try: + from webqa_agent.browser.context_manager import PersistentContextManager + await PersistentContextManager.save_storage_state( + session.context, + snapshot_id, + base_dir=self._get_snapshot_dir() + ) + logging.info(f"{icon['check']} Saved persistent snapshot to '{snapshot_id}' for '{case_name}'") + except Exception as e: + logging.warning(f"Failed to save snapshot for '{case_name}': {e}") + + except Exception as e: + logging.error(f"Exception in fixture case '{case_name}': {e}", exc_info=True) + async with results_lock: + completed_count += 1 + results.append(SubTestResult( + name=case_name, + status=TestStatus.FAILED, + metrics={'total_steps': 0, 'passed_steps': 0, 'failed_steps': 0}, + steps=[], + messages={}, + start_time=datetime.now().isoformat(), + end_time=datetime.now().isoformat(), + final_summary=f'Exception: {str(e)}', + report=[], + )) + + finally: + # Reset test_id context + test_id_var.reset(token) + if case_result is not None: + case_config = case.get('_config', {}) + self._save_case_result(case_result, case_name, idx, case_config=case_config) + self._clear_case_screenshots(case_result) + if session: + failed = case_result is None or case_result.status == TestStatus.FAILED + await session_pool.release(session, failed=failed) + + # Phase 3: Execute normal cases concurrently + if not normal_cases: + logging.info(f"{icon['check']} All fixture cases completed. No normal cases to execute.") + await session_pool.close_all() + return results + logging.info(f"{icon['rocket']} Starting concurrent execution of {len(normal_cases)} normal cases") + + # Fill queue with normal cases + case_queue: asyncio.Queue = asyncio.Queue() # Queue for normal cases + for idx, case in normal_cases: await case_queue.put((idx, case)) async def worker(worker_id: int): @@ -194,6 +286,10 @@ async def execute_single_case(self, session: BrowserSession, case: Dict[str, Any # Initialize tester and execute steps tester = await self._initialize_tester(session, case_name, url=url, cookies=cookies, ignore_rules=ignore_rules) + # Load fixture state AFTER navigation (cookies + localStorage + sessionStorage) + if case.get('use_snapshot'): + await self._load_fixture_state(session, case, case_config) + # Execute steps executed_steps, case_status, error_messages, prev_step_context = await self._execute_steps( tester, case.get('steps', []) @@ -220,6 +316,79 @@ async def execute_single_case(self, session: BrowserSession, case: Dict[str, Any # Private Methods - Tester Lifecycle # ======================================================================== + def _get_snapshot_dir(self) -> str: + """Get snapshot base directory within report directory. + + Returns: + str: Snapshot base directory path ({report_dir}/snapshots/) + """ + return os.path.join(self.report_dir, 'snapshots') + + async def _load_fixture_state( + self, + session: BrowserSession, + case: Dict[str, Any], + case_config: Dict[str, Any] + ) -> None: + """Load fixture state (cookies + localStorage + sessionStorage) after page navigation. + + Args: + session: Browser session + case: Case configuration with use_snapshot field + case_config: Case-specific configuration + """ + snapshot_name = case.get('use_snapshot') + if not snapshot_name: + return + + try: + from webqa_agent.browser.context_manager import PersistentContextManager + storage_path = await PersistentContextManager.get_storage_state_path(snapshot_name, self._get_snapshot_dir()) + if not storage_path: + logging.warning(f"Snapshot '{snapshot_name}' not found, case will run without pre-loaded state") + return + # load storage_state JSON, including cookies and origins + with open(storage_path, 'r', encoding='utf-8') as f: + storage_state = json.load(f) + + page = session.page + current_url = page.url + + # load cookies + if 'cookies' in storage_state and storage_state['cookies']: + await session.context.add_cookies(storage_state['cookies']) + logging.info(f"Loaded {len(storage_state['cookies'])} cookies from snapshot '{snapshot_name}'") + + # load localStorage and sessionStorage + if 'origins' in storage_state: + for origin_data in storage_state['origins']: + origin = origin_data.get('origin') + if not origin or not current_url.startswith(origin): + continue + + # Inject localStorage + if 'localStorage' in origin_data: + for item in origin_data['localStorage']: + await page.evaluate( + f"window.localStorage.setItem({json.dumps(item['name'])}, {json.dumps(item['value'])})" + ) + logging.debug(f"Injected {len(origin_data['localStorage'])} localStorage items") + + # Inject sessionStorage + if 'sessionStorage' in origin_data: + for item in origin_data['sessionStorage']: + await page.evaluate( + f"window.sessionStorage.setItem({json.dumps(item['name'])}, {json.dumps(item['value'])})" + ) + logging.debug(f"Injected {len(origin_data['sessionStorage'])} sessionStorage items") + + # Reload page to apply storage + await page.reload(wait_until='domcontentloaded') + logging.info(f"Reloaded page to apply snapshot '{snapshot_name}' state") + + except Exception as e: + logging.warning(f"Failed to load snapshot '{snapshot_name}': {e}") + async def _initialize_tester( self, session: BrowserSession, diff --git a/webqa_agent/utils/log_icon.py b/webqa_agent/utils/log_icon.py index f02e1a8..4295325 100644 --- a/webqa_agent/utils/log_icon.py +++ b/webqa_agent/utils/log_icon.py @@ -23,5 +23,4 @@ 'sleep': '💤', 'lightbulb': '💡', 'hourglass': '⏳', - 'repeat': '🔁', }