add method to issue ssh commands

Signed-off-by: Olamide Ojo <[email protected]>

Author: olamidepeterojo

CHANGELOG.md

@@ -8,6 +8,7 @@ All notable changes to the Zowe Client Python SDK will be documented in this fil
 
 - Turning off logger at the class-constructor level [#316](https://github.com/zowe/zowe-client-python-sdk/issues/316)
 - Added support for commonly used environmental variables, like `REQUESTS_CA_BUNDLE` and `CURL_CA_BUNDLE`. [#346](https://github.com/zowe/zowe-client-python-sdk/issues/346)
+- Add method to issue SSH commands. [#253](https://github.com/zowe/zowe-client-python-sdk/issues/253)
 
 ### Bug Fixes
 

README.md

@@ -54,6 +54,7 @@ jsonschema
 pyyaml
 requests>=2.22
 urllib3
+paramiko
 ```
 
 It also has an optional dependency on the Zowe Secrets SDK for storing client secrets which can be installed with the `secrets` extra:

requirements.txt

@@ -3,6 +3,7 @@ deepmerge==1.1.0
 jsonschema==4.17.3
 PyYAML==6.0.1
 requests==2.32.0
+paramiko
 
 # Dev deps
 black
@@ -24,3 +25,4 @@ wheel
 -e ./src/zos_jobs
 -e ./src/zos_tso
 -e ./src/zosmf
+-e ./src/zos_uss

src/zos_uss/README.md

@@ -0,0 +1,19 @@
+z/OS UNIX System Services (USS) Package
+=======================================
+
+Provides APIs to interact with z/OS UNIX System Services (USS) over SSH (using z/OSMF or other SSH connections).
+
+Examples
+--------
+
+### Issue a command in the z/OS USS environment
+
+```
+from zowe.core_for_zowe_sdk import ProfileManager
+from zowe.zos_uss_for_zowe_sdk import Uss
+
+profile = ProfileManager().load(profile_name="zosmf")
+
+with Uss(profile) as uss:
+    print(uss.execute_command(command="ls -la", cwd="/u/home"))
+```

src/zos_uss/setup.py

@@ -0,0 +1,38 @@
+"""Zowe Client Python SDK.
+
+This program and the accompanying materials are made available under the terms of the
+Eclipse Public License v2.0 which accompanies this distribution, and is available at
+
+https://www.eclipse.org/legal/epl-v20.html
+
+SPDX-License-Identifier: EPL-2.0
+
+Copyright Contributors to the Zowe Project.
+"""
+
+import sys
+
+from setuptools import find_namespace_packages, setup
+
+sys.path.insert(0, "..")
+from _version import __version__
+from setup import resolve_sdk_dep
+
+setup(
+    name="zowe_zos_uss_for_zowe_sdk",
+    version=__version__,
+    description="Zowe Python SDK - z/OS UNIX System Services (USS) package",
+    long_description=open("README.md", "r").read(),
+    long_description_content_type="text/markdown",
+    url="https://github.com/zowe/zowe-client-python-sdk",
+    author="Zowe",
+    author_email="[email protected]",
+    license="EPL-2.0",
+    classifiers=[
+        "Programming Language :: Python :: 3",
+        "Programming Language :: Python :: 3.9",
+        "License :: OSI Approved :: Eclipse Public License 2.0 (EPL-2.0)",
+    ],
+    install_requires=[resolve_sdk_dep("core", "~=" + __version__)],
+    packages=find_namespace_packages(include=["zowe.*"]),
+)

src/zos_uss/zowe/zos_uss_for_zowe_sdk/__init__.py

@@ -0,0 +1,13 @@
+"""Zowe Client Python SDK.
+
+This program and the accompanying materials are made available under the terms of the
+Eclipse Public License v2.0 which accompanies this distribution, and is available at
+
+https://www.eclipse.org/legal/epl-v20.html
+
+SPDX-License-Identifier: EPL-2.0
+
+Copyright Contributors to the Zowe Project.
+"""
+
+from .uss import Uss

src/zos_uss/zowe/zos_uss_for_zowe_sdk/uss.py

@@ -0,0 +1,74 @@
+"""Zowe Client Python SDK.
+
+This program and the accompanying materials are made available under the terms of the
+Eclipse Public License v2.0 which accompanies this distribution, and is available at
+
+https://www.eclipse.org/legal/epl-v20.html
+
+SPDX-License-Identifier: EPL-2.0
+
+Copyright Contributors to the Zowe Project.
+"""
+
+from typing import Optional
+import paramiko
+from zowe.core_for_zowe_sdk import SdkApi
+
+
+class Uss(SdkApi):
+    """
+    Class to interact with Unix System Services (USS) on z/OS via SSH.
+
+    Parameters
+    ----------
+    connection : dict
+        A dictionary containing SSH connection details like hostname, username, password, and port.
+    log : bool
+        Flag to enable or disable logging.
+    """
+
+    def __init__(self, connection: dict, log: bool = True):
+        super().__init__(connection, "/zosmf/restuss", logger_name=__name__, log=log)
+        self.connection = connection
+        self.ssh_client = None
+
+    def connect(self):
+        """
+        Establish an SSH connection using the connection details.
+        """
+        self.ssh_client = paramiko.SSHClient()
+        self.ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        self.ssh_client.connect(
+            hostname=self.connection["host"],
+            username=self.connection["user"],
+            password=self.connection.get("password"),
+            port=self.connection.get("port", 22),
+        )
+
+    def disconnect(self):
+        """
+        Close the SSH connection.
+        """
+        if self.ssh_client:
+            self.ssh_client.close()
+
+    def execute_command(self, command: str, cwd: Optional[str] = None):
+        """
+        Execute a Unix command over SSH.
+
+        Parameters
+        ----------
+        command : str
+            The command to execute.
+        cwd : Optional[str]
+            The working directory for the command.
+
+        Returns
+        -------
+        tuple
+            A tuple of (stdout, stderr).
+        """
+        if cwd:
+            command = f"cd {cwd} && {command}"
+        stdin, stdout, stderr = self.ssh_client.exec_command(command)
+        return stdout.read().decode(), stderr.read().decode()

tests/integration/test_zos_uss.py

@@ -0,0 +1,39 @@
+"""Integration tests for the Zowe Python SDK z/OS UNIX System Services (USS) package."""
+
+import unittest
+from unittest.mock import patch, MagicMock
+from zowe.zos_uss_for_zowe_sdk import Uss
+
+class TestUss(unittest.TestCase):
+    def setUp(self):
+        self.profile = {
+            "hostname": "example.com",
+            "username": "user",
+            "password": "pass",
+            "port": 22,
+        }
+        self.uss = Uss(self.profile)
+
+    @patch("paramiko.SSHClient")
+    def test_connect(self, mock_ssh_client):
+        mock_client = MagicMock()
+        mock_ssh_client.return_value = mock_client
+        self.uss.connect()
+        mock_client.connect.assert_called_with(
+            hostname="example.com", username="user", password="pass", port=22
+        )
+
+    @patch("paramiko.SSHClient")
+    def test_execute_command(self, mock_ssh_client):
+        mock_client = MagicMock()
+        mock_ssh_client.return_value = mock_client
+        mock_stdout = MagicMock()
+        mock_stdout.read.return_value = b"command output"
+        mock_stderr = MagicMock()
+        mock_stderr.read.return_value = b""
+        mock_client.exec_command.return_value = (None, mock_stdout, mock_stderr)
+
+        self.uss.connect()
+        output, error = self.uss.execute_command("ls -l")
+        self.assertEqual(output, "command output")
+        self.assertEqual(error, "")

tests/unit/test_zos_uss.py

@@ -0,0 +1,48 @@
+"""Unit tests for the Zowe Python SDK z/OS UNIX System Services (USS) package."""
+
+import unittest
+from unittest.mock import patch, MagicMock
+
+from zowe.zos_uss_for_zowe_sdk import Uss
+
+
+class TestUss(unittest.TestCase):
+    def setUp(self):
+        self.profile = {
+            "host": "mock-url.com",
+            "user": "Username",
+            "password": "Password",
+            "port": 22,
+        }
+        self.uss = Uss(connection=self.profile)
+
+    @patch("paramiko.SSHClient")
+    def test_connect(self, mock_ssh_client):
+        self.uss.connect()
+        mock_ssh_client.assert_called_once()
+        self.assertIsNotNone(self.uss.ssh_client)
+
+    @patch("paramiko.SSHClient")
+    def test_disconnect(self, mock_ssh_client):
+        mock_ssh = mock_ssh_client.return_value
+        self.uss.ssh_client = mock_ssh
+        self.uss.disconnect()
+        mock_ssh.close.assert_called_once()
+
+    @patch("paramiko.SSHClient")
+    def test_execute_command(self, mock_ssh_client):
+        mock_ssh = mock_ssh_client.return_value
+        mock_stdout = MagicMock()
+        mock_stderr = MagicMock()
+        mock_stdout.read.return_value = b"Command executed successfully"
+        mock_stderr.read.return_value = b""
+        mock_ssh.exec_command.return_value = (None, mock_stdout, mock_stderr)
+
+        self.uss.ssh_client = mock_ssh
+        stdout, stderr = self.uss.execute_command("ls -la")
+        self.assertEqual(stdout, "Command executed successfully")
+        self.assertEqual(stderr, "")
+
+
+if __name__ == "__main__":
+    unittest.main()