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

Merge pull request 'Fix multiple bugs in 'clan networking' command' (#4389) from Qubasa/clan-core:deploy_network into main

Browse Source 
Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/4389
This commit is contained in:
Luis Hebendanz
2025-07-21 07:35:54 +00:00
parent aed48be645 4055508588
commit e3247d9c36
5 changed files with 298 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
clanServices/tor/default.nix
View File

@@ -8,7 +8,29 @@
"Network"
];
roles.default = {
roles.client = {
perInstance =
{
...
}:
{
nixosModule =
{
...
}:
{
config = {
services.tor = {
enable = true;
torsocks.enable = true;
client.enable = true;
};
};
};
};
};
roles.server = {
# interface =
# { lib, ... }:
# {
@@ -42,7 +64,7 @@
generator = "tor_${instanceName}";
file = "hostname";
};
}) roles.default.machines;
}) roles.server.machines;
};
nixosModule =
{

98
pkgs/clan-cli/clan_lib/import_utils/__init__.py Normal file
View File

@@ -0,0 +1,98 @@
import contextlib
import importlib
import inspect
from dataclasses import dataclass
from pathlib import Path
from typing import Any, TypeVar, cast
T = TypeVar("T")
@dataclass(frozen=True)
class ClassSource:
module_name: str
file_path: Path
object_name: str
line_number: int | None = None
def vscode_clickable_path(self) -> str:
"""Return a VSCode-clickable path for the class source."""
return (
f"{self.module_name}.{self.object_name}: {self.file_path}:{self.line_number}"
if self.line_number is not None
else f"{self.module_name}.{self.object_name}: {self.file_path}"
)
def __repr__(self) -> str:
return self.vscode_clickable_path()
def __str__(self) -> str:
return self.vscode_clickable_path()
def import_with_source[T](
module_name: str,
class_name: str,
base_class: type[T],
*args: Any,
**kwargs: Any,
) -> T:
"""
Import a class from a module and instantiate it with source information.
This function dynamically imports a class and adds source location metadata
that can be used for debugging. The instantiated object will have VSCode-clickable
paths in its string representation.
Args:
module_name: The fully qualified module name to import
class_name: The name of the class to import from the module
base_class: The base class type for type checking
*args: Additional positional arguments to pass to the class constructor
**kwargs: Additional keyword arguments to pass to the class constructor
Returns:
An instance of the imported class with source information
Example:
>>> from .network import NetworkTechnologyBase, ClassSource
>>> tech = import_with_source(
... "clan_lib.network.tor",
... "NetworkTechnology",
... NetworkTechnologyBase
... )
>>> print(tech) # Outputs: ~/Projects/clan-core/.../tor.py:7
"""
# Import the module
module = importlib.import_module(module_name)
# Get the class from the module
cls = getattr(module, class_name)
# Get the line number of the class definition
line_number = None
with contextlib.suppress(Exception):
line_number = inspect.getsourcelines(cls)[1]
# Get the file path
file_path_str = module.__file__
assert file_path_str is not None, f"Module {module_name} file path cannot be None"
# Make the path relative to home for better readability
try:
file_path = Path(file_path_str).relative_to(Path.home())
file_path = Path("~", file_path)
except ValueError:
# If not under home directory, use absolute path
file_path = Path(file_path_str)
# Create source information
source = ClassSource(
module_name=module_name,
file_path=file_path,
object_name=class_name,
line_number=line_number,
)
# Instantiate the class with source information
return cast(T, cls(source, *args, **kwargs))

145
pkgs/clan-cli/clan_lib/import_utils/import_utils_test.py Normal file
View File

@@ -0,0 +1,145 @@
import tempfile
from pathlib import Path
from textwrap import dedent
from typing import Any, cast
import pytest
from clan_lib.import_utils import import_with_source
from clan_lib.network.network import NetworkTechnologyBase
def test_import_with_source(tmp_path: Path) -> None:
"""Test importing a class with source information."""
# Create a temporary module file
module_dir = tmp_path / "test_module"
module_dir.mkdir()
# Create __init__.py to make it a package
(module_dir / "__init__.py").write_text("")
# Create a test module with a NetworkTechnology class
test_module_path = module_dir / "test_tech.py"
test_module_path.write_text(
dedent("""
from clan_lib.network.network import NetworkTechnologyBase
class NetworkTechnology(NetworkTechnologyBase):
def __init__(self, source):
super().__init__(source)
self.test_value = "test"
def is_running(self) -> bool:
return True
""")
)
# Add the temp directory to sys.path
import sys
sys.path.insert(0, str(tmp_path))
try:
# Import the class using import_with_source
instance = import_with_source(
"test_module.test_tech",
"NetworkTechnology",
cast(Any, NetworkTechnologyBase),
)
# Verify the instance is created correctly
assert isinstance(instance, NetworkTechnologyBase)
assert instance.is_running() is True
assert hasattr(instance, "test_value")
assert instance.test_value == "test"
# Verify source information
assert instance.source.module_name == "test_module.test_tech"
assert instance.source.file_path.name == "test_tech.py"
assert instance.source.object_name == "NetworkTechnology"
assert instance.source.line_number == 4 # Line where class is defined
# Test string representations
str_repr = str(instance)
assert "test_tech.py:" in str_repr
assert "NetworkTechnology" in str_repr
assert str(instance.source.line_number) in str_repr
repr_repr = repr(instance)
assert "NetworkTechnology" in repr_repr
assert "test_tech.py:" in repr_repr
assert "test_module.test_tech.NetworkTechnology" in repr_repr
finally:
# Clean up sys.path
sys.path.remove(str(tmp_path))
def test_import_with_source_with_args() -> None:
"""Test importing a class with additional constructor arguments."""
# Create a temporary test file
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
f.write(
dedent("""
from clan_lib.network.network import NetworkTechnologyBase
class NetworkTechnology(NetworkTechnologyBase):
def __init__(self, source, extra_arg, keyword_arg=None):
super().__init__(source)
self.extra_arg = extra_arg
self.keyword_arg = keyword_arg
def is_running(self) -> bool:
return False
""")
)
temp_file = Path(f.name)
# Import module dynamically
import importlib.util
import sys
spec = importlib.util.spec_from_file_location("temp_module", temp_file)
assert spec is not None
assert spec.loader is not None
module = importlib.util.module_from_spec(spec)
sys.modules["temp_module"] = module
spec.loader.exec_module(module)
try:
# Import with additional arguments
instance = import_with_source(
"temp_module",
"NetworkTechnology",
cast(Any, NetworkTechnologyBase),
"extra_value",
keyword_arg="keyword_value",
)
# Verify arguments were passed correctly
assert instance.extra_arg == "extra_value" # type: ignore[attr-defined]
assert instance.keyword_arg == "keyword_value" # type: ignore[attr-defined]
assert instance.source.object_name == "NetworkTechnology"
finally:
# Clean up
del sys.modules["temp_module"]
temp_file.unlink()
def test_import_with_source_module_not_found() -> None:
"""Test error handling when module is not found."""
with pytest.raises(ModuleNotFoundError):
import_with_source(
"non_existent_module", "SomeClass", cast(Any, NetworkTechnologyBase)
)
def test_import_with_source_class_not_found() -> None:
"""Test error handling when class is not found in module."""
with pytest.raises(AttributeError):
import_with_source(
"clan_lib.network.network",
"NonExistentClass",
cast(Any, NetworkTechnologyBase),
)

35
pkgs/clan-cli/clan_lib/network/network.py
View File

@@ -1,5 +1,5 @@
import importlib
import logging
import textwrap
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass
@@ -9,6 +9,7 @@ from typing import Any
from clan_cli.vars.get import get_machine_var
from clan_lib.errors import ClanError
from clan_lib.flake import Flake
from clan_lib.import_utils import ClassSource, import_with_source
from clan_lib.ssh.parse import parse_ssh_uri
from clan_lib.ssh.remote import Remote, check_machine_ssh_reachable
@@ -26,17 +27,33 @@ class Peer:
return self._host["plain"]
if "var" in self._host and isinstance(self._host["var"], dict):
_var: dict[str, str] = self._host["var"]
machine_name = _var["machine"]
generator = _var["generator"]
var = get_machine_var(
str(self.flake),
_var["machine"],
f"{_var['generator']}/{_var['file']}",
machine_name,
f"{generator}/{_var['file']}",
)
if not var.exists:
msg = (
textwrap.dedent(f"""
It looks like you added a networking module to your machine, but forgot
to deploy your changes. Please run "clan machines update {machine_name}"
so that the appropriate vars are generated and deployed properly.
""")
.rstrip("\n")
.lstrip("\n")
)
raise ClanError(msg)
return var.value.decode()
msg = f"Unknown Var Type {self._host}"
raise ClanError(msg)
@dataclass
class NetworkTechnologyBase(ABC):
source: ClassSource
@abstractmethod
def is_running(self) -> bool:
pass
@@ -70,8 +87,12 @@ class Network:
@cached_property
def module(self) -> NetworkTechnologyBase:
module = importlib.import_module(self.module_name)
return module.NetworkTechnology()
res = import_with_source(
self.module_name,
"NetworkTechnology",
NetworkTechnologyBase, # type: ignore[type-abstract]
)
return res
def is_running(self) -> bool:
return self.module.is_running()
@@ -117,7 +138,9 @@ def get_network_overview(networks: dict[str, Network]) -> dict:
result[network_name]["status"] = None
result[network_name]["peers"] = {}
network_online = False
if network.module.is_running():
module = network.module
log.debug(f"Using network module: {module}")
if module.is_running():
result[network_name]["status"] = True
network_online = True
for peer_name in network.peers:

4
pkgs/clan-cli/clan_lib/network/tor.py
View File

@@ -1,4 +1,4 @@
from urllib.error import URLError
from urllib.error import HTTPError
from urllib.request import urlopen
from .network import NetworkTechnologyBase
@@ -14,7 +14,7 @@ class NetworkTechnology(NetworkTechnologyBase):
response = urlopen("http://127.0.0.1:9050", timeout=5)
content = response.read().decode("utf-8", errors="ignore")
return "tor" in content.lower()
except URLError as e:
except HTTPError as e:
return "tor" in str(e).lower()
except Exception:
return False