cli: use nix_shell also in ssh.py
This commit is contained in:
@@ -3,6 +3,8 @@ import json
|
|||||||
import subprocess
|
import subprocess
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
|
from .nix import nix_shell
|
||||||
|
|
||||||
|
|
||||||
def ssh(
|
def ssh(
|
||||||
host: str,
|
host: str,
|
||||||
@@ -10,15 +12,10 @@ def ssh(
|
|||||||
password: Optional[str] = None,
|
password: Optional[str] = None,
|
||||||
ssh_args: list[str] = [],
|
ssh_args: list[str] = [],
|
||||||
) -> None:
|
) -> None:
|
||||||
nix_shell_args = []
|
packages = ["tor", "openssh"]
|
||||||
password_args = []
|
password_args = []
|
||||||
if password:
|
if password:
|
||||||
nix_shell_args = [
|
packages.append("sshpass")
|
||||||
"nix",
|
|
||||||
"shell",
|
|
||||||
"nixpkgs#sshpass",
|
|
||||||
"-c",
|
|
||||||
]
|
|
||||||
password_args = [
|
password_args = [
|
||||||
"sshpass",
|
"sshpass",
|
||||||
"-p",
|
"-p",
|
||||||
@@ -32,23 +29,22 @@ def ssh(
|
|||||||
"StrictHostKeyChecking=no",
|
"StrictHostKeyChecking=no",
|
||||||
f"{user}@{host}",
|
f"{user}@{host}",
|
||||||
]
|
]
|
||||||
cmd = nix_shell_args + ["torify"] + password_args + _ssh_args
|
cmd = nix_shell(packages, ["torify"] + password_args + _ssh_args)
|
||||||
subprocess.run(cmd)
|
subprocess.run(cmd)
|
||||||
|
|
||||||
|
|
||||||
def qrcode_scan(pictureFile: str) -> str:
|
def qrcode_scan(pictureFile: str) -> str:
|
||||||
return (
|
return (
|
||||||
subprocess.run(
|
subprocess.run(
|
||||||
[
|
nix_shell(
|
||||||
"nix",
|
["zbar"],
|
||||||
"shell",
|
[
|
||||||
"nixpkgs#zbar",
|
"zbarimg",
|
||||||
"-c",
|
"--quiet",
|
||||||
"zbarimg",
|
"--raw",
|
||||||
"--quiet",
|
pictureFile,
|
||||||
"--raw",
|
],
|
||||||
pictureFile,
|
),
|
||||||
],
|
|
||||||
stdout=subprocess.PIPE,
|
stdout=subprocess.PIPE,
|
||||||
check=True,
|
check=True,
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -1,5 +1,7 @@
|
|||||||
|
import os
|
||||||
import sys
|
import sys
|
||||||
from typing import Union
|
from contextlib import contextmanager
|
||||||
|
from typing import Iterator, Union
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
import pytest_subprocess.fake_process
|
import pytest_subprocess.fake_process
|
||||||
@@ -18,45 +20,72 @@ def test_no_args(
|
|||||||
assert captured.err.startswith("usage:")
|
assert captured.err.startswith("usage:")
|
||||||
|
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def mock_env(**environ: str) -> Iterator[None]:
|
||||||
|
original_environ = dict(os.environ)
|
||||||
|
os.environ.update(environ)
|
||||||
|
try:
|
||||||
|
yield
|
||||||
|
finally:
|
||||||
|
os.environ.clear()
|
||||||
|
os.environ.update(original_environ)
|
||||||
|
|
||||||
|
|
||||||
# using fp fixture from pytest-subprocess
|
# using fp fixture from pytest-subprocess
|
||||||
def test_ssh_no_pass(fp: pytest_subprocess.fake_process.FakeProcess) -> None:
|
def test_ssh_no_pass(fp: pytest_subprocess.fake_process.FakeProcess) -> None:
|
||||||
host = "somehost"
|
with mock_env(CLAN_NIXPKGS="/mocked-nixpkgs"):
|
||||||
user = "user"
|
host = "somehost"
|
||||||
cmd: list[Union[str, utils.Any]] = [
|
user = "user"
|
||||||
"torify",
|
cmd: list[Union[str, utils.Any]] = [
|
||||||
"ssh",
|
"nix",
|
||||||
"-o",
|
"shell",
|
||||||
"UserKnownHostsFile=/dev/null",
|
"-f",
|
||||||
"-o",
|
"/mocked-nixpkgs",
|
||||||
"StrictHostKeyChecking=no",
|
"tor",
|
||||||
f"{user}@{host}",
|
"openssh",
|
||||||
fp.any(),
|
"-c",
|
||||||
]
|
"torify",
|
||||||
fp.register(cmd)
|
"ssh",
|
||||||
clan_cli.ssh.ssh(
|
"-o",
|
||||||
host=host,
|
"UserKnownHostsFile=/dev/null",
|
||||||
user=user,
|
"-o",
|
||||||
)
|
"StrictHostKeyChecking=no",
|
||||||
assert fp.call_count(cmd) == 1
|
f"{user}@{host}",
|
||||||
|
fp.any(),
|
||||||
|
]
|
||||||
|
fp.register(cmd)
|
||||||
|
clan_cli.ssh.ssh(
|
||||||
|
host=host,
|
||||||
|
user=user,
|
||||||
|
)
|
||||||
|
assert fp.call_count(cmd) == 1
|
||||||
|
|
||||||
|
|
||||||
def test_ssh_with_pass(fp: pytest_subprocess.fake_process.FakeProcess) -> None:
|
def test_ssh_with_pass(fp: pytest_subprocess.fake_process.FakeProcess) -> None:
|
||||||
host = "somehost"
|
with mock_env(CLAN_NIXPKGS="/mocked-nixpkgs"):
|
||||||
user = "user"
|
host = "somehost"
|
||||||
cmd: list[Union[str, utils.Any]] = [
|
user = "user"
|
||||||
"nix",
|
cmd: list[Union[str, utils.Any]] = [
|
||||||
"shell",
|
"nix",
|
||||||
"nixpkgs#sshpass",
|
"shell",
|
||||||
"-c",
|
"-f",
|
||||||
fp.any(),
|
"/mocked-nixpkgs",
|
||||||
]
|
"tor",
|
||||||
fp.register(cmd)
|
"openssh",
|
||||||
clan_cli.ssh.ssh(
|
"sshpass",
|
||||||
host=host,
|
"-c",
|
||||||
user=user,
|
"torify",
|
||||||
password="XXX",
|
"sshpass",
|
||||||
)
|
"-p",
|
||||||
assert fp.call_count(cmd) == 1
|
fp.any(),
|
||||||
|
]
|
||||||
|
fp.register(cmd)
|
||||||
|
clan_cli.ssh.ssh(
|
||||||
|
host=host,
|
||||||
|
user=user,
|
||||||
|
password="XXX",
|
||||||
|
)
|
||||||
|
assert fp.call_count(cmd) == 1
|
||||||
|
|
||||||
|
|
||||||
def test_qrcode_scan(fp: pytest_subprocess.fake_process.FakeProcess) -> None:
|
def test_qrcode_scan(fp: pytest_subprocess.fake_process.FakeProcess) -> None:
|
||||||
|
|||||||
Reference in New Issue
Block a user