Cache unavailable packages so we don't ask for

password from user everytime node runs.

Author: sazid

Framework/Built_In_Automation/Web/Selenium/linux_system.py

@@ -1,12 +1,16 @@
 import os
+import re
 import shutil
 import subprocess
 from pathlib import Path
 
 
 class LinuxSystemHelper:
-    def __init__(self):
+    def __init__(self, unavailable_cache_file=None):
         self.os_release = self._load_os_release()
+        self.unavailable_cache_file = (
+            Path(unavailable_cache_file) if unavailable_cache_file else None
+        )
 
     def _load_os_release(self):
         data = {}
@@ -179,3 +183,59 @@ def get_missing_packages(self, packages):
         return [
             package for package in packages if not self._is_package_installed(package)
         ]
+
+    def _load_unavailable_packages(self):
+        if not self.unavailable_cache_file or not self.unavailable_cache_file.exists():
+            return set()
+
+        try:
+            with open(self.unavailable_cache_file, "r") as f:
+                return {line.strip() for line in f if line.strip()}
+        except Exception:
+            return set()
+
+    def _save_unavailable_packages(self, packages):
+        if not self.unavailable_cache_file:
+            return
+
+        try:
+            self.unavailable_cache_file.parent.mkdir(parents=True, exist_ok=True)
+            with open(self.unavailable_cache_file, "w") as f:
+                for package in sorted(packages):
+                    f.write(f"{package}\n")
+        except Exception:
+            return
+
+    def add_unavailable_packages(self, packages):
+        package_set = {pkg for pkg in packages if pkg}
+        if not package_set:
+            return set()
+
+        existing = self._load_unavailable_packages()
+        updated = existing | package_set
+        newly_added = updated - existing
+        if newly_added:
+            self._save_unavailable_packages(updated)
+        return newly_added
+
+    def filter_cached_unavailable_packages(self, packages):
+        cached = self._load_unavailable_packages()
+        allowed = [package for package in packages if package not in cached]
+        skipped = [package for package in packages if package in cached]
+        return allowed, skipped
+
+    def get_cached_unavailable_packages(self):
+        return sorted(self._load_unavailable_packages())
+
+    def extract_unavailable_packages_from_apt_output(self, output):
+        patterns = [
+            r"E:\s+Package '([^']+)' has no installation candidate",
+            r"E:\s+Unable to locate package\s+(\S+)",
+            r"Package\s+(\S+)\s+is not available, but is referred to by another package",
+        ]
+
+        unavailable = set()
+        for pattern in patterns:
+            unavailable.update(re.findall(pattern, output))
+
+        return sorted(unavailable)

Framework/Built_In_Automation/Web/Selenium/utils.py

@@ -30,6 +30,7 @@ class ChromeForTesting:
     CHROME_BASE_DIR = ZEUZ_NODE_DOWNLOADS_DIR / "chrome_for_testing"
     CHROME_VERSIONS_DIR = CHROME_BASE_DIR / "versions"
     CHROME_INFO_FILE = CHROME_BASE_DIR / "info.json"
+    CHROME_LINUX_UNAVAILABLE_DEPS_FILE = CHROME_BASE_DIR / "unavailable_linux_deps.txt"
 
     def __init__(self):
         self.system = platform.system().lower()
@@ -41,7 +42,9 @@ def __init__(self):
             self.platform_key = "mac-arm64" if self.arch == "arm64" else "mac-x64"
         elif self.system == "linux":
             self.platform_key = "linux64"
-            self.linux_helper = LinuxSystemHelper()
+            self.linux_helper = LinuxSystemHelper(
+                unavailable_cache_file=self.CHROME_LINUX_UNAVAILABLE_DEPS_FILE
+            )
             self._install_linux_dependencies()
         else:
             raise OSError(f"Unsupported platform: {self.system}/{self.arch}")
@@ -73,18 +76,94 @@ def _install_linux_dependencies(self):
                     )
                     return
 
+                install_deps, cached_unavailable = (
+                    self.linux_helper.filter_cached_unavailable_packages(missing_deps)
+                )
+                if cached_unavailable:
+                    print(
+                        "Warning: Skipping previously unavailable packages: "
+                        f"{', '.join(cached_unavailable)}"
+                    )
+
+                if not install_deps:
+                    print(
+                        "All missing dependencies are marked unavailable. "
+                        "Skipping apt install."
+                    )
+                    return
+
                 privilege_cmd, privilege_mode = (
                     self.linux_helper.get_privilege_escalation_command()
                 )
                 print(
                     f"Installing Chrome dependencies for Ubuntu using {privilege_mode}..."
                 )
                 subprocess.run(privilege_cmd + ["apt-get", "update", "-qq"], check=True)
-                subprocess.run(
-                    privilege_cmd + ["apt-get", "install", "-y"] + missing_deps,
-                    check=True,
+                install_result = subprocess.run(
+                    privilege_cmd + ["apt-get", "install", "-y"] + install_deps,
+                    capture_output=True,
+                    text=True,
+                    check=False,
                 )
-                print("Dependencies installed successfully.")
+                install_ok = install_result.returncode == 0
+
+                if install_result.returncode != 0:
+                    apt_output = f"{install_result.stdout}\n{install_result.stderr}"
+                    unavailable_from_apt = (
+                        self.linux_helper.extract_unavailable_packages_from_apt_output(
+                            apt_output
+                        )
+                    )
+
+                    if unavailable_from_apt:
+                        newly_cached = self.linux_helper.add_unavailable_packages(
+                            unavailable_from_apt
+                        )
+                        if newly_cached:
+                            print(
+                                "Warning: Caching unavailable packages for future runs: "
+                                f"{', '.join(sorted(newly_cached))}"
+                            )
+
+                        retry_deps = [
+                            package
+                            for package in install_deps
+                            if package not in unavailable_from_apt
+                        ]
+
+                        if retry_deps:
+                            retry_result = subprocess.run(
+                                privilege_cmd
+                                + ["apt-get", "install", "-y"]
+                                + retry_deps,
+                                capture_output=True,
+                                text=True,
+                                check=False,
+                            )
+                            if retry_result.returncode != 0:
+                                print("Warning: Could not install all dependencies.")
+                                print(
+                                    retry_result.stderr.strip()
+                                    or retry_result.stdout.strip()
+                                )
+                            else:
+                                install_ok = True
+                        else:
+                            print(
+                                "Warning: All attempted packages were unavailable in apt. "
+                                "Continuing without interruption."
+                            )
+                    else:
+                        print("Warning: Could not install dependencies.")
+                        print(
+                            install_result.stderr.strip()
+                            or install_result.stdout.strip()
+                        )
+
+                if install_ok:
+                    print("Dependencies installed successfully.")
+                else:
+                    print("Dependency installation completed with warnings.")
             else:
                 return