tmp

Author: mahbd

.gitignore

@@ -61,4 +61,5 @@ custom_profiles/
 /tools/security/*
 Apps/Windows/inspector.exe
 Apps/Windows/Element.xml
-Framework/settings.conf.lock
+Framework/settings.conf.lock
+Framework/Built_In_Automation/Desktop/Linux/latest_app.txt

Framework/Built_In_Automation/Desktop/Linux/BuiltInFunctions.py

@@ -101,6 +101,47 @@ def queryValue(self) -> Value: ...
 
 MODULE_NAME = inspect.getmodulename(__file__) or "BuiltInFunctions"
 ui_xml_strings = [] # needed for generating XML tree
+LATEST_APP_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "latest_app.txt")
+
+
+def save_latest_app_name(app_name: str):
+    """Save the latest used application name to a file."""
+    try:
+        with open(LATEST_APP_FILE, "w") as f:
+            f.write(app_name)
+    except Exception as e:
+        CommonUtil.ExecLog(MODULE_NAME, f"Failed to save latest app name: {e}", 3)
+
+
+def get_latest_app_name() -> str | None:
+    """Get the latest used application name from the file."""
+    try:
+        if os.path.exists(LATEST_APP_FILE):
+            with open(LATEST_APP_FILE, "r") as f:
+                return f.read().strip()
+    except Exception as e:
+        CommonUtil.ExecLog(MODULE_NAME, f"Failed to get latest app name: {e}", 3)
+    return None
+
+
+def capture_screenshot(file_path: str) -> bool:
+    """Capture screenshot using available tools (scrot, gnome-screenshot, import)."""
+    tools = [
+        ["scrot", file_path],
+        ["gnome-screenshot", "-f", file_path],
+        ["import", "-window", "root", file_path]
+    ]
+
+    for cmd in tools:
+        try:
+            subprocess.run(cmd, check=True, capture_output=True)
+            if os.path.exists(file_path) and os.path.getsize(file_path) > 0:
+                return True
+        except (subprocess.CalledProcessError, FileNotFoundError):
+            continue
+    
+    CommonUtil.ExecLog(MODULE_NAME, "Failed to capture screenshot. Ensure scrot, gnome-screenshot, or imagemagick is installed.", 3)
+    return False
 
 
 def convert_data_set_to_dict(data_set: DataSet) -> dict[str, str]:
@@ -597,6 +638,7 @@ def open_app(data_set: DataSet) -> Literal["passed", "zeuz_failed"]:
             exit_code = os.system(command)
             if exit_code == 0:
                 CommonUtil.ExecLog(sModuleInfo, f"Successfully launched '{app_name}' with command: {command}", 1)
+                save_latest_app_name(app_name)
                 return "passed"
             else:
                 CommonUtil.ExecLog(sModuleInfo, f"Failed to launch '{app_name}' (exit code: {exit_code})", 3)
@@ -611,6 +653,7 @@ def open_app(data_set: DataSet) -> Literal["passed", "zeuz_failed"]:
             exit_code = os.system(command)
             if exit_code == 0:
                 CommonUtil.ExecLog(sModuleInfo, f"Successfully launched '{app_name}' with command: {command}", 1)
+                save_latest_app_name(app_name)
                 return "passed"
             else:
                 CommonUtil.ExecLog(sModuleInfo, f"Failed to launch '{app_name}' (exit code: {exit_code})", 3)

Framework/nodejs_appium_installer.py

@@ -4,6 +4,7 @@
 import subprocess
 import tarfile
 import zipfile
+import shutil
 from pathlib import Path
 import json
 import requests
@@ -99,7 +100,7 @@ def install_nodejs():
     print("Downloading Node.js...")
     response = requests.get(url, verify=False)
     response.raise_for_status()
-    with open(archive_path, 'wb') as out_file:
+    with open(archive_path, "wb") as out_file:
         out_file.write(response.content)
 
     try:
@@ -196,7 +197,9 @@ def check_appium_drivers():
             text=True,
         )
         drivers_data = json.loads(result.stdout)
-        return [name for name, info in drivers_data.items() if info.get("installed", False)]
+        return [
+            name for name, info in drivers_data.items() if info.get("installed", False)
+        ]
     except:  # noqa: E722
         return []
 
@@ -213,7 +216,9 @@ def check_installations():
     if npm_path.exists():
         try:
             result = subprocess.run(
-                [str(npm_path), "list", "-g", "--json", "appium"], capture_output=True, text=True
+                [str(npm_path), "list", "-g", "--json", "appium"],
+                capture_output=True,
+                text=True,
             )
             npm_data = json.loads(result.stdout)
             appium_installed = "appium" in npm_data.get("dependencies", {})
@@ -234,9 +239,73 @@ def install_missing_drivers(missing_drivers):
     install_drivers(missing_drivers)
 
 
+def check_and_remove_global_appium():
+    """Check for and remove existing global Appium installations not managed by us."""
+    print("Checking for conflicting global Appium installations...")
+
+    # Method 1: Check using 'which appium'
+    appium_bin = shutil.which("appium")
+    if appium_bin:
+        appium_path = Path(appium_bin).resolve()
+        node_dir = get_node_dir().resolve()
+
+        try:
+            # Check if appium is within our node directory
+            appium_path.relative_to(node_dir)
+            # If it is, we are good
+        except ValueError:
+            print(f"Found conflicting Appium at {appium_path}")
+            print("Uninstalling old Appium version...")
+            try:
+                is_windows = platform.system() == "Windows"
+                subprocess.run(
+                    ["npm", "uninstall", "-g", "appium"], check=True, shell=is_windows
+                )
+                print("Successfully uninstalled conflicting Appium")
+            except Exception as e:
+                print(f"Warning: Failed to uninstall conflicting Appium: {e}")
+            return
+
+    # Method 2: Check using 'npm list -g appium' (if npm is available in system path)
+    # This catches cases where appium is installed but not in PATH
+    npm_bin = shutil.which("npm")
+    if npm_bin:
+        try:
+            # Check if this npm is ours
+            npm_path = Path(npm_bin).resolve()
+            node_dir = get_node_dir().resolve()
+            try:
+                npm_path.relative_to(node_dir)
+                # If it is our npm, skip this check as we handle our own appium
+                return
+            except ValueError:
+                pass
+
+            # Check for global appium using system npm
+            is_windows = platform.system() == "Windows"
+            result = subprocess.run(
+                ["npm", "list", "-g", "--json", "appium"],
+                capture_output=True,
+                text=True,
+                shell=is_windows,
+            )
+            npm_data = json.loads(result.stdout)
+            if "appium" in npm_data.get("dependencies", {}):
+                print("Found conflicting Appium in global npm modules")
+                print("Uninstalling old Appium version...")
+                subprocess.run(
+                    ["npm", "uninstall", "-g", "appium"], check=True, shell=is_windows
+                )
+                print("Successfully uninstalled conflicting Appium")
+        except Exception as e:
+            # Don't fail if npm check fails, just log warning
+            print(f"Warning: Failed to check/uninstall global Appium via npm: {e}")
+
+
 def setup_nodejs_appium():
     """Main setup function."""
     try:
+        check_and_remove_global_appium()
         update_path()  # Ensure Node.js is in PATH from the start
 
         print("Checking Node.js and Appium installation...")

Installer/setup_linux_inspector.sh

@@ -10,6 +10,7 @@ APT_PACKAGES=(
   libgirepository1.0-dev
   libcairo2-dev
   xdotool
+  scrot
 )
 
 DNF_PACKAGES=(
@@ -20,6 +21,7 @@ DNF_PACKAGES=(
   xdotool
   python3-devel
   cairo-gobject-devel
+  scrot
 )
 
 PACMAN_PACKAGES=(
@@ -30,6 +32,7 @@ PACMAN_PACKAGES=(
   cairo
   xdotool
   gobject-introspection
+  scrot
 )
 
 BREW_PACKAGES=(
@@ -38,6 +41,7 @@ BREW_PACKAGES=(
   cairo
   xdotool
   gobject-introspection
+  scrot
 )
 
 # Function to join array into space-separated string

server/linux.py

@@ -0,0 +1,107 @@
+import hashlib
+import os
+import base64
+import asyncio
+import requests
+from typing import Literal
+from fastapi import APIRouter
+from pydantic import BaseModel
+
+from Framework.Utilities import ConfigModule, CommonUtil
+
+try:
+    from Framework.Built_In_Automation.Desktop.Linux import BuiltInFunctions
+except ImportError:
+    BuiltInFunctions = None
+
+router = APIRouter(prefix="/linux", tags=["linux"])
+
+SCREENSHOT_PATH = "linux_screen.png"
+
+class InspectorResponse(BaseModel):
+    """Response model for the /inspector endpoint."""
+    status: Literal["ok", "error"] = "ok"
+    ui_xml: str | None = None
+    screenshot: str | None = None  # Base64 encoded image
+    error: str | None = None
+
+
+@router.get("/inspect")
+def inspect(app_name: str | None = None):
+    """Get the Linux UI DOM and screenshot."""
+    if BuiltInFunctions is None:
+        return InspectorResponse(status="error", error="Linux automation module not available")
+
+    try:
+        # Determine app name
+        target_app = app_name
+        if not target_app:
+            target_app = BuiltInFunctions.get_latest_app_name()
+        
+        if not target_app:
+             return InspectorResponse(status="error", error="No application specified and no latest app found.")
+
+        # Capture UI
+        xml_content = BuiltInFunctions.get_ui_tree(target_app)
+        if not xml_content:
+             return InspectorResponse(status="error", error=f"Failed to get UI tree for app: {target_app}")
+
+        # Capture Screenshot
+        full_screenshot_path = os.path.abspath(SCREENSHOT_PATH)
+        
+        screenshot_base64 = None
+        if BuiltInFunctions.capture_screenshot(full_screenshot_path):
+            try:
+                with open(full_screenshot_path, 'rb') as img_file:
+                    screenshot_bytes = img_file.read()
+                    screenshot_base64 = base64.b64encode(screenshot_bytes).decode('utf-8')
+            except Exception:
+                pass
+
+        return InspectorResponse(
+            status="ok",
+            ui_xml=xml_content,
+            screenshot=screenshot_base64
+        )
+    except Exception as e:
+        return InspectorResponse(
+            status="error",
+            error=str(e)
+        )
+
+
+async def upload_linux_ui_dump():
+    """Continuously upload Linux UI dump if changed."""
+    if BuiltInFunctions is None:
+        return
+
+    prev_xml_hash = ""
+    while True:
+        try:
+            target_app = BuiltInFunctions.get_latest_app_name()
+            if target_app:
+                xml_content = BuiltInFunctions.get_ui_tree(target_app)
+                if xml_content:
+                    new_xml_hash = hashlib.sha256(xml_content.encode('utf-8')).hexdigest()
+                    
+                    if prev_xml_hash != new_xml_hash:
+                        prev_xml_hash = new_xml_hash
+                        
+                        url = ConfigModule.get_config_value("Authentication", "server_address").strip() + "/node_ai_contents/"
+                        apiKey = ConfigModule.get_config_value("Authentication", "api-key").strip()
+                        
+                        res = requests.post(
+                            url,
+                            headers={"X-Api-Key": apiKey},
+                            json={
+                                "dom_linux": {"dom": xml_content},
+                                "node_id": CommonUtil.MachineInfo().getLocalUser().lower(),
+                                "app_name": target_app # Extra info might be useful
+                            }
+                        )
+                        if res.ok:
+                            CommonUtil.ExecLog("", "Linux UI dump uploaded successfully", iLogLevel=1)
+        except Exception as e:
+            CommonUtil.ExecLog("", f"Error uploading Linux UI dump: {str(e)}", iLogLevel=3)
+        
+        await asyncio.sleep(5)

server/main.py

@@ -10,6 +10,7 @@
 from server.node_operator import router as operator_router
 from server.mobile import router as mobile_router, upload_android_ui_dump
 from server.mac import router as mac_router
+from server.linux import router as linux_router
 import asyncio
 
 class EndpointFilter(logging.Filter):
@@ -40,6 +41,7 @@ def main() -> FastAPI:
     v1router.include_router(evaluator_router)
     v1router.include_router(mobile_router)
     v1router.include_router(mac_router)
+    v1router.include_router(linux_router)
     app = FastAPI()
     app.include_router(v1router)