Merge branch 'release/1.1.0'

CHANGELOG.md

@@ -4,6 +4,21 @@ All notable changes to this project will be documented in this file. Dates are d
 
 Generated by [`auto-changelog`](https://github.com/CookPete/auto-changelog).
 
+#### [1.1.0](https://github.com/rdkcentral/python_raft/compare/1.0.0...1.1.0)
+
+- GH90 - interactive ssh shell [`#91`](https://github.com/rdkcentral/python_raft/pull/91)
+- Fix #75: Fixed error passing port as string. [`#75`](https://github.com/rdkcentral/python_raft/issues/75)
+- Fix #72: Added missing copyright header to test_deviceManager [`#72`](https://github.com/rdkcentral/python_raft/issues/72)
+- Fix #71: Added missing args in docstring for constructTestPath in [`#71`](https://github.com/rdkcentral/python_raft/issues/71)
+- Fix #71: Fix rackController to use --rack argument correctly [`#71`](https://github.com/rdkcentral/python_raft/issues/71)
+- Fix #77: Removed S20 import from powerControl [`#77`](https://github.com/rdkcentral/python_raft/issues/77)
+- Update #71: Updates to Singleton for rackController support [`4e88c59`](https://github.com/rdkcentral/python_raft/commit/4e88c59f49a83e79b8665ad7f0a1ff048f6cf40d)
+- Update #75: Moved pingTest and added session to device class. [`5a967fe`](https://github.com/rdkcentral/python_raft/commit/5a967fedd6a5a7be2efc6d8963a82963704e66b6)
+- update test adn singleton logic [`2edf093`](https://github.com/rdkcentral/python_raft/commit/2edf09353bb0c5c80fd73ee117a09b98eaedb749)
+
 #### 1.0.0
 
+> 14 May 2024
+
 - Initial commit [`f7868c3`](https://github.com/rdkcentral/python_raft/commit/f7868c3c1ccfb768cfe91cc8d289b3a1420c6c45)
+- reset changelog [`9976d08`](https://github.com/rdkcentral/python_raft/commit/9976d083e060326fc5146d2ef68288c413be8b2a)

examples/code/example_ssh_unittest.py

@@ -0,0 +1,115 @@
+#!/usr/bin/env python3
+#** *****************************************************************************
+# *
+# * If not stated otherwise in this file or this component's LICENSE file the
+# * following copyright and licenses apply:
+# *
+# * Copyright 2023 RDK Management
+# *
+# * Licensed under the Apache License, Version 2.0 (the "License");
+# * you may not use this file except in compliance with the License.
+# * You may obtain a copy of the License at
+# *
+# *
+# http://www.apache.org/licenses/LICENSE-2.0
+# *
+# * Unless required by applicable law or agreed to in writing, software
+# * distributed under the License is distributed on an "AS IS" BASIS,
+# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# * See the License for the specific language governing permissions and
+# * limitations under the License.
+# *
+#* ******************************************************************************
+"""A simple test that does the following:
+1. SSH's into our dut (with the getting_started_rack_config.yml this is our local PC)
+2. Creates a folder called RAFT_test_files
+3. Creates a file in that folder called RAFT_Test_File
+4. Lists the contents of the RAFT_test_files directory to check that the RAFT_Test_File has been created
+5. Cleans up after itself and removes the RAFT_test_files folder
+6. Exits with test success if the file is created, failure when the file is not created.
+"""
+
+import sys
+from os import path
+import re
+
+# Since this test is in a sub-directory we need to add the directory above
+# so we can import the framework correctly
+MY_PATH = path.abspath(__file__)
+MY_DIR = path.dirname(MY_PATH)
+sys.path.append(path.join(MY_DIR,'../../'))
+from framework.core.raftUnittest import RAFTUnitTestCase, RAFTUnitTestMain
+from framework.core.singleton import SINGLETON
+from framework.core.utilities import utilities
+# Constants for the file paths we need for testing
+TEST_FILE = 'RAFT_Test_File'
+TEST_DIRECTORY = '~/RAFT_test_files'
+
+class TestExampleSSH(RAFTUnitTestCase):
+    """
+    A test class for verifying file creation on the DUT via SSH.
+    """
+
+    def _list_files(self,directory:str):
+        """List the files in the given directory.
+
+        Args:
+            directory (str): Path of directory to list files in.
+
+        Returns:
+            str: Space separated list of files from the directory.
+        """
+        # List the files in the test directory
+        self.dut.session.write(f'ls {directory}')
+        file_list = utilities.strip_ansi_escapes(self.dut.session.read_all())
+        # Clear the session buffer now we've capured its contents
+        self.dut.session.write('clear')
+        self.log.info(f'Current file list: [{file_list}]')
+        return file_list
+
+    def setUp(self):
+        """
+        Perform pre-test setup tasks.
+
+        This method will run before each test method.
+        """
+        self.log.info('Pre-test check')
+        self.log.info('Make the test directory')
+        # Ping Test to check Box alive
+        if self.dut.pingTest() is False:
+            raise ConnectionError('Cannot reach dut')
+        # Open the console session
+        self.dut.session.open()
+        # Create test directory if if doesn't already exist
+        self.dut.session.write(f'mkdir -p {TEST_DIRECTORY}')
+        file_list = self._list_files(TEST_DIRECTORY)
+        if TEST_FILE in file_list:
+            self.log.info(f'{TEST_FILE} found in testing directory')
+            self.dut.session.write(f'rm {TEST_DIRECTORY}/{TEST_FILE}')
+            self.log.info('Rerunning pre-test check to ensure file has been removed')
+
+    def test_fileCreation(self):
+        """
+        The main test method that verifies file creation on the DUT
+
+        Any method starting with the word test at the start of it name,
+        will run as an individual test method.
+        """
+        self.log.stepStart(f'Creating {TEST_FILE} in {TEST_DIRECTORY}')
+        # Create the test file in the test directory
+        self.dut.session.write(f'touch {TEST_DIRECTORY}/{TEST_FILE}')
+        file_list = self._list_files(TEST_DIRECTORY)
+        # Test that the file is in the list object
+        self.assertIn(TEST_FILE, file_list)
+
+    def tearDown(self):
+        """
+        Performs post-test cleanup tasks:
+
+        This method will run after each test method.
+        """
+        self.log.info(f'Removing {TEST_DIRECTORY}')
+        self.dut.session.write(f'rm -rf {TEST_DIRECTORY}')
+
+if __name__ == '__main__':
+    RAFTUnitTestMain()

framework/core/commandModules/sshConsole.py

@@ -29,6 +29,7 @@
 #*   **
 #/* ******************************************************************************
 
+import time
 from paramiko import SSHClient
 
 
@@ -46,62 +47,109 @@ class sshConsole(consoleInterface):
         known_hosts (str, optional): Filepath of known_hosts file to use.
     """
 
-    def __init__(self, address, username, password, key=None, known_hosts=None) -> None:
+    def __init__(self, address, username, password, key=None, known_hosts=None, port=22) -> None:
         self.address = address
         self.username = username
         self.password = password
         self.key = key
+        self.port = port
         self.console = SSHClient()
         self.console.load_system_host_keys(known_hosts)
         self.buffer = []
         self.stdout = None
         self.type="ssh"
+        self.shell = None
+        self.full_output = ""
 
     def open(self):
         """Open the SSH session.
         """
-        self.console.connect(self.address, username = self.username, password = self.password, key_filename=self.key)
+        self.console.connect(self.address, username = self.username, password = self.password, key_filename=self.key, port=self.port)
+
+    def open_interactive_shell(self):
+        """Open an interactive shell session."""
+        # Open an interactive shell
+        self.shell = self.console.invoke_shell()
+
+        # Ensure the shell is ready
+        while not self.shell.send_ready():
+            time.sleep(1)
 
     def write(self, message):
-        """Write a message into the console.
+        """Write a message in the interactive shell.
 
         Args:
             message (str): String to write into the console.
         """
-        if self.stdout:
-            self.buffer.extend(self.stdout.readlines())
-        self.stdin, self.stdout, self.stderr = self.console.exec_command(message, get_pty=True)
+
+        if self.shell is None:
+            self.open_interactive_shell()
+
+        self.shell.send(message + '\n')
+
+        output = self.read()
+        self.full_output += output
 
+        return output
+
+    def read(self, timeout=10):
+        """Read the output from the shell with a timeout.
+
+        Args:
+            timeout (int): The maximum time to wait for in the message in seconds. Defaults to 10. 
+        """
+        output = ""
+
+        # Set a timeout period
+        end_time = time.time() + timeout
+
+        while time.time() < end_time:
+            if self.shell.recv_ready():
+                output += self.shell.recv(4096).decode('utf-8')
+                # Reset the timeout when new data arrives
+                end_time = time.time() + timeout
+            else:
+                # Small delay to prevent busy waiting
+                time.sleep(0.1)
+
+        return output
+
     def read_all(self):
-        """Capture all lines that are displayed in the console.
+        """Retrieve the accumulated output in the console.
 
         Returns:
-            List: List of strings, with each being a line displayed in the console.
+            Str: A single string containing all the accumulated output in the console.
         """
-        data = ""
-        self.buffer.extend(self.stdout.readlines())
-        self.stdout = None
-        while self.buffer.__len__() > 0:
-            data = data + self.buffer.pop(0)
-        return data
+        return self.full_output
 
-    def read_until(self, value):
-        """Read the console until a message appears.
+    def read_until(self, value, timeout=10):
+        """Read the console output until a specific message appears.
 
         Args:
             value (str): The message to wait for in the console.
+            timeout (int): The maximum time to wait for in the message in seconds. Defaults to 10. 
 
         Returns:
-            List: List of strings, with each being a line displayed in the console up to the value entered.
+            Str: The console output up to the specified value.
+
         """
-        data = ""
-        self.buffer.extend(self.stdout.readlines())
-        self.stdout = None
-        while self.buffer.__len__() > 0:
-            data = data + self.buffer.pop(0)
-            if value in data:
-                break
-        return data
+        output = ""
+        end_time = time.time() + timeout
+
+        while time.time() < end_time:
+            if self.shell.recv_ready():
+                output += self.shell.recv(4096).decode('utf-8')
+                # Reset the timeout if new data is received
+                end_time = time.time() + timeout
+
+                # Check if the target value is in the current output
+                if value in output:
+                    break
+            else:
+                # Small delay to prevent busy waiting
+                time.sleep(0.1)
+
+        return output
 
     def close(self):
         """Close the SSH session