OsOperations::create_clone is added

postgrespro · dmitry-lipetsk · Jun 30, 2025 · Jun 26, 2025 · Jun 26, 2025 · Jun 26, 2025
commit 1062a973d763651cc7a560782b0d766be72e2d81
diff --git a/testgres/operations/local_ops.py b/testgres/operations/local_ops.py
@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import getpass
 import logging
 import os
@@ -11,6 +13,7 @@
 import psutil
 import typing
 import threading
+import copy
 
 from ..exceptions import ExecUtilException
 from ..exceptions import InvalidOperationException
@@ -29,13 +32,26 @@
 
 
 class LocalOperations(OsOperations):
+    sm_dummy_conn_params = ConnectionParams()
     sm_single_instance: OsOperations = None
     sm_single_instance_guard = threading.Lock()
 
+    # TODO: make it read-only
+    conn_params: ConnectionParams
+    host: str
+    ssh_key: typing.Optional[str]
+    remote: bool
+    username: str
+
     def __init__(self, conn_params=None):
+        super().__init__()
+
+        if conn_params is __class__.sm_dummy_conn_params:
+            return
+
         if conn_params is None:
             conn_params = ConnectionParams()
-        super(LocalOperations, self).__init__(conn_params.username)
+
         self.conn_params = conn_params
         self.host = conn_params.host
         self.ssh_key = None
@@ -58,6 +74,15 @@ def get_single_instance() -> OsOperations:
         assert type(__class__.sm_single_instance) == __class__  # noqa: E721
         return __class__.sm_single_instance
 
+    def create_clone(self) -> LocalOperations:
+        clone = __class__(__class__.sm_dummy_conn_params)
+        clone.conn_params = copy.copy(self.conn_params)
+        clone.host = self.host
+        clone.ssh_key = self.ssh_key
+        clone.remote = self.remote
+        clone.username = self.username
+        return clone
+
     @staticmethod
     def _process_output(encoding, temp_file_path):
         """Process the output of a command from a temporary file."""

diff --git a/testgres/operations/os_ops.py b/testgres/operations/os_ops.py
@@ -1,4 +1,5 @@
-import getpass
+from __future__ import annotations
+
 import locale
 
 
@@ -17,9 +18,11 @@ def get_default_encoding():
 
 
 class OsOperations:
-    def __init__(self, username=None):
-        self.ssh_key = None
-        self.username = username or getpass.getuser()
+    def __init__(self):
+        pass
+
+    def create_clone(self) -> OsOperations:
+        raise NotImplementedError()
 
     # Command execution
     def exec_command(self, cmd, **kwargs):

diff --git a/testgres/operations/remote_ops.py b/testgres/operations/remote_ops.py
@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import getpass
 import os
 import posixpath
@@ -7,6 +9,7 @@
 import io
 import logging
 import typing
+import copy
 
 from ..exceptions import ExecUtilException
 from ..exceptions import InvalidOperationException
@@ -42,11 +45,29 @@ def cmdline(self):
 
 
 class RemoteOperations(OsOperations):
+    sm_dummy_conn_params = ConnectionParams()
+
+    conn_params: ConnectionParams
+    host: str
+    port: int
+    ssh_key: str
+    ssh_args: list
+    remote: bool
+    username: str
+    ssh_dest: str
+
     def __init__(self, conn_params: ConnectionParams):
         if not platform.system().lower() == "linux":
             raise EnvironmentError("Remote operations are supported only on Linux!")
 
-        super().__init__(conn_params.username)
+        if conn_params is None:
+            raise ValueError("Argument 'conn_params' is None.")
+
+        super().__init__()
+
+        if conn_params is __class__.sm_dummy_conn_params:
+            return
+
         self.conn_params = conn_params
         self.host = conn_params.host
         self.port = conn_params.port
@@ -63,6 +84,18 @@ def __init__(self, conn_params: ConnectionParams):
     def __enter__(self):
         return self
 
+    def create_clone(self) -> RemoteOperations:
+        clone = __class__(__class__.sm_dummy_conn_params)
+        clone.conn_params = copy.copy(self.conn_params)
+        clone.host = self.host
+        clone.port = self.port
+        clone.ssh_key = self.ssh_key
+        clone.ssh_args = copy.copy(self.ssh_args)
+        clone.remote = self.remote
+        clone.username = self.username
+        clone.ssh_dest = self.ssh_dest
+        return clone
+
     def exec_command(
         self, cmd, wait_exit=False, verbose=False, expect_error=False,
         encoding=None, shell=True, text=False, input=None, stdin=None, stdout=None,

diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py
@@ -37,6 +37,13 @@ def os_ops(self, request: pytest.FixtureRequest) -> OsOperations:
         assert isinstance(request.param, OsOperations)
         return request.param
 
+    def test_create_clone(self, os_ops: OsOperations):
+        assert isinstance(os_ops, OsOperations)
+        clone = os_ops.create_clone()
+        assert clone is not None
+        assert clone is not os_ops
+        assert type(clone) == type(os_ops)  # noqa: E721
+
     def test_exec_command_success(self, os_ops: OsOperations):
         """
         Test exec_command for successful command execution.