added tests for init module

tests/test_init.py

@@ -0,0 +1,183 @@
+import os
+import tempfile
+from unittest.mock import patch
+
+import git
+import pytest
+
+from tracksuite.init import LocalHostClient, SSHClient, setup_remote
+
+
+@pytest.fixture
+def ssh_client():
+    return SSHClient("test_host", "test_user")
+
+
+def side_effect_for_run_cmd(ssh_command):
+    return ssh_command
+
+
+@pytest.fixture
+def mock_run_cmd_returns_input(mocker):
+    with patch("tracksuite.init.run_cmd") as mock:
+        mock.side_effect = side_effect_for_run_cmd
+        yield mock
+
+
+def test_ssh_client_exec(ssh_client, mock_run_cmd_returns_input):
+
+    # Execute the method under test
+    command_to_execute = ["echo Hello World"]
+    result = ssh_client.exec(command_to_execute)
+
+    # Assert that the mock was called with the expected command
+    expected_ssh_command = f'ssh test_user@test_host "{command_to_execute[0]}; "'
+    mock_run_cmd_returns_input.assert_called_with(expected_ssh_command)
+
+    # Assert that the result is what our side_effect function returns
+    assert (
+        result == expected_ssh_command
+    ), "The mock did not return the expected dynamic value"
+
+
+class DummyReturnCode:
+    def __init__(self, returncode):
+        self.returncode = returncode
+
+
+@pytest.fixture
+def mock_run_cmd_returns_true(mocker):
+    with patch("tracksuite.init.run_cmd") as mock:
+        mock.return_value = DummyReturnCode(0)
+        yield mock
+
+
+def test_ssh_client_is_path(ssh_client, mock_run_cmd_returns_true):
+
+    ssh_client = SSHClient("test_host", "test_user")
+
+    # Execute the method under test
+    result = ssh_client.is_path("/tmp")
+    assert result is True
+
+
+def test_localhost_client_different_user():
+    with pytest.raises(Exception):
+        LocalHostClient("localhost", "invalid_user")
+
+
+def test_localhost_client_same_user():
+    current_user = os.getenv("USER")
+    LocalHostClient("localhost", current_user)
+
+
+def test_localhost_client_exec():
+    current_user = os.getenv("USER")
+    localhost = LocalHostClient("localhost", current_user)
+    command_to_execute = ["echo Hello World"]
+    result = localhost.exec(command_to_execute)
+    assert result.returncode == 0
+
+
+def test_localhost_client_is_path():
+    current_user = os.getenv("USER")
+    localhost = LocalHostClient("localhost", current_user)
+
+    with tempfile.TemporaryDirectory() as temp_dir:
+        test_dir = os.path.join(temp_dir, "test_dir")
+        localhost.exec(f"mkdir {test_dir}")
+        localhost.is_path(test_dir)
+
+
+def test_setup_remote():
+
+    with tempfile.TemporaryDirectory() as temp_dir:
+
+        remote_path = os.path.join(temp_dir, "remote")
+        current_user = os.getenv("USER")
+        setup_remote(
+            host="localhost",
+            user=current_user,
+            target_dir=remote_path,
+        )
+        assert os.path.exists(remote_path)
+        assert os.path.exists(os.path.join(remote_path, ".git"))
+
+        repo = git.Repo(remote_path)
+        commit_history = repo.iter_commits()
+        for commit in commit_history:
+            assert "first commit" in commit.message
+
+
+def test_setup_remote_with_backup():
+
+    with tempfile.TemporaryDirectory() as temp_dir:
+
+        repo_path = os.path.join(temp_dir, "my_repo.git")
+        repo = git.Repo.init(repo_path, bare=True)
+
+        remote_path = os.path.join(temp_dir, "remote")
+        current_user = os.getenv("USER")
+        setup_remote(
+            host="localhost",
+            user=current_user,
+            target_dir=remote_path,
+            remote=repo_path,
+        )
+        assert os.path.exists(remote_path)
+        assert os.path.exists(os.path.join(remote_path, ".git"))
+
+        commit_history = repo.iter_commits()
+        for commit in commit_history:
+            assert "first commit" in commit.message
+            print(commit.message)
+
+
+def test_setup_remote_with_backup_fail():
+
+    with tempfile.TemporaryDirectory() as temp_dir:
+
+        repo_path = os.path.join(temp_dir, "my_repo.git")
+        repo = git.Repo.init(repo_path, bare=True)
+
+        # Add a dummy commit
+        repo.index.commit("Dummy commit 1")
+        repo.index.commit("Dummy commit 2")
+        commit_history = repo.iter_commits()
+        for commit in commit_history:
+            print(commit.message)
+
+        remote_path = os.path.join(temp_dir, "remote")
+        current_user = os.getenv("USER")
+        with pytest.raises(Exception):
+            setup_remote(
+                host="localhost",
+                user=current_user,
+                target_dir=remote_path,
+                remote=repo_path,
+            )
+
+
+def test_setup_remote_with_backup_force():
+
+    with tempfile.TemporaryDirectory() as temp_dir:
+
+        repo_path = os.path.join(temp_dir, "my_repo.git")
+        repo = git.Repo.init(repo_path, bare=True)
+
+        # Add a dummy commit
+        repo.index.commit("Dummy commit 1")
+        repo.index.commit("Dummy commit 2")
+        commit_history = repo.iter_commits()
+        for commit in commit_history:
+            print(commit.message)
+
+        remote_path = os.path.join(temp_dir, "remote")
+        current_user = os.getenv("USER")
+        setup_remote(
+            host="localhost",
+            user=current_user,
+            target_dir=remote_path,
+            remote=repo_path,
+            force=True,
+        )

tracksuite/init.py

@@ -53,10 +53,11 @@ def __init__(self, host, user, ssh_options=None):
     def is_path(self, path):
         # Build the ssh command
         cmd = [f"[ -d {path} ] && exit 0 || exit 1"]
+        ret = self.exec(cmd)
         try:
             ret = self.exec(cmd)
             return ret.returncode == 0
-        except:
+        except Exception:
             return False
 
     def exec(self, commands, dir=None):
@@ -68,6 +69,7 @@ def exec(self, commands, dir=None):
             ssh_command += f"cd {dir}; "
         for cmd in commands:
             ssh_command += f"{cmd}; "
+        ssh_command = ssh_command + '"'
         value = run_cmd(ssh_command)
         return value
 
@@ -79,6 +81,12 @@ class LocalHostClient(Client):
 
     def __init__(self, host, user):
         assert host == "localhost"
+        if user != os.getenv("USER"):
+            raise ValueError(
+                "Localhost user cannot be different than executing user. "+
+                "To deploy with a different user, use a different host."
+            )
+
         super().__init__(host, user)
 
     def is_path(self, path):
@@ -124,7 +132,8 @@ def setup_remote(host, user, target_dir, remote=None, force=False):
     ret = ssh.exec(f"mkdir -p {target_dir}; exit 0")
     if not ssh.is_path(target_dir):
         raise Exception(
-            f"Target directory {target_dir} not properly created on {host} with user {user}\n\n" + ret.stdout
+            f"Target directory {target_dir} not properly created on {host} with user {user}\n\n"
+            + ret.stdout
         )
 
     target_git = os.path.join(target_dir, ".git")
@@ -150,7 +159,9 @@ def setup_remote(host, user, target_dir, remote=None, force=False):
 
         # making sure we can clone the repository
         if not ssh.is_path(target_git):
-            print(f'Target directory {target_dir} not properaly created on {host} with user {user}')
+            print(
+                f"Target directory {target_dir} not properaly created on {host} with user {user}"
+            )
             raise Exception(ret.stdout)
 
         with tempfile.TemporaryDirectory() as tmp_repo:

tracksuite/utils.py

@@ -4,9 +4,9 @@
 class CmdError(Exception):
     def __init__(self, cause, process_ret):
         super().__init__(
-            f'ERROR: Command failed with {cause} ({process_ret.returncode})'
-            f'\nCalled: {process_ret.args}'
-            f'\nOutput: {process_ret.output}'
+            f"ERROR: Command failed with {cause} ({process_ret.returncode})"
+            f"\nCalled: {process_ret.args}"
+            f"\nOutput: {process_ret.output}"
         )
 
 
@@ -22,17 +22,22 @@ def run_cmd(cmd, timeout=300, **kwargs):
     """
     try:
         ret = subprocess.run(
-            cmd, shell=True, check=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
-            timeout=timeout, encoding='utf8',
-            **kwargs
+            cmd,
+            shell=True,
+            check=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            timeout=timeout,
+            encoding="utf8",
+            **kwargs,
         )
     except subprocess.TimeoutExpired as exc:
         exc.returncode = -1
-        raise CmdError('timeout', exc)
+        raise CmdError("timeout", exc)
     except subprocess.CalledProcessError as exc:
-        raise CmdError('error', exc)
+        raise CmdError("error", exc)
     except Exception as exc:
         exc.returncode = 99
         exc.output = str(exc)
-        raise CmdError('foreign error', exc)
+        raise CmdError("foreign error", exc)
     return ret