yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request 'Revert "clan-cli: deprecate nix_shell() in favor of run_cmd()"' (#1774) from fix-deployment into main

Browse Source 
Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/1774
This commit is contained in:
Mic92
2024-07-17 12:31:48 +00:00
parent 55fc9dd00d 0d6e2539e3
commit 5814e9790f
24 changed files with 116 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
pkgs/clan-app/tests/conftest.py
View File

@@ -3,7 +3,7 @@ from pathlib import Path
import pytest
from clan_cli.custom_logger import setup_logging
from clan_cli.nix import run_cmd
from clan_cli.nix import nix_shell
pytest_plugins = [
"temporary_dir",
@@ -27,12 +27,12 @@ def pytest_sessionstart(session: pytest.Session) -> None:
@pytest.fixture
def git_repo(tmp_path: Path) -> Path:
# initialize a git repository
cmd = run_cmd(["git"], ["git", "init"])
cmd = nix_shell(["nixpkgs#git"], ["git", "init"])
subprocess.run(cmd, cwd=tmp_path, check=True)
# set user.name and user.email
cmd = run_cmd(["git"], ["git", "config", "user.name", "test"])
cmd = nix_shell(["nixpkgs#git"], ["git", "config", "user.name", "test"])
subprocess.run(cmd, cwd=tmp_path, check=True)
cmd = run_cmd(["git"], ["git", "config", "user.email", "test@test.test"])
cmd = nix_shell(["nixpkgs#git"], ["git", "config", "user.email", "test@test.test"])
subprocess.run(cmd, cwd=tmp_path, check=True)
# return the path to the git repository
return tmp_path

4
pkgs/clan-cli/clan_cli/api/directory.py
View File

@@ -5,7 +5,7 @@ from pathlib import Path
from typing import Any, Literal
from clan_cli.errors import ClanError
from clan_cli.nix import run_cmd, run_no_stdout
from clan_cli.nix import nix_shell, run_no_stdout
from . import API
@@ -117,7 +117,7 @@ def show_block_devices() -> Blockdevices:
Abstract api method to show block devices.
It must return a list of block devices.
"""
cmd = run_cmd(["util-linux"], ["lsblk", "--json"])
cmd = nix_shell(["nixpkgs#util-linux"], ["lsblk", "--json"])
proc = run_no_stdout(cmd)
res = proc.stdout.strip()

6
pkgs/clan-cli/clan_cli/api/mdns_discovery.py
View File

@@ -3,7 +3,7 @@ import re
from dataclasses import dataclass
from clan_cli.cmd import run_no_stdout
from clan_cli.nix import run_cmd
from clan_cli.nix import nix_shell
from . import API
@@ -89,8 +89,8 @@ def parse_avahi_output(output: str) -> DNSInfo:
@API.register
def show_mdns() -> DNSInfo:
cmd = run_cmd(
["avahi"],
cmd = nix_shell(
["nixpkgs#avahi"],
[
"avahi-browse",
"--all",

12
pkgs/clan-cli/clan_cli/clan/create.py
View File

@@ -10,7 +10,7 @@ from clan_cli.inventory import Inventory, InventoryMeta
from ..cmd import CmdOut, run
from ..errors import ClanError
from ..nix import nix_command, run_cmd
from ..nix import nix_command, nix_shell
default_template_url: str = "git+https://git.clan.lol/clan/clan-core"
minimal_template_url: str = "git+https://git.clan.lol/clan/clan-core#templates.minimal"
@@ -64,19 +64,21 @@ def create_clan(options: CreateOptions) -> CreateClanResponse:
out = run(command, cwd=directory)
## Begin: setup git
command = run_cmd(["git"], ["git", "init"])
command = nix_shell(["nixpkgs#git"], ["git", "init"])
out = run(command, cwd=directory)
cmd_responses["git init"] = out
command = run_cmd(["git"], ["git", "add", "."])
command = nix_shell(["nixpkgs#git"], ["git", "add", "."])
out = run(command, cwd=directory)
cmd_responses["git add"] = out
command = run_cmd(["git"], ["git", "config", "user.name", "clan-tool"])
command = nix_shell(["nixpkgs#git"], ["git", "config", "user.name", "clan-tool"])
out = run(command, cwd=directory)
cmd_responses["git config"] = out
command = run_cmd(["git"], ["git", "config", "user.email", "clan@example.com"])
command = nix_shell(
["nixpkgs#git"], ["git", "config", "user.email", "clan@example.com"]
)
out = run(command, cwd=directory)
cmd_responses["git config"] = out
## End: setup git

8
pkgs/clan-cli/clan_cli/facts/generate.py
View File

@@ -19,7 +19,7 @@ from ..errors import ClanError
from ..git import commit_files
from ..machines.inventory import get_all_machines, get_selected_machines
from ..machines.machines import Machine
from ..nix import run_cmd
from ..nix import nix_shell
from .check import check_secrets
from .public_modules import FactStoreBase
from .secret_modules import SecretStoreBase
@@ -39,10 +39,10 @@ def read_multiline_input(prompt: str = "Finish with Ctrl-D") -> str:
def bubblewrap_cmd(generator: str, facts_dir: Path, secrets_dir: Path) -> list[str]:
# fmt: off
return run_cmd(
return nix_shell(
[
"bash",
"bubblewrap",
"nixpkgs#bash",
"nixpkgs#bubblewrap",
],
[
"bwrap",

18
pkgs/clan-cli/clan_cli/facts/secret_modules/password_store.py
View File

@@ -3,7 +3,7 @@ import subprocess
from pathlib import Path
from clan_cli.machines.machines import Machine
from clan_cli.nix import run_cmd
from clan_cli.nix import nix_shell
from . import SecretStoreBase
@@ -16,8 +16,8 @@ class SecretStore(SecretStoreBase):
self, service: str, name: str, value: bytes, groups: list[str]
) -> Path | None:
subprocess.run(
run_cmd(
["pass"],
nix_shell(
["nixpkgs#pass"],
["pass", "insert", "-m", f"machines/{self.machine.name}/{name}"],
),
input=value,
@@ -27,8 +27,8 @@ class SecretStore(SecretStoreBase):
def get(self, service: str, name: str) -> bytes:
return subprocess.run(
run_cmd(
["pass"],
nix_shell(
["nixpkgs#pass"],
["pass", "show", f"machines/{self.machine.name}/{name}"],
),
check=True,
@@ -49,8 +49,8 @@ class SecretStore(SecretStoreBase):
hashes = []
hashes.append(
subprocess.run(
run_cmd(
["git"],
nix_shell(
["nixpkgs#git"],
[
"git",
"-C",
@@ -68,8 +68,8 @@ class SecretStore(SecretStoreBase):
if symlink.is_symlink():
hashes.append(
subprocess.run(
run_cmd(
["git"],
nix_shell(
["nixpkgs#git"],
[
"git",
"-C",

6
pkgs/clan-cli/clan_cli/facts/upload.py
View File

@@ -7,7 +7,7 @@ from tempfile import TemporaryDirectory
from ..cmd import Log, run
from ..completions import add_dynamic_completer, complete_machines
from ..machines.machines import Machine
from ..nix import run_cmd
from ..nix import nix_shell
log = logging.getLogger(__name__)
@@ -25,8 +25,8 @@ def upload_secrets(machine: Machine) -> None:
ssh_cmd = host.ssh_cmd()
run(
run_cmd(
["rsync"],
nix_shell(
["nixpkgs#rsync"],
[
"rsync",
"-e",

6
pkgs/clan-cli/clan_cli/flash.py
View File

@@ -19,7 +19,7 @@ from .completions import add_dynamic_completer, complete_machines
from .errors import ClanError
from .facts.secret_modules import SecretStoreBase
from .machines.machines import Machine
from .nix import run_cmd
from .nix import nix_shell
log = logging.getLogger(__name__)
@@ -82,8 +82,8 @@ def flash_machine(
disko_install.extend(["--option", "dry-run", "true"])
disko_install.extend(extra_args)
cmd = run_cmd(
["disko"],
cmd = nix_shell(
["nixpkgs#disko"],
disko_install,
)
run(cmd, log=Log.BOTH, error_msg=f"Failed to flash {machine}")

10
pkgs/clan-cli/clan_cli/machines/hardware.py
View File

@@ -9,7 +9,7 @@ from clan_cli.errors import ClanError
from ..cmd import run, run_no_stdout
from ..completions import add_dynamic_completer, complete_machines
from ..nix import nix_config, nix_eval, run_cmd
from ..nix import nix_config, nix_eval, nix_shell
from .types import machine_name_type
log = logging.getLogger(__name__)
@@ -98,12 +98,12 @@ def generate_machine_hardware_info(
Generate hardware information for a machine
and place the resulting *.nix file in the machine's directory.
"""
cmd = run_cmd(
cmd = nix_shell(
[
"openssh",
"sshpass",
"nixpkgs#openssh",
"nixpkgs#sshpass",
# Provides nixos-generate-config on non-NixOS systems
"nixos-install-tools",
"nixpkgs#nixos-install-tools",
],
[
*(["sshpass", "-p", f"{password}"] if password else []),

6
pkgs/clan-cli/clan_cli/machines/install.py
View File

@@ -12,7 +12,7 @@ from ..cmd import Log, run
from ..completions import add_dynamic_completer, complete_machines
from ..facts.generate import generate_facts
from ..machines.machines import Machine
from ..nix import run_cmd
from ..nix import nix_shell
from ..ssh.cli import is_ipv6, is_reachable, qrcode_scan
log = logging.getLogger(__name__)
@@ -81,8 +81,8 @@ def install_nixos(
cmd.append(target_host)
run(
run_cmd(
["nixos-anywhere"],
nix_shell(
["nixpkgs#nixos-anywhere"],
cmd,
),
log=Log.BOTH,

19
pkgs/clan-cli/clan_cli/nix/__init__.py
View File

@@ -100,6 +100,19 @@ def nix_metadata(flake_url: str | Path) -> dict[str, Any]:
return data
def nix_shell(packages: list[str], cmd: list[str]) -> list[str]:
# we cannot use nix-shell inside the nix sandbox
# in our tests we just make sure we have all the packages
if os.environ.get("IN_NIX_SANDBOX") or os.environ.get("CLAN_NO_DYNAMIC_DEPS"):
return cmd
return [
*nix_command(["shell", "--inputs-from", f"{nixpkgs_flake()!s}"]),
*packages,
"-c",
*cmd,
]
# lazy loads list of allowed and static programs
class Programs:
allowed_programs = None
@@ -122,7 +135,7 @@ class Programs:
return program in cls.static_programs
# Alternative implementation of run_cmd() to replace run_cmd() at some point
# Alternative implementation of nix_shell() to replace nix_shell() at some point
# Features:
# - allow list for programs (need to be specified in allowed-programs.json)
# - be abe to compute a closure of all deps for testing
@@ -130,9 +143,7 @@ class Programs:
def run_cmd(programs: list[str], cmd: list[str]) -> list[str]:
for program in programs:
if not Programs.is_allowed(program):
raise ValueError(
f"Program {program} is not allowed because it is not part of clan_cli/nix/allowed-programs.json."
)
raise ValueError(f"Program not allowed: {program}")
if os.environ.get("IN_NIX_SANDBOX"):
return cmd
missing_packages = [

3
pkgs/clan-cli/clan_cli/nix/allowed-programs.json
View File

@@ -2,13 +2,10 @@
"age",
"bash",
"bubblewrap",
"disko",
"e2fsprogs",
"git",
"gnupg",
"mypy",
"nix",
"nixos-anywhere",
"openssh",
"qemu",
"rsync",

4
pkgs/clan-cli/clan_cli/secrets/import_sops.py
View File

@@ -11,7 +11,7 @@ from ..completions import (
complete_users,
)
from ..errors import ClanError
from ..nix import run_cmd
from ..nix import nix_shell
from .secrets import encrypt_secret, sops_secrets_folder
@@ -28,7 +28,7 @@ def import_sops(args: argparse.Namespace) -> None:
if args.input_type:
cmd += ["--input-type", args.input_type]
cmd += ["--output-type", "json", "--decrypt", args.sops_file]
cmd = run_cmd(["sops"], cmd)
cmd = nix_shell(["nixpkgs#sops"], cmd)
res = run(cmd, error_msg=f"Could not import sops file {file}")
secrets = json.loads(res.stdout)

18
pkgs/clan-cli/clan_cli/secrets/sops.py
View File

@@ -12,7 +12,7 @@ from typing import IO
from ..cmd import Log, run
from ..dirs import user_config_dir
from ..errors import ClanError
from ..nix import run_cmd
from ..nix import nix_shell
from .folders import sops_machines_folder, sops_users_folder
@@ -23,7 +23,7 @@ class SopsKey:
def get_public_key(privkey: str) -> str:
cmd = run_cmd(["age"], ["age-keygen", "-y"])
cmd = nix_shell(["nixpkgs#age"], ["age-keygen", "-y"])
try:
res = subprocess.run(
cmd, input=privkey, stdout=subprocess.PIPE, text=True, check=True
@@ -36,7 +36,7 @@ def get_public_key(privkey: str) -> str:
def generate_private_key(out_file: Path | None = None) -> tuple[str, str]:
cmd = run_cmd(["age"], ["age-keygen"])
cmd = nix_shell(["nixpkgs#age"], ["age-keygen"])
try:
proc = run(cmd)
res = proc.stdout.strip()
@@ -125,8 +125,8 @@ def update_keys(secret_path: Path, keys: list[str]) -> list[Path]:
with sops_manifest(keys) as manifest:
secret_path = secret_path / "secret"
time_before = secret_path.stat().st_mtime
cmd = run_cmd(
["sops"],
cmd = nix_shell(
["nixpkgs#sops"],
[
"sops",
"--config",
@@ -152,7 +152,7 @@ def encrypt_file(
if not content:
args = ["sops", "--config", str(manifest)]
args.extend([str(secret_path)])
cmd = run_cmd(["sops"], args)
cmd = nix_shell(["nixpkgs#sops"], args)
# Don't use our `run` here, because it breaks editor integration.
# We never need this in our UI.
p = subprocess.run(cmd, check=False)
@@ -180,7 +180,7 @@ def encrypt_file(
# we pass an empty manifest to pick up existing configuration of the user
args = ["sops", "--config", str(manifest)]
args.extend(["-i", "--encrypt", str(f.name)])
cmd = run_cmd(["sops"], args)
cmd = nix_shell(["nixpkgs#sops"], args)
run(cmd, log=Log.BOTH)
# atomic copy of the encrypted file
with NamedTemporaryFile(dir=folder, delete=False) as f2:
@@ -195,8 +195,8 @@ def encrypt_file(
def decrypt_file(secret_path: Path) -> str:
with sops_manifest([]) as manifest:
cmd = run_cmd(
["sops"],
cmd = nix_shell(
["nixpkgs#sops"],
["sops", "--config", str(manifest), "--decrypt", str(secret_path)],
)
res = run(cmd, error_msg=f"Could not decrypt {secret_path}")

14
pkgs/clan-cli/clan_cli/ssh/cli.py
View File

@@ -6,7 +6,7 @@ import socket
import subprocess
from pathlib import Path
from ..nix import run_cmd
from ..nix import nix_shell
log = logging.getLogger(__name__)
@@ -18,13 +18,13 @@ def ssh(
ssh_args: list[str] = [],
torify: bool = False,
) -> None:
packages = ["openssh"]
packages = ["nixpkgs#openssh"]
if torify:
packages.append("tor")
packages.append("nixpkgs#tor")
password_args = []
if password:
packages.append("sshpass")
packages.append("nixpkgs#sshpass")
password_args = [
"sshpass",
"-p",
@@ -48,15 +48,15 @@ def ssh(
if torify:
cmd_args.insert(0, "torify")
cmd = run_cmd(packages, cmd_args)
cmd = nix_shell(packages, cmd_args)
subprocess.run(cmd)
def qrcode_scan(picture_file: str) -> str:
return (
subprocess.run(
run_cmd(
["zbar"],
nix_shell(
["nixpkgs#zbar"],
[
"zbarimg",
"--quiet",

8
pkgs/clan-cli/clan_cli/vars/generate.py
View File

@@ -20,7 +20,7 @@ from ..errors import ClanError
from ..git import commit_files
from ..machines.inventory import get_all_machines, get_selected_machines
from ..machines.machines import Machine
from ..nix import run_cmd
from ..nix import nix_shell
from .check import check_secrets
from .public_modules import FactStoreBase
from .secret_modules import SecretStoreBase
@@ -40,10 +40,10 @@ def read_multiline_input(prompt: str = "Finish with Ctrl-D") -> str:
def bubblewrap_cmd(generator: str, generator_dir: Path, dep_tmpdir: Path) -> list[str]:
# fmt: off
return run_cmd(
return nix_shell(
[
"bash",
"bubblewrap",
"nixpkgs#bash",
"nixpkgs#bubblewrap",
],
[
"bwrap",

18
pkgs/clan-cli/clan_cli/vars/secret_modules/password_store.py
View File

@@ -3,7 +3,7 @@ import subprocess
from pathlib import Path
from clan_cli.machines.machines import Machine
from clan_cli.nix import run_cmd
from clan_cli.nix import nix_shell
from . import SecretStoreBase
@@ -16,8 +16,8 @@ class SecretStore(SecretStoreBase):
self, service: str, name: str, value: bytes, groups: list[str]
) -> Path | None:
subprocess.run(
run_cmd(
["pass"],
nix_shell(
["nixpkgs#pass"],
["pass", "insert", "-m", f"machines/{self.machine.name}/{name}"],
),
input=value,
@@ -27,8 +27,8 @@ class SecretStore(SecretStoreBase):
def get(self, service: str, name: str) -> bytes:
return subprocess.run(
run_cmd(
["pass"],
nix_shell(
["nixpkgs#pass"],
["pass", "show", f"machines/{self.machine.name}/{name}"],
),
check=True,
@@ -49,8 +49,8 @@ class SecretStore(SecretStoreBase):
hashes = []
hashes.append(
subprocess.run(
run_cmd(
["git"],
nix_shell(
["nixpkgs#git"],
[
"git",
"-C",
@@ -68,8 +68,8 @@ class SecretStore(SecretStoreBase):
if symlink.is_symlink():
hashes.append(
subprocess.run(
run_cmd(
["git"],
nix_shell(
["nixpkgs#git"],
[
"git",
"-C",

6
pkgs/clan-cli/clan_cli/vars/upload.py
View File

@@ -7,7 +7,7 @@ from tempfile import TemporaryDirectory
from ..cmd import Log, run
from ..completions import add_dynamic_completer, complete_machines
from ..machines.machines import Machine
from ..nix import run_cmd
from ..nix import nix_shell
log = logging.getLogger(__name__)
@@ -25,8 +25,8 @@ def upload_secrets(machine: Machine) -> None:
ssh_cmd = host.ssh_cmd()
run(
run_cmd(
["rsync"],
nix_shell(
["nixpkgs#rsync"],
[
"rsync",
"-e",

12
pkgs/clan-cli/clan_cli/vms/run.py
View File

@@ -13,7 +13,7 @@ from ..dirs import module_root, user_cache_dir, vm_state_dir
from ..errors import ClanError
from ..facts.generate import generate_facts
from ..machines.machines import Machine
from ..nix import run_cmd
from ..nix import nix_shell
from .inspect import VmConfig, inspect_vm
from .qemu import qemu_command
from .virtiofsd import start_virtiofsd
@@ -82,8 +82,8 @@ def prepare_disk(
file_name: str = "disk.img",
) -> Path:
disk_img = directory / file_name
cmd = run_cmd(
["qemu"],
cmd = nix_shell(
["nixpkgs#qemu"],
[
"qemu-img",
"create",
@@ -170,11 +170,11 @@ def run_vm(
qga_socket_file=qga_socket_file,
)
packages = ["qemu"]
packages = ["nixpkgs#qemu"]
env = os.environ.copy()
if vm.graphics and not vm.waypipe:
packages.append("virt-viewer")
packages.append("nixpkgs#virt-viewer")
remote_viewer_mimetypes = module_root() / "vms" / "mimetypes"
env["XDG_DATA_DIRS"] = (
f"{remote_viewer_mimetypes}:{env.get('XDG_DATA_DIRS', '')}"
@@ -185,7 +185,7 @@ def run_vm(
start_virtiofsd(virtiofsd_socket),
):
run(
run_cmd(packages, qemu_cmd.args),
nix_shell(packages, qemu_cmd.args),
env=env,
log=Log.BOTH,
error_msg=f"Could not start vm {machine}",

6
pkgs/clan-cli/clan_cli/vms/virtiofsd.py
View File

@@ -6,7 +6,7 @@ from collections.abc import Iterator
from pathlib import Path
from ..errors import ClanError
from ..nix import run_cmd
from ..nix import nix_shell
@contextlib.contextmanager
@@ -14,8 +14,8 @@ def start_virtiofsd(socket_path: Path) -> Iterator[None]:
sandbox = "namespace"
if shutil.which("newuidmap") is None:
sandbox = "none"
virtiofsd = run_cmd(
["virtiofsd"],
virtiofsd = nix_shell(
["nixpkgs#virtiofsd"],
[
"virtiofsd",
"--socket-path",

6
pkgs/clan-cli/clan_cli/vms/waypipe.py
View File

@@ -5,7 +5,7 @@ import time
from collections.abc import Iterator
from ..errors import ClanError
from ..nix import run_cmd
from ..nix import nix_shell
VMADDR_CID_HYPERVISOR = 2
@@ -25,8 +25,8 @@ def start_waypipe(cid: int | None, title_prefix: str) -> Iterator[None]:
if cid is None:
yield
return
waypipe = run_cmd(
["waypipe"],
waypipe = nix_shell(
["nixpkgs#waypipe"],
[
"waypipe",
"--vsock",

8
pkgs/clan-cli/tests/conftest.py
View File

@@ -4,7 +4,7 @@ from pathlib import Path
import pytest
from clan_cli.custom_logger import setup_logging
from clan_cli.nix import run_cmd
from clan_cli.nix import nix_shell
pytest_plugins = [
"temporary_dir",
@@ -32,12 +32,12 @@ def pytest_sessionstart(session: pytest.Session) -> None:
@pytest.fixture
def git_repo(tmp_path: Path) -> Path:
# initialize a git repository
cmd = run_cmd(["git"], ["git", "init"])
cmd = nix_shell(["nixpkgs#git"], ["git", "init"])
subprocess.run(cmd, cwd=tmp_path, check=True)
# set user.name and user.email
cmd = run_cmd(["git"], ["git", "config", "user.name", "test"])
cmd = nix_shell(["nixpkgs#git"], ["git", "config", "user.name", "test"])
subprocess.run(cmd, cwd=tmp_path, check=True)
cmd = run_cmd(["git"], ["git", "config", "user.email", "test@test.test"])
cmd = nix_shell(["nixpkgs#git"], ["git", "config", "user.email", "test@test.test"])
subprocess.run(cmd, cwd=tmp_path, check=True)
# return the path to the git repository
return tmp_path

10
pkgs/clan-cli/tests/test_secrets_password_store.py
View File

@@ -10,7 +10,7 @@ from clan_cli.clan_uri import FlakeId
from clan_cli.facts.secret_modules.password_store import SecretStore
from clan_cli.machines.facts import machine_get_fact
from clan_cli.machines.machines import Machine
from clan_cli.nix import run_cmd
from clan_cli.nix import nix_shell
from clan_cli.ssh import HostGroup
@@ -38,10 +38,14 @@ def test_upload_secret(
"""
)
subprocess.run(
run_cmd(["gnupg"], ["gpg", "--batch", "--gen-key", str(gpg_key_spec)]),
nix_shell(
["nixpkgs#gnupg"], ["gpg", "--batch", "--gen-key", str(gpg_key_spec)]
),
check=True,
)
subprocess.run(run_cmd(["pass"], ["pass", "init", "test@local"]), check=True)
subprocess.run(
nix_shell(["nixpkgs#pass"], ["pass", "init", "test@local"]), check=True
)
cli.run(["facts", "generate", "vm1"])
store = SecretStore(

4
pkgs/clan-cli/tests/test_vms_cli.py
View File

@@ -40,7 +40,6 @@ def run_vm_in_thread(machine_name: str) -> None:
# wait for qmp socket to exist
def wait_vm_up(state_dir: Path) -> None:
socket_file = state_dir / "qga.sock"
# don't lower this value -> trouble if test machine is slow
timeout: float = 100
while True:
if timeout <= 0:
@@ -56,8 +55,7 @@ def wait_vm_up(state_dir: Path) -> None:
# wait for vm to be down by checking if qga socket is down
def wait_vm_down(state_dir: Path) -> None:
socket_file = state_dir / "qga.sock"
# don't lower this value -> trouble if test machine is slow
timeout: float = 100
timeout: float = 300
while socket_file.exists():
if timeout <= 0:
raise TimeoutError(