fix: different strategy for PDF page in run mode and gen mode

Author: seancoding-day

webqa_agent/crawler/deep_crawler.py

@@ -7,6 +7,7 @@
 from itertools import groupby
 from pathlib import Path
 from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
+from urllib.parse import urlparse
 
 from playwright.async_api import Page
 from pydantic import BaseModel, Field
@@ -18,6 +19,7 @@
 # UTILITY FUNCTIONS
 # ============================================================================
 
+
 def get_time() -> str:
     """Get the current time as a formatted string.
 
@@ -224,19 +226,24 @@ async def _detect_page_type(self, page: Page) -> Tuple[str, Optional[str]]:
                 - ("UNSUPPORTED_PAGE", "download") for download files
         """
         try:
-            # === Layer 1: URL Extension Check (Fastest, Most Reliable) ===
+            # === Layer 1: URL Path Extension Check (Fastest, Most Reliable) ===
+            # IMPORTANT: Only check path component, NOT query params or fragments
+            # This correctly handles: https://example.com/page?file=test.pdf (NOT a PDF page)
+            # vs: https://example.com/documents/file.pdf (IS a PDF page)
             url = page.url.lower()
+            parsed_url = urlparse(url)
+            url_path = parsed_url.path  # Only the path: "/documents/file.pdf"
 
-            # PDF detection via URL
-            if url.endswith('.pdf'):
-                logging.info(f'Detected PDF via URL suffix: {url}')
+            # PDF detection via URL path extension
+            if url_path.endswith('.pdf'):
+                logging.info(f'Detected PDF via URL path: {url_path} (full URL: {url})')
                 return ('UNSUPPORTED_PAGE', 'pdf')
 
-            # Download file detection via URL
-            download_extensions = ['.zip', '.rar', '.exe', '.dmg', '.pkg', '.deb', '.tar', '.gz']
+            # Download file detection via URL path extension
+            download_extensions = ['.zip', '.rar', '.exe', '.dmg', '.pkg', '.deb', '.tar', '.gz', '.7z']
             for ext in download_extensions:
-                if url.endswith(ext):
-                    logging.info(f'Detected download file via URL: {url}')
+                if url_path.endswith(ext):
+                    logging.info(f'Detected download file via URL path: {url_path}')
                     return ('UNSUPPORTED_PAGE', 'download')
 
             # === Layer 2: PDF Embed Element Detection (Catches Chromium PDF Viewer) ===
@@ -312,12 +319,9 @@ async def crawl(
         # Multi-layer detection of unsupported page types (PDF, plugins, etc.)
         page_status, page_type = await self._detect_page_type(page)
         if page_status == 'UNSUPPORTED_PAGE':
-            logging.warning(f'Detected unsupported page type: {page_type}, skipping crawl')
-            return CrawlResultModel(
-                flat_element_map=ElementMap(data={}),
-                element_tree={},
-                page_status=page_status,
-                page_type=page_type
+            logging.info(
+                f'Detected unsupported page type: {page_type}, '
+                f'attempting to extract available data in degraded mode'
             )
 
         try:
@@ -327,7 +331,9 @@ async def crawl(
                     _, merged_id_map = await self.crawl_all_frames(page=page, enable_highlight=highlight)
                     return CrawlResultModel(
                         flat_element_map=ElementMap(data=merged_id_map or {}),
-                        element_tree={}
+                        element_tree={},
+                        page_status=page_status,  # Pass detected page status for multi-frame pages
+                        page_type=page_type        # Pass detected page type for multi-frame pages
                     )
             except Exception:
                 pass
@@ -350,7 +356,9 @@ async def crawl(
 
             result = CrawlResultModel(
                 flat_element_map=ElementMap(data=flat_elements or {}),
-                element_tree=self.element_tree or {}
+                element_tree=self.element_tree or {},
+                page_status=page_status,  # Pass detected page status
+                page_type=page_type        # Pass detected page type
             )
 
             if cache_dom and self.element_tree:
@@ -372,7 +380,10 @@ async def crawl(
 
         except Exception as e:
             logging.error(f'JavaScript injection failed during element detection: {e}')
-            return CrawlResultModel()
+            return CrawlResultModel(
+                page_status=page_status,  # Pass detected page status even on failure
+                page_type=page_type        # Pass detected page type even on failure
+            )
 
     def extract_interactive_elements(self, get_new_elems: bool = False) -> Dict:
         """Extract interactive elements with comprehensive attribute
@@ -541,7 +552,7 @@ async def _accumulate_iframe_offsets(f):
                     el = await cur.frame_element()
                     rect = await el.evaluate('(el) => el.getBoundingClientRect()')
                     total_left += rect.get('left', 0) or 0
-                    total_top  += rect.get('top', 0) or 0
+                    total_top += rect.get('top', 0) or 0
                 except Exception:
                     pass
                 cur = parent
@@ -579,7 +590,7 @@ async def _accumulate_iframe_offsets(f):
 
                 # Get frame URL for later action execution in correct frame context
                 frame_url = frame.url
-                
+
                 for k, v in (iframe_id_map or {}).items():
                     try:
                         # frame document -> frame viewport

webqa_agent/executor/case_executor.py

@@ -450,7 +450,8 @@ async def _initialize_tester(
         tester = UITester(
             llm_config=self.llm_config,
             browser_session=session,
-            ignore_rules=_ignore_rules
+            ignore_rules=_ignore_rules,
+            execution_mode='run'  # RUN mode: trust user-specified operations in YAML
         )
         await tester.initialize()
         tester.set_current_test_name(case_name)

webqa_agent/testers/case_gen/graph.py

@@ -61,7 +61,11 @@ async def plan_test_cases(state: MainGraphState) -> Dict[str, List[Dict[str, Any
     s = await sp.acquire(timeout=120.0)
     try:
         await s.navigate_to(state['url'], cookies=state.get('cookies'))
-        ui_tester = UITester(llm_config=llm_cfg, browser_session=s)
+        ui_tester = UITester(
+            llm_config=llm_cfg,
+            browser_session=s,
+            execution_mode='gen'  # GEN mode: conservative approach for AI exploration
+        )
         await ui_tester.initialize()
         page = await ui_tester.get_current_page()
         dp = DeepCrawler(page)
@@ -392,7 +396,11 @@ async def worker(worker_id: int):
 
                 await s.navigate_to(state['url'], cookies=state.get('cookies'))
 
-                ui_tester = UITester(llm_config=state['llm_config'], browser_session=s)
+                ui_tester = UITester(
+                    llm_config=state['llm_config'],
+                    browser_session=s,
+                    execution_mode='gen'  # GEN mode: conservative approach for AI exploration
+                )
                 await ui_tester.initialize()
 
                 # Set testcase context