Merge pull request 'move ssh cli to cli submodule' (#113) from Mic92-mic92 into main

pkgs/clan-cli/clan_cli/__init__.py

@@ -1,8 +1,9 @@
 import argparse
 import sys
 
-from . import admin, secrets, ssh
+from . import admin, secrets
 from .errors import ClanError
+from .ssh import cli as ssh_cli
 
 has_argcomplete = True
 try:
@@ -27,7 +28,7 @@ def main() -> None:
     #    warn(f"The config command does not work in the nix sandbox: {e}")
 
     parser_ssh = subparsers.add_parser("ssh", help="ssh to a remote machine")
-    ssh.register_parser(parser_ssh)
+    ssh_cli.register_parser(parser_ssh)
 
     parser_secrets = subparsers.add_parser("secrets", help="manage secrets")
     secrets.register_parser(parser_secrets)

pkgs/clan-cli/clan_cli/ssh/__init__.py

@@ -144,7 +144,7 @@ class HostKeyCheck(Enum):
     NONE = 2
 
 
-class DeployHost:
+class Host:
     def __init__(
         self,
         host: str,
@@ -158,7 +158,7 @@ class DeployHost:
         verbose_ssh: bool = False,
     ) -> None:
         """
-        Creates a DeployHost
+        Creates a Host
         @host the hostname to connect to via ssh
         @port the port to connect to via ssh
         @forward_agent: wheter to forward ssh agent
@@ -495,7 +495,7 @@ T = TypeVar("T")
 
 
 class HostResult(Generic[T]):
-    def __init__(self, host: DeployHost, result: Union[T, Exception]) -> None:
+    def __init__(self, host: Host, result: Union[T, Exception]) -> None:
         self.host = host
         self._result = result
 
@@ -518,12 +518,12 @@ class HostResult(Generic[T]):
         return self._result
 
 
-DeployResults = List[HostResult[subprocess.CompletedProcess[str]]]
+Results = List[HostResult[subprocess.CompletedProcess[str]]]
 
 
 def _worker(
-    func: Callable[[DeployHost], T],
-    host: DeployHost,
+    func: Callable[[Host], T],
+    host: Host,
     results: List[HostResult[T]],
     idx: int,
 ) -> None:
@@ -534,15 +534,15 @@ def _worker(
         results[idx] = HostResult(host, e)
 
 
-class DeployGroup:
-    def __init__(self, hosts: List[DeployHost]) -> None:
+class Group:
+    def __init__(self, hosts: List[Host]) -> None:
         self.hosts = hosts
 
     def _run_local(
         self,
         cmd: Union[str, List[str]],
-        host: DeployHost,
-        results: DeployResults,
+        host: Host,
+        results: Results,
         stdout: FILE = None,
         stderr: FILE = None,
         extra_env: Dict[str, str] = {},
@@ -569,8 +569,8 @@ class DeployGroup:
     def _run_remote(
         self,
         cmd: Union[str, List[str]],
-        host: DeployHost,
-        results: DeployResults,
+        host: Host,
+        results: Results,
         stdout: FILE = None,
         stderr: FILE = None,
         extra_env: Dict[str, str] = {},
@@ -621,8 +621,8 @@ class DeployGroup:
         check: bool = True,
         verbose_ssh: bool = False,
         timeout: float = math.inf,
-    ) -> DeployResults:
-        results: DeployResults = []
+    ) -> Results:
+        results: Results = []
         threads = []
         for host in self.hosts:
             fn = self._run_local if local else self._run_remote
@@ -662,7 +662,7 @@ class DeployGroup:
         check: bool = True,
         verbose_ssh: bool = False,
         timeout: float = math.inf,
-    ) -> DeployResults:
+    ) -> Results:
         """
         Command to run on the remote host via ssh
         @stdout if not None stdout of the command will be redirected to this file i.e. stdout=subprocss.PIPE
@@ -671,7 +671,7 @@ class DeployGroup:
         @verbose_ssh: Enables verbose logging on ssh connections
         @timeout: Timeout in seconds for the command to complete
 
-        @return a lists of tuples containing DeployNode and the result of the command for this DeployNode
+        @return a lists of tuples containing Host and the result of the command for this Host
         """
         return self._run(
             cmd,
@@ -693,7 +693,7 @@ class DeployGroup:
         cwd: Union[None, str, Path] = None,
         check: bool = True,
         timeout: float = math.inf,
-    ) -> DeployResults:
+    ) -> Results:
         """
         Command to run locally for each host in the group in parallel
         @cmd the commmand to run
@@ -703,7 +703,7 @@ class DeployGroup:
         @extra_env environment variables to override whe running the command
         @timeout: Timeout in seconds for the command to complete
 
-        @return a lists of tuples containing DeployNode and the result of the command for this DeployNode
+        @return a lists of tuples containing Host and the result of the command for this Host
         """
         return self._run(
             cmd,
@@ -717,7 +717,7 @@ class DeployGroup:
         )
 
     def run_function(
-        self, func: Callable[[DeployHost], T], check: bool = True
+        self, func: Callable[[Host], T], check: bool = True
     ) -> List[HostResult[T]]:
         """
         Function to run for each host in the group in parallel
@@ -745,9 +745,9 @@ class DeployGroup:
             self._reraise_errors(results)
         return results
 
-    def filter(self, pred: Callable[[DeployHost], bool]) -> "DeployGroup":
-        """Return a new DeployGroup with the results filtered by the predicate"""
-        return DeployGroup(list(filter(pred, self.hosts)))
+    def filter(self, pred: Callable[[Host], bool]) -> "Group":
+        """Return a new Group with the results filtered by the predicate"""
+        return Group(list(filter(pred, self.hosts)))
 
 
 @overload

pkgs/clan-cli/clan_cli/ssh/cli.py

@@ -3,7 +3,7 @@ import json
 import subprocess
 from typing import Optional
 
-from .nix import nix_shell
+from ..nix import nix_shell
 
 
 def ssh(

pkgs/clan-cli/tests/age_keys.py

@@ -24,7 +24,7 @@ KEYS = [
 
 
 @pytest.fixture
-def test_keys() -> list[KeyPair]:
+def age_keys() -> list[KeyPair]:
     """
     Root directory of the tests
     """

pkgs/clan-cli/tests/conftest.py

@@ -3,4 +3,4 @@ import sys
 
 sys.path.append(os.path.join(os.path.dirname(__file__), "helpers"))
 
-pytest_plugins = ["temporary_dir", "clan_flake", "root", "test_keys"]
+pytest_plugins = ["temporary_dir", "clan_flake", "root", "age_keys"]

pkgs/clan-cli/tests/test_import_sops_cli.py

@@ -6,21 +6,21 @@ from environment import mock_env
 from secret_cli import SecretCli
 
 if TYPE_CHECKING:
-    from test_keys import KeyPair
+    from age_keys import KeyPair
 
 
 def test_import_sops(
     test_root: Path,
     clan_flake: Path,
     capsys: pytest.CaptureFixture,
-    test_keys: list["KeyPair"],
+    age_keys: list["KeyPair"],
 ) -> None:
     cli = SecretCli()
 
-    with mock_env(SOPS_AGE_KEY=test_keys[1].privkey):
-        cli.run(["machines", "add", "machine1", test_keys[0].pubkey])
-        cli.run(["users", "add", "user1", test_keys[1].pubkey])
-        cli.run(["users", "add", "user2", test_keys[2].pubkey])
+    with mock_env(SOPS_AGE_KEY=age_keys[1].privkey):
+        cli.run(["machines", "add", "machine1", age_keys[0].pubkey])
+        cli.run(["users", "add", "user1", age_keys[1].pubkey])
+        cli.run(["users", "add", "user2", age_keys[2].pubkey])
         cli.run(["groups", "add-user", "group1", "user1"])
         cli.run(["groups", "add-user", "group1", "user2"])
 

pkgs/clan-cli/tests/test_secrets_cli.py

@@ -9,22 +9,22 @@ from secret_cli import SecretCli
 from clan_cli.errors import ClanError
 
 if TYPE_CHECKING:
-    from test_keys import KeyPair
+    from age_keys import KeyPair
 
 
 def _test_identities(
     what: str,
     clan_flake: Path,
     capsys: pytest.CaptureFixture,
-    test_keys: list["KeyPair"],
+    age_keys: list["KeyPair"],
 ) -> None:
     cli = SecretCli()
     sops_folder = clan_flake / "sops"
 
-    cli.run([what, "add", "foo", test_keys[0].pubkey])
+    cli.run([what, "add", "foo", age_keys[0].pubkey])
     assert (sops_folder / what / "foo" / "key.json").exists()
     with pytest.raises(ClanError):
-        cli.run([what, "add", "foo", test_keys[0].pubkey])
+        cli.run([what, "add", "foo", age_keys[0].pubkey])
 
     cli.run(
         [
@@ -32,7 +32,7 @@ def _test_identities(
             "add",
             "-f",
             "foo",
-            test_keys[0].privkey,
+            age_keys[0].privkey,
         ]
     )
     capsys.readouterr()  # empty the buffer
@@ -54,19 +54,19 @@ def _test_identities(
 
 
 def test_users(
-    clan_flake: Path, capsys: pytest.CaptureFixture, test_keys: list["KeyPair"]
+    clan_flake: Path, capsys: pytest.CaptureFixture, age_keys: list["KeyPair"]
 ) -> None:
-    _test_identities("users", clan_flake, capsys, test_keys)
+    _test_identities("users", clan_flake, capsys, age_keys)
 
 
 def test_machines(
-    clan_flake: Path, capsys: pytest.CaptureFixture, test_keys: list["KeyPair"]
+    clan_flake: Path, capsys: pytest.CaptureFixture, age_keys: list["KeyPair"]
 ) -> None:
-    _test_identities("machines", clan_flake, capsys, test_keys)
+    _test_identities("machines", clan_flake, capsys, age_keys)
 
 
 def test_groups(
-    clan_flake: Path, capsys: pytest.CaptureFixture, test_keys: list["KeyPair"]
+    clan_flake: Path, capsys: pytest.CaptureFixture, age_keys: list["KeyPair"]
 ) -> None:
     cli = SecretCli()
     capsys.readouterr()  # empty the buffer
@@ -77,13 +77,13 @@ def test_groups(
         cli.run(["groups", "add-machine", "group1", "machine1"])
     with pytest.raises(ClanError):  # user does not exist yet
         cli.run(["groups", "add-user", "groupb1", "user1"])
-    cli.run(["machines", "add", "machine1", test_keys[0].pubkey])
+    cli.run(["machines", "add", "machine1", age_keys[0].pubkey])
     cli.run(["groups", "add-machine", "group1", "machine1"])
 
     # Should this fail?
     cli.run(["groups", "add-machine", "group1", "machine1"])
 
-    cli.run(["users", "add", "user1", test_keys[0].pubkey])
+    cli.run(["users", "add", "user1", age_keys[0].pubkey])
     cli.run(["groups", "add-user", "group1", "user1"])
 
     capsys.readouterr()  # empty the buffer
@@ -99,7 +99,7 @@ def test_groups(
 
 
 def test_secrets(
-    clan_flake: Path, capsys: pytest.CaptureFixture, test_keys: list["KeyPair"]
+    clan_flake: Path, capsys: pytest.CaptureFixture, age_keys: list["KeyPair"]
 ) -> None:
     cli = SecretCli()
     capsys.readouterr()  # empty the buffer
@@ -125,18 +125,18 @@ def test_secrets(
         cli.run(["list"])
         assert capsys.readouterr().out == "key\n"
 
-        cli.run(["machines", "add", "machine1", test_keys[0].pubkey])
+        cli.run(["machines", "add", "machine1", age_keys[0].pubkey])
         cli.run(["machines", "add-secret", "machine1", "key"])
 
-        with mock_env(SOPS_AGE_KEY=test_keys[0].privkey, SOPS_AGE_KEY_FILE=""):
+        with mock_env(SOPS_AGE_KEY=age_keys[0].privkey, SOPS_AGE_KEY_FILE=""):
             capsys.readouterr()
             cli.run(["get", "key"])
             assert capsys.readouterr().out == "foo"
         cli.run(["machines", "remove-secret", "machine1", "key"])
 
-        cli.run(["users", "add", "user1", test_keys[1].pubkey])
+        cli.run(["users", "add", "user1", age_keys[1].pubkey])
         cli.run(["users", "add-secret", "user1", "key"])
-        with mock_env(SOPS_AGE_KEY=test_keys[1].privkey, SOPS_AGE_KEY_FILE=""):
+        with mock_env(SOPS_AGE_KEY=age_keys[1].privkey, SOPS_AGE_KEY_FILE=""):
             capsys.readouterr()
             cli.run(["get", "key"])
             assert capsys.readouterr().out == "foo"
@@ -151,7 +151,7 @@ def test_secrets(
         capsys.readouterr()  # empty the buffer
         cli.run(["set", "--group", "admin-group", "key2"])
 
-        with mock_env(SOPS_AGE_KEY=test_keys[1].privkey, SOPS_AGE_KEY_FILE=""):
+        with mock_env(SOPS_AGE_KEY=age_keys[1].privkey, SOPS_AGE_KEY_FILE=""):
             capsys.readouterr()
             cli.run(["get", "key"])
             assert capsys.readouterr().out == "foo"

pkgs/clan-cli/tests/test_ssh_cli.py

@@ -6,7 +6,8 @@ import pytest_subprocess.fake_process
 from environment import mock_env
 from pytest_subprocess import utils
 
-import clan_cli.ssh
+import clan_cli
+from clan_cli.ssh import cli
 
 
 def test_no_args(
@@ -40,7 +41,7 @@ def test_ssh_no_pass(fp: pytest_subprocess.fake_process.FakeProcess) -> None:
             fp.any(),
         ]
         fp.register(cmd)
-        clan_cli.ssh.ssh(
+        cli.ssh(
             host=host,
             user=user,
         )
@@ -64,7 +65,7 @@ def test_ssh_with_pass(fp: pytest_subprocess.fake_process.FakeProcess) -> None:
             fp.any(),
         ]
         fp.register(cmd)
-        clan_cli.ssh.ssh(
+        cli.ssh(
             host=host,
             user=user,
             password="XXX",
@@ -75,5 +76,5 @@ def test_ssh_with_pass(fp: pytest_subprocess.fake_process.FakeProcess) -> None:
 def test_qrcode_scan(fp: pytest_subprocess.fake_process.FakeProcess) -> None:
     cmd: list[Union[str, utils.Any]] = [fp.any()]
     fp.register(cmd, stdout="https://test.test")
-    result = clan_cli.ssh.qrcode_scan("test.png")
+    result = cli.qrcode_scan("test.png")
     assert result == "https://test.test"