introduce ClanError type

This commit is contained in:
Jörg Thalheim
2023-07-26 15:50:27 +02:00
parent b4ba9c70cd
commit 4101b9adb4
2 changed files with 12 additions and 5 deletions

View File

@@ -0,0 +1,4 @@
class ClanError(Exception):
"""Base class for exceptions in this module."""
pass

View File

@@ -8,6 +8,7 @@ from pathlib import Path
from tempfile import TemporaryDirectory from tempfile import TemporaryDirectory
from typing import Any, Iterator, Optional from typing import Any, Iterator, Optional
from ..errors import ClanError
from ..nix import nix_shell from ..nix import nix_shell
@@ -30,11 +31,11 @@ def try_connect_port(port: int) -> bool:
return result == 0 return result == 0
def find_free_port(port_range: range) -> int: def find_free_port(port_range: range) -> Optional[int]:
for port in port_range: for port in port_range:
if try_bind_port(port): if try_bind_port(port):
return port return port
raise Exception("cannot find a free port") return None
class ZerotierController: class ZerotierController:
@@ -86,6 +87,8 @@ class ZerotierController:
def zerotier_controller() -> Iterator[ZerotierController]: def zerotier_controller() -> Iterator[ZerotierController]:
# This check could be racy but it's unlikely in practice # This check could be racy but it's unlikely in practice
controller_port = find_free_port(range(10000, 65535)) controller_port = find_free_port(range(10000, 65535))
if controller_port is None:
raise ClanError("cannot find a free port for zerotier controller")
cmd = nix_shell(["bash", "zerotierone"], ["bash", "-c", "command -v zerotier-one"]) cmd = nix_shell(["bash", "zerotierone"], ["bash", "-c", "command -v zerotier-one"])
res = subprocess.run( res = subprocess.run(
cmd, cmd,
@@ -95,10 +98,10 @@ def zerotier_controller() -> Iterator[ZerotierController]:
) )
zerotier_exe = res.stdout.strip() zerotier_exe = res.stdout.strip()
if zerotier_exe is None: if zerotier_exe is None:
raise Exception("cannot find zerotier-one executable") raise ClanError("cannot find zerotier-one executable")
if not zerotier_exe.startswith("/nix/store"): if not zerotier_exe.startswith("/nix/store"):
raise Exception( raise ClanError(
f"zerotier-one executable needs to come from /nix/store: {zerotier_exe}" f"zerotier-one executable needs to come from /nix/store: {zerotier_exe}"
) )
@@ -136,7 +139,7 @@ def zerotier_controller() -> Iterator[ZerotierController]:
while not try_connect_port(controller_port): while not try_connect_port(controller_port):
status = p.poll() status = p.poll()
if status is not None: if status is not None:
raise Exception( raise ClanError(
f"zerotier-one has been terminated unexpected with {status}" f"zerotier-one has been terminated unexpected with {status}"
) )
time.sleep(0.1) time.sleep(0.1)