started refactoring of SSH__Python methods

osbot_utils/helpers/ssh/SSH.py

@@ -6,85 +6,25 @@
 from osbot_utils.decorators.lists.index_by import index_by
 from osbot_utils.decorators.methods.cache_on_self import cache_on_self
 from osbot_utils.helpers.ssh.SSH__Execute import SSH__Execute
+from osbot_utils.helpers.ssh.SSH__Python import SSH__Python
 from osbot_utils.utils.Dev import pprint
 from osbot_utils.utils.Env import get_env
 from osbot_utils.utils.Functions import function_source_code
 from osbot_utils.utils.Misc import timestamp_utc_now, str_to_bool, str_to_int
-from osbot_utils.utils.Process import start_process, run_process
-from osbot_utils.utils.Status import status_error
+from osbot_utils.utils.Process          import start_process, run_process
+from osbot_utils.utils.Status           import status_error
 
 
 class SSH(Kwargs_To_Self):
 
-
-
-
-
-    def execute_python__code(self, python_code, python_executable='python3'):
-        python_command  = f"{python_executable} -c \"{python_code}\""
-        return self.execute_command(python_command)
-
-    def execute_python__code__return_stdout(self, *args, **kwargs):
-        return self.execute_python__code(*args, **kwargs).get('stdout').strip()
-
-    def execute_python__function(self, function, python_executable='python3'):
-        function_name   = function.__name__
-        function_code   = function_source_code(function)
-        exec_code       = f"{function_code}\nresult= {function_name}(); print(result)"
-        python_command  = f"{python_executable} -c \"{exec_code}\""
-        return self.execute_command(python_command)
-
-    def execute_python__function__return_stderr(self, *args, **kwargs):
-        return self.execute_python__function(*args, **kwargs).get('stderr').strip()
-
-    def execute_python__function__return_stdout(self, *args, **kwargs):
-        return self.execute_python__function(*args, **kwargs).get('stdout').strip()
-
-
-
-    # helpers for common linux methods
-
-
-
-    def rm(self, path=''):
-        command = f'rm {path}'
-        return self.execute_command__return_stderr(command)
-
-
-
-
-
-
-    def whoami(self):
-        command = f'whoami'                                     # todo: security-vuln: add protection against code injection
-        return self.execute_command__return_stdout(command)
-
-
-
-
-
-    # def ifconfig(self):
-    #     command = "export PATH=$PATH:/sbin && ifconfig"               # todo add example with PATH modification
-    #     return self.execute_command__return_stdout(command)
-
-    # def ifconfig(self):                                               # todo add command to execute in separate bash (see when it is needed)
-    #     command = "bash -l -c 'ifconfig'"
-    #     return self.execute_command__return_stdout(command)
-    # if port_forward:      # todo: add support for port forward   (this will need async execution)
-    #     local_port  = port_forward.get('local_port' )
-    #     remote_ip   = port_forward.get('remote_ip'  )
-    #     remote_port = port_forward.get('remote_port')
-
-
     def setup(self):
         self.ssh_execute().setup()
         return self
 
-
-    # todo: refactor all methods above to specfic classes related to what those methods are
-    #       then create methods (like bellow) to provide a more user friendly interface
-
     @cache_on_self
     def ssh_execute(self):
         return SSH__Execute()
 
+    @cache_on_self
+    def ssh_python(self):
+        return SSH__Python(ssh_execute = self.ssh_execute())

osbot_utils/helpers/ssh/SSH__Linux.py

@@ -54,6 +54,10 @@ def mkdir(self, folder):
         command = f'mkdir -p {folder}'
         return self.ssh.execute_command(command)
 
+    def rm(self, path=''):
+        command = f'rm {path}'
+        return self.ssh.execute_command__return_stderr(command)
+
     def rmdir(self, folder):
         command = f'rmdir {folder}'
         return self.ssh.execute_command(command)
@@ -74,3 +78,20 @@ def which(self, target):
         command = f'which {target}'                                     # todo: security-vuln: add protection against code injection
         return self.ssh.execute_command__return_stdout(command)
 
+    def whoami(self):
+        command = f'whoami'
+        return self.ssh.execute_command__return_stdout(command)
+
+    # todo: add methods below (and respective tests)
+
+    # def ifconfig(self):
+    #     command = "export PATH=$PATH:/sbin && ifconfig"               # todo add example with PATH modification
+    #     return self.execute_command__return_stdout(command)
+
+    # def ifconfig(self):                                               # todo add command to execute in separate bash (see when it is needed)
+    #     command = "bash -l -c 'ifconfig'"
+    #     return self.execute_command__return_stdout(command)
+    # if port_forward:      # todo: add support for port forward   (this will need async execution)
+    #     local_port  = port_forward.get('local_port' )
+    #     remote_ip   = port_forward.get('remote_ip'  )
+    #     remote_port = port_forward.get('remote_port')

osbot_utils/helpers/ssh/SSH__Python.py

@@ -0,0 +1,26 @@
+from osbot_utils.base_classes.Type_Safe import Type_Safe
+from osbot_utils.helpers.ssh.SSH__Execute import SSH__Execute
+
+
+class SSH__Python(Type_Safe):
+    ssh_execute: SSH__Execute
+
+    def execute_python__code(self, python_code, python_executable='python3'):
+        python_command  = f"{python_executable} -c \"{python_code}\""
+        return self.ssh_execute.execute_command(python_command)
+
+    def execute_python__code__return_stdout(self, *args, **kwargs):
+        return self.execute_python__code(*args, **kwargs).get('stdout').strip()
+
+    def execute_python__function(self, function, python_executable='python3'):
+        function_name   = function.__name__
+        function_code   = function_source_code(function)
+        exec_code       = f"{function_code}\nresult= {function_name}(); print(result)"
+        python_command  = f"{python_executable} -c \"{exec_code}\""
+        return self.execute_command(python_command)
+
+    def execute_python__function__return_stderr(self, *args, **kwargs):
+        return self.execute_python__function(*args, **kwargs).get('stderr').strip()
+
+    def execute_python__function__return_stdout(self, *args, **kwargs):
+        return self.execute_python__function(*args, **kwargs).get('stdout').strip()

tests/helpers/ssh/test_SSH__Python.py

@@ -0,0 +1,44 @@
+from osbot_utils.helpers.ssh.SSH__Execute import SSH__Execute
+from osbot_utils.helpers.ssh.SSH__Python import SSH__Python
+from osbot_utils.helpers.ssh.TestCase__SSH import TestCase__SSH
+
+
+class test_SSH__Python(TestCase__SSH):
+    ssh_python = SSH__Python
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.ssh_python = cls.ssh.ssh_python()
+
+    def test_setUpClass(self):
+        assert type(self.ssh_python            ) is SSH__Python
+        assert type(self.ssh_python.ssh_execute) is SSH__Execute
+
+
+    def test_execute_python__code(self):
+        with self.ssh_python as _:
+            assert _.execute_python__code('print(40+2)').get('stdout') == '42\n'
+
+    def test__execute_python__code__return_stdout(self):
+        multi_line_command = """
+import sys
+print(sys.version_info.major, sys.version_info.minor, sys.version_info.micro)
+"""
+        with self.ssh_python as _:
+            assert '3 12 2' in _.execute_python__code__return_stdout(multi_line_command)
+
+# assert _.execute_python__code__return_stdout(multi_line_command) == '3 9 16'#
+
+# def an_function():
+#     return 'Hello from the EC2 instance!'
+# assert _.execute_python__function(an_function).get('stdout') == 'Hello from the EC2 instance!\n'
+# pprint(exec_code)
+
+# def test_osbot_utils():
+#     from osbot_utils.utils.Misc import str_to_base64
+#     an_value = 'this will be base64 encoded!'
+#     return str_to_base64(an_value)
+#
+# function_return_value = _.execute_python__function__return_stdout(test_osbot_utils)
+# assert base64_to_str(function_return_value) == 'this will be base64 encoded!'