Making big modules 'optional' for Docker

framework/core/__init__.py

@@ -32,13 +32,31 @@
 
 #__all__ = ["testControl", "logModule", "rcCodes", "rackController", "slotInfo" ]
 #expose only the components required for the upper classes to operate
-from . capture import capture
+try:
+  from . capture import capture
+except ModuleNotFoundError as e:
+  # Only treat known optional image-processing dependencies as optional.
+  _optional_capture_deps = {"cv2", "pytesseract", "PIL"}
+  if e.name in _optional_capture_deps:
+    class _CaptureMissingDependency:
+      def __call__(self, *args, **kwargs):
+        raise ImportError(
+          "Image capture functionality is unavailable because optional "
+          "dependencies (cv2, pytesseract, PIL) are not installed."
+        ) from e
+    capture = _CaptureMissingDependency()
+  else:
+    # Unexpected missing module (e.g., bug inside capture.py) - do not hide it.
+    raise
 from . commonRemote import commonRemoteClass
 from . deviceManager import deviceManager
 from . logModule import logModule
 from . logModule import DEBUG, INFO, WARNING, ERROR, CRITICAL
 from . testControl import testController
-from . webpageController import webpageController
+try:
+  from . webpageController import webpageController
+except ModuleNotFoundError:
+  webpageController = None  # selenium not installed (e.g. Docker)
 from . rcCodes import rcCode as rc
 from . utilities import utilities
 

framework/core/commandModules/serialClass.py

@@ -200,5 +200,5 @@ def flush(self) -> bool:
             True if can successfully clear the serial console.
         """
         self.log.info("Clearing Serial console log")
-        self.serialCon.reset_input_buffer()
+        self.serialCon.flushInput()
         return True

framework/core/commandModules/telnetClass.py

@@ -30,7 +30,76 @@
 #*
 #* ******************************************************************************
 
-import telnetlib
+try:
+    import telnetlib
+except ModuleNotFoundError:
+    # telnetlib was removed in Python 3.13, provide a minimal compatibility layer
+    import socket
+    import time
+
+    class Telnet:
+        def __init__(self, host=None, port=23, timeout=socket._GLOBAL_DEFAULT_TIMEOUT):
+            self.host = host
+            self.port = port
+            self.timeout = timeout
+            self.sock = None
+            if host:
+                self.open(host, port, timeout)
+
+        def open(self, host, port=23, timeout=socket._GLOBAL_DEFAULT_TIMEOUT):
+            self.host = host
+            self.port = port
+            self.sock = socket.create_connection((host, port), timeout)
+
+        def close(self):
+            if self.sock:
+                self.sock.close()
+                self.sock = None
+
+        def write(self, buffer):
+            if self.sock:
+                self.sock.send(buffer)
+
+        def read_until(self, match, timeout=None):
+            if not self.sock:
+                return b''
+
+            buffer = b''
+            start_time = time.time()
+            while True:
+                if timeout and (time.time() - start_time) > timeout:
+                    break
+                try:
+                    data = self.sock.recv(1024)
+                    if not data:
+                        break
+                    buffer += data
+                    if match in buffer:
+                        break
+                except socket.timeout:
+                    break
+                except Exception:
+                    break
+            return buffer
+
+        def read_all(self):
+            if not self.sock:
+                return b''
+            buffer = b''
+            try:
+                while True:
+                    data = self.sock.recv(1024)
+                    if not data:
+                        break
+                    buffer += data
+            except Exception:
+                pass
+            return buffer
+
+    # Create a mock telnetlib module
+    class telnetlib:
+        Telnet = Telnet
+
 import socket
 
 from .consoleInterface import consoleInterface
@@ -153,15 +222,15 @@ def read_until(self,value: str, timeout: int = 10) -> str:
         message = value.encode()
         result = self.tn.read_until(message,self.timeout)
         return result.decode()
-    
+
     def read_all(self) -> str:
         """Read all readily available information displayed in the console.
 
         Returns:
             str: Information currently displayed in the console.
         """
         return self.read_eager()
-    
+
     def write(self,message:list|str, lineFeed:str="\r\n", wait_for_prompt:bool=False) -> bool:
         """Write a message into the session console.
         Optional: waits for prompt.

framework/core/deviceManager.py

@@ -67,6 +67,11 @@ def __init__(self, log:logModule, logPath:str, configElements:dict):
         """
         for element in configElements:
             config = configElements.get(element)
+        # Respect 'enabled: false' in console config — skip disabled consoles
+        if config.get("enabled") is False:
+            self.type = None
+            self.session = None
+            return
         self.type = config.get("type")
         self.prompt = config.get("prompt")
         # Create a new console since it hasn't been created
@@ -172,7 +177,9 @@ def __init__(self, log:logModule, logPath:str, devices:dict):
             consoles = device.get("consoles")
             for element in consoles:
                 for name in element:
-                    self.consoles[name] = consoleClass(log, logPath, element )
+                    console = consoleClass(log, logPath, element)
+                    if console.type is not None:  # skip disabled consoles
+                        self.consoles[name] = console
             config = device.get("outbound")
             if config != None:
                 self.outBoundClient = outboundClientClass(log, **config)
@@ -218,9 +225,15 @@ def getConsoleSession(self, consoleName:str="default" ):
         Returns:
             consoleClass: Console class, or None on failure
         """
-        console = self.consoles[consoleName]
-        if console == None:
-            self.log.error("Invalid consoleName [{}]".format(consoleName))
+        console = self.consoles.get(consoleName)
+        # If requested console was disabled/missing, fall back to first enabled console
+        if console is None and self.consoles:
+            fallback = next(iter(self.consoles))
+            self.log.info("Console '{}' not available, falling back to '{}'".format(consoleName, fallback))
+            console = self.consoles[fallback]
+        if console is None:
+            self.log.error("No consoles available (all disabled?)")
+            return None
         return console.session
 
     def pingTest(self, logPingTime=False):

framework/core/logModule.py

@@ -118,7 +118,7 @@ def __init__(self, moduleName, level=INFO):
     def __del__(self):
         """Deletes the logger instance.
         """
-        while self.log.hasHandlers():
+        while self.log.handlers:
             self.log.removeHandler(self.log.handlers[0])
 
     def setFilename( self, logPath, logFileName ):
@@ -134,11 +134,11 @@ def setFilename( self, logPath, logFileName ):
             return
         self.logPath = logPath
         logFileName = os.path.join(logPath + logFileName)
-        self.logFile = logging.FileHandler(logFileName)
+        self.logFile = logging.FileHandler(logFileName, encoding='utf-8')
         self.logFile.setFormatter( self.format )
         self.log.addHandler( self.logFile )
         #Create the CSV Logger module
-        self.csvLogFile = logging.FileHandler( logFileName+".csv" )
+        self.csvLogFile = logging.FileHandler( logFileName+".csv", encoding='utf-8' )
         self.csvLogger.addHandler( self.csvLogFile )
         self.csvLogger.info("QcId, TestName, Result, Failed Step, Failure, Duration [hh:mm:ss]")
         self.log.info( "Log File: [{}]".format(logFileName) )

framework/core/testControl.py

@@ -33,7 +33,11 @@
 import time
 import signal
 import random
-import telnetlib
+try:
+    import telnetlib
+except ModuleNotFoundError:
+    # telnetlib was removed in Python 3.13, this is handled in telnetClass.py
+    telnetlib = None
 import os
 import shlex, subprocess
 import traceback
@@ -52,8 +56,22 @@
 from framework.core.configParser import configParser
 from framework.core.utilities import utilities
 from framework.core.decodeParams import decodeParams
-from framework.core.capture import capture
-from framework.core.webpageController import webpageController
+try:
+  from framework.core.capture import capture
+except ModuleNotFoundError as e:
+    # cv2/pytesseract/PIL are optional dependencies (e.g. in Docker images)
+    if e.name in ("cv2", "pytesseract", "PIL", "PIL.Image"):
+        capture = None
+    else:
+        raise
+try:
+  from framework.core.webpageController import webpageController
+except ModuleNotFoundError as e:
+    # selenium is an optional dependency (e.g. in Docker images)
+    if e.name == "selenium":
+        webpageController = None
+    else:
+        raise
 from framework.core.deviceManager import deviceManager
 
 class testController():
@@ -149,9 +167,30 @@ def __init__(self, testName="", qcId="", maxRunTime=TEST_MAX_RUN_TIME, level=log
         #Start the rest of the testControl requirements
         self.devices = deviceManager(self.slotInfo.config.get("devices"), self.log, self.testLogPath)
 
-        # Set up the session from the default console
+        # Set up the session from the enabled console (not just "default")
         self.dut = self.devices.getDevice( "dut" )
-        self.session = self.dut.getConsoleSession()
+
+        # CRITICAL: Select console based on which is enabled, not hardcoded "default"
+        # Check console configuration to find the enabled one
+        console_name = "default"  # Fallback to default
+        try:
+            # Get the raw config to check enabled status
+            dut_config = self.slotInfo.config.get("devices", [{}])[0].get("dut", {})
+            consoles_config = dut_config.get("consoles", [])
+
+            # Find the enabled console
+            for console_item in consoles_config:
+                for name, config in console_item.items():
+                    if config.get("enabled", False):
+                        console_name = name
+                        self.log.info(f"Using enabled console: {console_name} (type: {config.get('type')})")
+                        break
+                if console_name != "default":
+                    break
+        except Exception as e:
+            self.log.warn(f"Failed to detect enabled console, using default: {e}")
+
+        self.session = self.dut.getConsoleSession(console_name)
         self.outboundClient = self.dut.outBoundClient
         self.powerControl = self.dut.powerControl
         self.commonRemote = self.dut.remoteController