Add Playwright-based YouTube UI tests using Optics

optics_framework/api/action_keyword.py

@@ -498,7 +498,8 @@ def get_text(self, element: str) -> Optional[str]:
             else:
                 internal_logger.error(
                     'Get Text is not supported for vision based search yet.')
-                return None
+                result = self.element_source.locate(element)
+                return self.driver.get_text_element(result)
         else:
             internal_logger.error(
                 'Get Text is not supported for image based search yet.')

optics_framework/common/async_utils.py

@@ -0,0 +1,69 @@
+import asyncio
+import threading
+from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeoutError
+from typing import Any, Coroutine
+from optics_framework.common.logging_config import internal_logger
+from optics_framework.common.error import OpticsError, Code
+
+_persistent_loop: asyncio.AbstractEventLoop | None = None
+_loop_thread: threading.Thread | None = None
+_loop_lock = threading.Lock()
+
+# Shared executor to avoid thread churn
+_executor = ThreadPoolExecutor(max_workers=1)
+
+
+def _start_loop(loop: asyncio.AbstractEventLoop):
+    asyncio.set_event_loop(loop)
+    loop.run_forever()
+
+
+def _get_or_create_persistent_loop() -> asyncio.AbstractEventLoop:
+    global _persistent_loop, _loop_thread
+
+    with _loop_lock:
+        if _persistent_loop is None or _persistent_loop.is_closed():
+            internal_logger.info("[AsyncUtils] Creating persistent event loop")
+
+            _persistent_loop = asyncio.new_event_loop()
+            _loop_thread = threading.Thread(
+                target=_start_loop,
+                args=(_persistent_loop,),
+                daemon=True,
+                name="optics-async-loop"
+            )
+            _loop_thread.start()
+
+    return _persistent_loop
+
+
+def run_async(coro: Coroutine[Any, Any, Any]):
+    """
+    Safely run async coroutine from sync code.
+    - Works with pytest / nested loops
+    - Avoids deadlocks
+    - Stable for Playwright
+    """
+
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        pass
+
+    # 🔹 Always use persistent background loop to avoid deadlocks when called from async context
+    # When called from FastAPI/async context, using the running loop causes deadlocks because
+    # we're blocking synchronously while waiting for a coroutine scheduled on the same loop
+    loop = _get_or_create_persistent_loop()
+    future = asyncio.run_coroutine_threadsafe(coro, loop)
+    try:
+        return future.result(timeout=120)  # Increased timeout for browser operations
+    except (TimeoutError, FutureTimeoutError) as e:
+        # Cancel the coroutine if it's still running to prevent it from continuing
+        if not future.done():
+            future.cancel()
+        raise OpticsError(Code.E0102, f"Async operation timed out after 120 seconds: {str(e) or 'Operation exceeded timeout limit'}", cause=e)
+    except Exception:
+        # Cancel the coroutine if it's still running to prevent it from continuing
+        if not future.done():
+            future.cancel()
+        raise

optics_framework/common/utils.py

@@ -92,12 +92,28 @@ def determine_element_type(element):
     # Check if the input is an Image path
     if element.split(".")[-1] in ["jpg", "jpeg", "png", "bmp"]:
         return "Image"
+    # Check for Playwright-specific prefixes first
+    if element.lower().startswith("text="):
+        return "Text"
+    if element.lower().startswith("css="):
+        return "CSS"
+    if element.lower().startswith("xpath="):
+        return "XPath"
     # Check if the input is an XPath
     if element.startswith("/") or element.startswith("//") or element.startswith("("):
         return "XPath"
     # Check if it looks like an ID (heuristic: no slashes, no dots, usually alphanumeric/underscores)
     if element.lower().startswith("id:"):
         return "ID"
+    # Check if it's a CSS selector (has brackets, starts with # or ., or contains CSS selector patterns)
+    # CSS selectors can have: tag[attribute="value"], #id, .class, tag.class, etc.
+    if ("[" in element and "]" in element) or element.startswith("#") or element.startswith("."):
+        return "CSS"
+    # Check if it starts with a tag name followed by CSS selector characters
+    # Common HTML tags that might be CSS selectors
+    common_tags = ["input", "button", "div", "span", "a", "img", "select", "textarea", "form", "label", "p", "h1", "h2", "h3", "h4", "h5", "h6"]
+    if any(element.startswith(tag + "[") or element.startswith(tag + "#") or element.startswith(tag + ".") for tag in common_tags):
+        return "CSS"
     # Default case: consider the input as Text
     return "Text"