Merge pull request 'clan_cli: Rewrite ClanURI' (#914) from Qubasa-main into main
This commit is contained in:
commit
718c0a06e2
@ -5,13 +5,13 @@ import urllib.request
|
|||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from enum import Enum, member
|
from enum import Enum, member
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any, Self
|
from typing import Any
|
||||||
|
|
||||||
from .errors import ClanError
|
from .errors import ClanError
|
||||||
|
|
||||||
|
|
||||||
# Define an enum with different members that have different values
|
# Define an enum with different members that have different values
|
||||||
class ClanScheme(Enum):
|
class ClanUrl(Enum):
|
||||||
# Use the dataclass decorator to add fields and methods to the members
|
# Use the dataclass decorator to add fields and methods to the members
|
||||||
@member
|
@member
|
||||||
@dataclass
|
@dataclass
|
||||||
@ -19,7 +19,10 @@ class ClanScheme(Enum):
|
|||||||
url: str # The url field holds the HTTP URL
|
url: str # The url field holds the HTTP URL
|
||||||
|
|
||||||
def __str__(self) -> str:
|
def __str__(self) -> str:
|
||||||
return f"REMOTE({self.url})" # The __str__ method returns a custom string representation
|
return f"{self.url}" # The __str__ method returns a custom string representation
|
||||||
|
|
||||||
|
def __repr__(self) -> str:
|
||||||
|
return f"ClanUrl.REMOTE({self.url})"
|
||||||
|
|
||||||
@member
|
@member
|
||||||
@dataclass
|
@dataclass
|
||||||
@ -27,143 +30,145 @@ class ClanScheme(Enum):
|
|||||||
path: Path # The path field holds the local path
|
path: Path # The path field holds the local path
|
||||||
|
|
||||||
def __str__(self) -> str:
|
def __str__(self) -> str:
|
||||||
return f"LOCAL({self.path})" # The __str__ method returns a custom string representation
|
return f"{self.path}" # The __str__ method returns a custom string representation
|
||||||
|
|
||||||
|
def __repr__(self) -> str:
|
||||||
|
return f"ClanUrl.LOCAL({self.path})"
|
||||||
|
|
||||||
|
|
||||||
# Parameters defined here will be DELETED from the nested uri
|
# Parameters defined here will be DELETED from the nested uri
|
||||||
# so make sure there are no conflicts with other webservices
|
# so make sure there are no conflicts with other webservices
|
||||||
@dataclass
|
@dataclass
|
||||||
class ClanParameters:
|
class MachineParams:
|
||||||
flake_attr: str = "defaultVM"
|
dummy_opt: str = "dummy"
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class MachineData:
|
||||||
|
url: ClanUrl
|
||||||
|
name: str = "defaultVM"
|
||||||
|
params: MachineParams = dataclasses.field(default_factory=MachineParams)
|
||||||
|
|
||||||
|
def get_id(self) -> str:
|
||||||
|
return f"{self.url}#{self.name}"
|
||||||
|
|
||||||
|
|
||||||
# Define the ClanURI class
|
# Define the ClanURI class
|
||||||
class ClanURI:
|
class ClanURI:
|
||||||
|
_orig_uri: str
|
||||||
|
_nested_uri: str
|
||||||
|
_components: urllib.parse.ParseResult
|
||||||
|
url: ClanUrl
|
||||||
|
_machines: list[MachineData]
|
||||||
|
|
||||||
# Initialize the class with a clan:// URI
|
# Initialize the class with a clan:// URI
|
||||||
def __init__(self, uri: str) -> None:
|
def __init__(self, uri: str) -> None:
|
||||||
|
self._machines = []
|
||||||
|
|
||||||
# users might copy whitespace along with the uri
|
# users might copy whitespace along with the uri
|
||||||
uri = uri.strip()
|
uri = uri.strip()
|
||||||
self._full_uri = uri
|
self._orig_uri = uri
|
||||||
|
|
||||||
# Check if the URI starts with clan://
|
# Check if the URI starts with clan://
|
||||||
# If it does, remove the clan:// prefix
|
# If it does, remove the clan:// prefix
|
||||||
if uri.startswith("clan://"):
|
if uri.startswith("clan://"):
|
||||||
self._nested_uri = uri[7:]
|
self._nested_uri = uri[7:]
|
||||||
else:
|
else:
|
||||||
raise ClanError(f"Invalid scheme: expected clan://, got {uri}")
|
raise ClanError(f"Invalid uri: expected clan://, got {uri}")
|
||||||
|
|
||||||
# Parse the URI into components
|
# Parse the URI into components
|
||||||
# scheme://netloc/path;parameters?query#fragment
|
# url://netloc/path;parameters?query#fragment
|
||||||
self._components = urllib.parse.urlparse(self._nested_uri)
|
self._components = urllib.parse.urlparse(self._nested_uri)
|
||||||
|
|
||||||
# Parse the query string into a dictionary
|
# Replace the query string in the components with the new query string
|
||||||
query = urllib.parse.parse_qs(self._components.query)
|
clean_comps = self._components._replace(
|
||||||
|
query=self._components.query, fragment=""
|
||||||
|
)
|
||||||
|
|
||||||
# Create a new dictionary with only the parameters we want
|
# Parse the URL into a ClanUrl object
|
||||||
# example: https://example.com?flake_attr=myVM&password=1234
|
self.url = self._parse_url(clean_comps)
|
||||||
# becomes: https://example.com?password=1234
|
|
||||||
# clan_params = {"flake_attr": "myVM"}
|
# Parse the fragment into a list of machine queries
|
||||||
# query = {"password": ["1234"]}
|
# Then parse every machine query into a MachineParameters object
|
||||||
clan_params: dict[str, str] = {}
|
machine_frags = list(
|
||||||
for field in dataclasses.fields(ClanParameters):
|
filter(lambda x: len(x) > 0, self._components.fragment.split("#"))
|
||||||
if field.name in query:
|
)
|
||||||
values = query[field.name]
|
for machine_frag in machine_frags:
|
||||||
|
machine = self._parse_machine_query(machine_frag)
|
||||||
|
self._machines.append(machine)
|
||||||
|
|
||||||
|
# If there are no machine fragments, add a default machine
|
||||||
|
if len(machine_frags) == 0:
|
||||||
|
default_machine = MachineData(url=self.url)
|
||||||
|
self._machines.append(default_machine)
|
||||||
|
|
||||||
|
def _parse_url(self, comps: urllib.parse.ParseResult) -> ClanUrl:
|
||||||
|
comb = (
|
||||||
|
comps.scheme,
|
||||||
|
comps.netloc,
|
||||||
|
comps.path,
|
||||||
|
comps.params,
|
||||||
|
comps.query,
|
||||||
|
comps.fragment,
|
||||||
|
)
|
||||||
|
match comb:
|
||||||
|
case ("file", "", path, "", "", _) | ("", "", path, "", "", _): # type: ignore
|
||||||
|
url = ClanUrl.LOCAL.value(Path(path).expanduser().resolve()) # type: ignore
|
||||||
|
case _:
|
||||||
|
url = ClanUrl.REMOTE.value(comps.geturl()) # type: ignore
|
||||||
|
|
||||||
|
return url
|
||||||
|
|
||||||
|
def _parse_machine_query(self, machine_frag: str) -> MachineData:
|
||||||
|
comp = urllib.parse.urlparse(machine_frag)
|
||||||
|
query = urllib.parse.parse_qs(comp.query)
|
||||||
|
machine_name = comp.path
|
||||||
|
|
||||||
|
machine_params: dict[str, Any] = {}
|
||||||
|
for dfield in dataclasses.fields(MachineParams):
|
||||||
|
if dfield.name in query:
|
||||||
|
values = query[dfield.name]
|
||||||
if len(values) > 1:
|
if len(values) > 1:
|
||||||
raise ClanError(f"Multiple values for parameter: {field.name}")
|
raise ClanError(f"Multiple values for parameter: {dfield.name}")
|
||||||
clan_params[field.name] = values[0]
|
machine_params[dfield.name] = values[0]
|
||||||
|
|
||||||
# Remove the field from the query dictionary
|
# Remove the field from the query dictionary
|
||||||
# clan uri and nested uri share one namespace for query parameters
|
# clan uri and nested uri share one namespace for query parameters
|
||||||
# we need to make sure there are no conflicts
|
# we need to make sure there are no conflicts
|
||||||
del query[field.name]
|
del query[dfield.name]
|
||||||
# Reencode the query dictionary into a query string
|
params = MachineParams(**machine_params)
|
||||||
real_query = urllib.parse.urlencode(query, doseq=True)
|
machine = MachineData(url=self.url, name=machine_name, params=params)
|
||||||
|
return machine
|
||||||
|
|
||||||
# If the fragment contains a #, use the part after the # as the flake_attr
|
@property
|
||||||
# on multiple #, use the first one
|
def machine(self) -> MachineData:
|
||||||
if self._components.fragment != "":
|
return self._machines[0]
|
||||||
clan_params["flake_attr"] = self._components.fragment.split("#")[0]
|
|
||||||
|
|
||||||
# Replace the query string in the components with the new query string
|
def get_orig_uri(self) -> str:
|
||||||
self._components = self._components._replace(query=real_query, fragment="")
|
return self._orig_uri
|
||||||
|
|
||||||
# Create a ClanParameters object from the clan_params dictionary
|
def get_url(self) -> str:
|
||||||
self.params = ClanParameters(**clan_params)
|
return str(self.url)
|
||||||
|
|
||||||
comb = (
|
|
||||||
self._components.scheme,
|
|
||||||
self._components.netloc,
|
|
||||||
self._components.path,
|
|
||||||
self._components.params,
|
|
||||||
self._components.query,
|
|
||||||
self._components.fragment,
|
|
||||||
)
|
|
||||||
match comb:
|
|
||||||
case ("file", "", path, "", "", "") | ("", "", path, "", "", _): # type: ignore
|
|
||||||
self.scheme = ClanScheme.LOCAL.value(Path(path).expanduser().resolve()) # type: ignore
|
|
||||||
case _:
|
|
||||||
self.scheme = ClanScheme.REMOTE.value(self._components.geturl()) # type: ignore
|
|
||||||
|
|
||||||
def get_internal(self) -> str:
|
|
||||||
match self.scheme:
|
|
||||||
case ClanScheme.LOCAL.value(path):
|
|
||||||
return str(path)
|
|
||||||
case ClanScheme.REMOTE.value(url):
|
|
||||||
return url
|
|
||||||
case _:
|
|
||||||
raise ClanError(f"Unsupported uri components: {self.scheme}")
|
|
||||||
|
|
||||||
def get_full_uri(self) -> str:
|
|
||||||
return self._full_uri
|
|
||||||
|
|
||||||
def get_id(self) -> str:
|
|
||||||
return f"{self.get_internal()}#{self.params.flake_attr}"
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def from_path(
|
|
||||||
cls, # noqa
|
|
||||||
path: Path,
|
|
||||||
flake_attr: str | None = None,
|
|
||||||
params: dict[str, Any] | ClanParameters | None = None,
|
|
||||||
) -> Self:
|
|
||||||
return cls.from_str(str(path), flake_attr=flake_attr, params=params)
|
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_str(
|
def from_str(
|
||||||
cls, # noqa
|
cls, # noqa
|
||||||
url: str,
|
url: str,
|
||||||
flake_attr: str | None = None,
|
machine_name: str | None = None,
|
||||||
params: dict[str, Any] | ClanParameters | None = None,
|
) -> "ClanURI":
|
||||||
) -> Self:
|
clan_uri = ""
|
||||||
if flake_attr is not None and params is not None:
|
if not url.startswith("clan://"):
|
||||||
raise ClanError("flake_attr and params are mutually exclusive")
|
clan_uri += "clan://"
|
||||||
|
|
||||||
prefix = "clan://"
|
clan_uri += url
|
||||||
if url.startswith(prefix):
|
|
||||||
url = url[len(prefix) :]
|
|
||||||
|
|
||||||
if params is None and flake_attr is None:
|
if machine_name:
|
||||||
return cls(f"clan://{url}")
|
clan_uri += f"#{machine_name}"
|
||||||
|
|
||||||
comp = urllib.parse.urlparse(url)
|
return cls(clan_uri)
|
||||||
query = urllib.parse.parse_qs(comp.query)
|
|
||||||
|
|
||||||
if isinstance(params, dict):
|
|
||||||
query.update(params)
|
|
||||||
elif isinstance(params, ClanParameters):
|
|
||||||
query.update(params.__dict__)
|
|
||||||
elif flake_attr is not None:
|
|
||||||
query["flake_attr"] = [flake_attr]
|
|
||||||
else:
|
|
||||||
raise ClanError(f"Unsupported params type: {type(params)}")
|
|
||||||
|
|
||||||
new_query = urllib.parse.urlencode(query, doseq=True)
|
|
||||||
comp = comp._replace(query=new_query)
|
|
||||||
new_url = urllib.parse.urlunparse(comp)
|
|
||||||
return cls(f"clan://{new_url}")
|
|
||||||
|
|
||||||
def __str__(self) -> str:
|
def __str__(self) -> str:
|
||||||
return self.get_full_uri()
|
return self.get_orig_uri()
|
||||||
|
|
||||||
def __repr__(self) -> str:
|
def __repr__(self) -> str:
|
||||||
return f"ClanURI({self.get_full_uri()})"
|
return f"ClanURI({self})"
|
||||||
|
@ -79,8 +79,8 @@ def new_history_entry(url: str, machine: str) -> HistoryEntry:
|
|||||||
def add_all_to_history(uri: ClanURI) -> list[HistoryEntry]:
|
def add_all_to_history(uri: ClanURI) -> list[HistoryEntry]:
|
||||||
history = list_history()
|
history = list_history()
|
||||||
new_entries: list[HistoryEntry] = []
|
new_entries: list[HistoryEntry] = []
|
||||||
for machine in list_machines(uri.get_internal()):
|
for machine in list_machines(uri.get_url()):
|
||||||
new_entry = _add_maschine_to_history_list(uri.get_internal(), machine, history)
|
new_entry = _add_maschine_to_history_list(uri.get_url(), machine, history)
|
||||||
new_entries.append(new_entry)
|
new_entries.append(new_entry)
|
||||||
write_history_file(history)
|
write_history_file(history)
|
||||||
return new_entries
|
return new_entries
|
||||||
@ -89,9 +89,7 @@ def add_all_to_history(uri: ClanURI) -> list[HistoryEntry]:
|
|||||||
def add_history(uri: ClanURI) -> HistoryEntry:
|
def add_history(uri: ClanURI) -> HistoryEntry:
|
||||||
user_history_file().parent.mkdir(parents=True, exist_ok=True)
|
user_history_file().parent.mkdir(parents=True, exist_ok=True)
|
||||||
history = list_history()
|
history = list_history()
|
||||||
new_entry = _add_maschine_to_history_list(
|
new_entry = _add_maschine_to_history_list(uri.get_url(), uri.machine.name, history)
|
||||||
uri.get_internal(), uri.params.flake_attr, history
|
|
||||||
)
|
|
||||||
write_history_file(history)
|
write_history_file(history)
|
||||||
return new_entry
|
return new_entry
|
||||||
|
|
||||||
@ -121,9 +119,7 @@ def add_history_command(args: argparse.Namespace) -> None:
|
|||||||
|
|
||||||
# takes a (sub)parser and configures it
|
# takes a (sub)parser and configures it
|
||||||
def register_add_parser(parser: argparse.ArgumentParser) -> None:
|
def register_add_parser(parser: argparse.ArgumentParser) -> None:
|
||||||
parser.add_argument(
|
parser.add_argument("uri", type=ClanURI, help="Path to the flake", default=".")
|
||||||
"uri", type=ClanURI.from_str, help="Path to the flake", default="."
|
|
||||||
)
|
|
||||||
parser.add_argument(
|
parser.add_argument(
|
||||||
"--all", help="Add all machines", default=False, action="store_true"
|
"--all", help="Add all machines", default=False, action="store_true"
|
||||||
)
|
)
|
||||||
|
@ -4,7 +4,7 @@ import datetime
|
|||||||
|
|
||||||
from clan_cli.flakes.inspect import inspect_flake
|
from clan_cli.flakes.inspect import inspect_flake
|
||||||
|
|
||||||
from ..clan_uri import ClanParameters, ClanURI
|
from ..clan_uri import ClanURI
|
||||||
from ..errors import ClanCmdError
|
from ..errors import ClanCmdError
|
||||||
from ..locked_open import write_history_file
|
from ..locked_open import write_history_file
|
||||||
from ..nix import nix_metadata
|
from ..nix import nix_metadata
|
||||||
@ -28,9 +28,9 @@ def update_history() -> list[HistoryEntry]:
|
|||||||
)
|
)
|
||||||
uri = ClanURI.from_str(
|
uri = ClanURI.from_str(
|
||||||
url=str(entry.flake.flake_url),
|
url=str(entry.flake.flake_url),
|
||||||
params=ClanParameters(entry.flake.flake_attr),
|
machine_name=entry.flake.flake_attr,
|
||||||
)
|
)
|
||||||
flake = inspect_flake(uri.get_internal(), uri.params.flake_attr)
|
flake = inspect_flake(uri.get_url(), uri.machine.name)
|
||||||
flake.flake_url = str(flake.flake_url)
|
flake.flake_url = str(flake.flake_url)
|
||||||
entry = HistoryEntry(
|
entry = HistoryEntry(
|
||||||
flake=flake, last_used=datetime.datetime.now().isoformat()
|
flake=flake, last_used=datetime.datetime.now().isoformat()
|
||||||
|
@ -1,28 +1,28 @@
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
from clan_cli.clan_uri import ClanParameters, ClanScheme, ClanURI
|
from clan_cli.clan_uri import ClanURI, ClanUrl
|
||||||
|
|
||||||
|
|
||||||
def test_get_internal() -> None:
|
def test_get_url() -> None:
|
||||||
# Create a ClanURI object from a remote URI with parameters
|
# Create a ClanURI object from a remote URI with parameters
|
||||||
uri = ClanURI("clan://https://example.com?flake_attr=myVM&password=1234")
|
uri = ClanURI("clan://https://example.com?password=1234#myVM")
|
||||||
assert uri.get_internal() == "https://example.com?password=1234"
|
assert uri.get_url() == "https://example.com?password=1234"
|
||||||
|
|
||||||
uri = ClanURI("clan://~/Downloads")
|
uri = ClanURI("clan://~/Downloads")
|
||||||
assert uri.get_internal().endswith("/Downloads")
|
assert uri.get_url().endswith("/Downloads")
|
||||||
|
|
||||||
uri = ClanURI("clan:///home/user/Downloads")
|
uri = ClanURI("clan:///home/user/Downloads")
|
||||||
assert uri.get_internal() == "/home/user/Downloads"
|
assert uri.get_url() == "/home/user/Downloads"
|
||||||
|
|
||||||
uri = ClanURI("clan://file:///home/user/Downloads")
|
uri = ClanURI("clan://file:///home/user/Downloads")
|
||||||
assert uri.get_internal() == "/home/user/Downloads"
|
assert uri.get_url() == "/home/user/Downloads"
|
||||||
|
|
||||||
|
|
||||||
def test_local_uri() -> None:
|
def test_local_uri() -> None:
|
||||||
# Create a ClanURI object from a local URI
|
# Create a ClanURI object from a local URI
|
||||||
uri = ClanURI("clan://file:///home/user/Downloads")
|
uri = ClanURI("clan://file:///home/user/Downloads")
|
||||||
match uri.scheme:
|
match uri.url:
|
||||||
case ClanScheme.LOCAL.value(path):
|
case ClanUrl.LOCAL.value(path):
|
||||||
assert path == Path("/home/user/Downloads") # type: ignore
|
assert path == Path("/home/user/Downloads") # type: ignore
|
||||||
case _:
|
case _:
|
||||||
assert False
|
assert False
|
||||||
@ -32,8 +32,8 @@ def test_is_remote() -> None:
|
|||||||
# Create a ClanURI object from a remote URI
|
# Create a ClanURI object from a remote URI
|
||||||
uri = ClanURI("clan://https://example.com")
|
uri = ClanURI("clan://https://example.com")
|
||||||
|
|
||||||
match uri.scheme:
|
match uri.url:
|
||||||
case ClanScheme.REMOTE.value(url):
|
case ClanUrl.REMOTE.value(url):
|
||||||
assert url == "https://example.com" # type: ignore
|
assert url == "https://example.com" # type: ignore
|
||||||
case _:
|
case _:
|
||||||
assert False
|
assert False
|
||||||
@ -42,104 +42,87 @@ def test_is_remote() -> None:
|
|||||||
def test_direct_local_path() -> None:
|
def test_direct_local_path() -> None:
|
||||||
# Create a ClanURI object from a remote URI
|
# Create a ClanURI object from a remote URI
|
||||||
uri = ClanURI("clan://~/Downloads")
|
uri = ClanURI("clan://~/Downloads")
|
||||||
assert uri.get_internal().endswith("/Downloads")
|
assert uri.get_url().endswith("/Downloads")
|
||||||
|
|
||||||
|
|
||||||
def test_direct_local_path2() -> None:
|
def test_direct_local_path2() -> None:
|
||||||
# Create a ClanURI object from a remote URI
|
# Create a ClanURI object from a remote URI
|
||||||
uri = ClanURI("clan:///home/user/Downloads")
|
uri = ClanURI("clan:///home/user/Downloads")
|
||||||
assert uri.get_internal() == "/home/user/Downloads"
|
assert uri.get_url() == "/home/user/Downloads"
|
||||||
|
|
||||||
|
|
||||||
def test_remote_with_clanparams() -> None:
|
def test_remote_with_clanparams() -> None:
|
||||||
# Create a ClanURI object from a remote URI with parameters
|
# Create a ClanURI object from a remote URI with parameters
|
||||||
uri = ClanURI("clan://https://example.com")
|
uri = ClanURI("clan://https://example.com")
|
||||||
|
|
||||||
assert uri.params.flake_attr == "defaultVM"
|
assert uri.machine.name == "defaultVM"
|
||||||
|
|
||||||
match uri.scheme:
|
match uri.url:
|
||||||
case ClanScheme.REMOTE.value(url):
|
case ClanUrl.REMOTE.value(url):
|
||||||
assert url == "https://example.com" # type: ignore
|
assert url == "https://example.com" # type: ignore
|
||||||
case _:
|
case _:
|
||||||
assert False
|
assert False
|
||||||
|
|
||||||
|
|
||||||
def test_from_path_with_custom() -> None:
|
|
||||||
# Create a ClanURI object from a remote URI with parameters
|
|
||||||
uri_str = Path("/home/user/Downloads")
|
|
||||||
params = ClanParameters(flake_attr="myVM")
|
|
||||||
uri = ClanURI.from_path(uri_str, params=params)
|
|
||||||
assert uri.params.flake_attr == "myVM"
|
|
||||||
|
|
||||||
match uri.scheme:
|
|
||||||
case ClanScheme.LOCAL.value(path):
|
|
||||||
assert path == Path("/home/user/Downloads") # type: ignore
|
|
||||||
case _:
|
|
||||||
assert False
|
|
||||||
|
|
||||||
|
|
||||||
def test_from_path_with_default() -> None:
|
|
||||||
# Create a ClanURI object from a remote URI with parameters
|
|
||||||
uri_str = Path("/home/user/Downloads")
|
|
||||||
params = ClanParameters()
|
|
||||||
uri = ClanURI.from_path(uri_str, params=params)
|
|
||||||
assert uri.params.flake_attr == "defaultVM"
|
|
||||||
|
|
||||||
match uri.scheme:
|
|
||||||
case ClanScheme.LOCAL.value(path):
|
|
||||||
assert path == Path("/home/user/Downloads") # type: ignore
|
|
||||||
case _:
|
|
||||||
assert False
|
|
||||||
|
|
||||||
|
|
||||||
def test_from_str() -> None:
|
|
||||||
# Create a ClanURI object from a remote URI with parameters
|
|
||||||
uri_str = "https://example.com?password=asdasd&test=1234"
|
|
||||||
params = ClanParameters(flake_attr="myVM")
|
|
||||||
uri = ClanURI.from_str(url=uri_str, params=params)
|
|
||||||
assert uri.params.flake_attr == "myVM"
|
|
||||||
|
|
||||||
match uri.scheme:
|
|
||||||
case ClanScheme.REMOTE.value(url):
|
|
||||||
assert url == "https://example.com?password=asdasd&test=1234" # type: ignore
|
|
||||||
case _:
|
|
||||||
assert False
|
|
||||||
|
|
||||||
uri = ClanURI.from_str(url=uri_str, params={"flake_attr": "myVM"})
|
|
||||||
assert uri.params.flake_attr == "myVM"
|
|
||||||
|
|
||||||
uri = ClanURI.from_str(uri_str, "myVM")
|
|
||||||
assert uri.params.flake_attr == "myVM"
|
|
||||||
|
|
||||||
uri_str = "~/Downloads/democlan"
|
|
||||||
params = ClanParameters(flake_attr="myVM")
|
|
||||||
uri = ClanURI.from_str(url=uri_str, params=params)
|
|
||||||
assert uri.params.flake_attr == "myVM"
|
|
||||||
assert uri.get_internal().endswith("/Downloads/democlan")
|
|
||||||
|
|
||||||
uri_str = "~/Downloads/democlan"
|
|
||||||
uri = ClanURI.from_str(url=uri_str)
|
|
||||||
assert uri.params.flake_attr == "defaultVM"
|
|
||||||
assert uri.get_internal().endswith("/Downloads/democlan")
|
|
||||||
|
|
||||||
uri_str = "clan://~/Downloads/democlan"
|
|
||||||
uri = ClanURI.from_str(url=uri_str)
|
|
||||||
assert uri.params.flake_attr == "defaultVM"
|
|
||||||
assert uri.get_internal().endswith("/Downloads/democlan")
|
|
||||||
|
|
||||||
|
|
||||||
def test_remote_with_all_params() -> None:
|
def test_remote_with_all_params() -> None:
|
||||||
# Create a ClanURI object from a remote URI with parameters
|
uri = ClanURI("clan://https://example.com?password=12345#myVM#secondVM?dummy_opt=1")
|
||||||
uri = ClanURI("clan://https://example.com?flake_attr=myVM&password=1234")
|
assert uri.machine.name == "myVM"
|
||||||
assert uri.params.flake_attr == "myVM"
|
assert uri._machines[1].name == "secondVM"
|
||||||
|
assert uri._machines[1].params.dummy_opt == "1"
|
||||||
match uri.scheme:
|
match uri.url:
|
||||||
case ClanScheme.REMOTE.value(url):
|
case ClanUrl.REMOTE.value(url):
|
||||||
assert url == "https://example.com?password=1234" # type: ignore
|
assert url == "https://example.com?password=12345" # type: ignore
|
||||||
case _:
|
case _:
|
||||||
assert False
|
assert False
|
||||||
|
|
||||||
|
|
||||||
def test_with_hashtag() -> None:
|
def test_from_str_remote() -> None:
|
||||||
uri = ClanURI("clan://https://example.com?flake_attr=thirdVM#myVM#secondVM")
|
uri = ClanURI.from_str(url="https://example.com", machine_name="myVM")
|
||||||
assert uri.params.flake_attr == "myVM"
|
assert uri.get_url() == "https://example.com"
|
||||||
|
assert uri.get_orig_uri() == "clan://https://example.com#myVM"
|
||||||
|
assert uri.machine.name == "myVM"
|
||||||
|
assert len(uri._machines) == 1
|
||||||
|
match uri.url:
|
||||||
|
case ClanUrl.REMOTE.value(url):
|
||||||
|
assert url == "https://example.com" # type: ignore
|
||||||
|
case _:
|
||||||
|
assert False
|
||||||
|
|
||||||
|
|
||||||
|
def test_from_str_local() -> None:
|
||||||
|
uri = ClanURI.from_str(url="~/Projects/democlan", machine_name="myVM")
|
||||||
|
assert uri.get_url().endswith("/Projects/democlan")
|
||||||
|
assert uri.get_orig_uri() == "clan://~/Projects/democlan#myVM"
|
||||||
|
assert uri.machine.name == "myVM"
|
||||||
|
assert len(uri._machines) == 1
|
||||||
|
match uri.url:
|
||||||
|
case ClanUrl.LOCAL.value(path):
|
||||||
|
assert str(path).endswith("/Projects/democlan") # type: ignore
|
||||||
|
case _:
|
||||||
|
assert False
|
||||||
|
|
||||||
|
|
||||||
|
def test_from_str_local_no_machine() -> None:
|
||||||
|
uri = ClanURI.from_str("~/Projects/democlan")
|
||||||
|
assert uri.get_url().endswith("/Projects/democlan")
|
||||||
|
assert uri.get_orig_uri() == "clan://~/Projects/democlan"
|
||||||
|
assert uri.machine.name == "defaultVM"
|
||||||
|
assert len(uri._machines) == 1
|
||||||
|
match uri.url:
|
||||||
|
case ClanUrl.LOCAL.value(path):
|
||||||
|
assert str(path).endswith("/Projects/democlan") # type: ignore
|
||||||
|
case _:
|
||||||
|
assert False
|
||||||
|
|
||||||
|
|
||||||
|
def test_from_str_local_no_machine2() -> None:
|
||||||
|
uri = ClanURI.from_str("~/Projects/democlan#syncthing-peer1")
|
||||||
|
assert uri.get_url().endswith("/Projects/democlan")
|
||||||
|
assert uri.get_orig_uri() == "clan://~/Projects/democlan#syncthing-peer1"
|
||||||
|
assert uri.machine.name == "syncthing-peer1"
|
||||||
|
assert len(uri._machines) == 1
|
||||||
|
match uri.url:
|
||||||
|
case ClanUrl.LOCAL.value(path):
|
||||||
|
assert str(path).endswith("/Projects/democlan") # type: ignore
|
||||||
|
case _:
|
||||||
|
assert False
|
||||||
|
@ -6,7 +6,7 @@ from cli import Cli
|
|||||||
from fixtures_flakes import FlakeForTest
|
from fixtures_flakes import FlakeForTest
|
||||||
from pytest import CaptureFixture
|
from pytest import CaptureFixture
|
||||||
|
|
||||||
from clan_cli.clan_uri import ClanParameters, ClanURI
|
from clan_cli.clan_uri import ClanURI
|
||||||
from clan_cli.dirs import user_history_file
|
from clan_cli.dirs import user_history_file
|
||||||
from clan_cli.history.add import HistoryEntry
|
from clan_cli.history.add import HistoryEntry
|
||||||
|
|
||||||
@ -19,8 +19,7 @@ def test_history_add(
|
|||||||
test_flake_with_core: FlakeForTest,
|
test_flake_with_core: FlakeForTest,
|
||||||
) -> None:
|
) -> None:
|
||||||
cli = Cli()
|
cli = Cli()
|
||||||
params = ClanParameters(flake_attr="vm1")
|
uri = ClanURI.from_str(str(test_flake_with_core.path), "vm1")
|
||||||
uri = ClanURI.from_path(test_flake_with_core.path, params=params)
|
|
||||||
cmd = [
|
cmd = [
|
||||||
"history",
|
"history",
|
||||||
"add",
|
"add",
|
||||||
@ -40,8 +39,7 @@ def test_history_list(
|
|||||||
test_flake_with_core: FlakeForTest,
|
test_flake_with_core: FlakeForTest,
|
||||||
) -> None:
|
) -> None:
|
||||||
cli = Cli()
|
cli = Cli()
|
||||||
params = ClanParameters(flake_attr="vm1")
|
uri = ClanURI.from_str(str(test_flake_with_core.path), "vm1")
|
||||||
uri = ClanURI.from_path(test_flake_with_core.path, params=params)
|
|
||||||
cmd = [
|
cmd = [
|
||||||
"history",
|
"history",
|
||||||
"list",
|
"list",
|
||||||
|
@ -13,7 +13,7 @@ from typing import IO, ClassVar
|
|||||||
|
|
||||||
import gi
|
import gi
|
||||||
from clan_cli import vms
|
from clan_cli import vms
|
||||||
from clan_cli.clan_uri import ClanScheme, ClanURI
|
from clan_cli.clan_uri import ClanURI, ClanUrl
|
||||||
from clan_cli.history.add import HistoryEntry
|
from clan_cli.history.add import HistoryEntry
|
||||||
from clan_cli.machines.machines import Machine
|
from clan_cli.machines.machines import Machine
|
||||||
|
|
||||||
@ -116,12 +116,12 @@ class VMObject(GObject.Object):
|
|||||||
url=self.data.flake.flake_url, flake_attr=self.data.flake.flake_attr
|
url=self.data.flake.flake_url, flake_attr=self.data.flake.flake_attr
|
||||||
)
|
)
|
||||||
match uri.scheme:
|
match uri.scheme:
|
||||||
case ClanScheme.LOCAL.value(path):
|
case ClanUrl.LOCAL.value(path):
|
||||||
self.machine = Machine(
|
self.machine = Machine(
|
||||||
name=self.data.flake.flake_attr,
|
name=self.data.flake.flake_attr,
|
||||||
flake=path, # type: ignore
|
flake=path, # type: ignore
|
||||||
)
|
)
|
||||||
case ClanScheme.REMOTE.value(url):
|
case ClanUrl.REMOTE.value(url):
|
||||||
self.machine = Machine(
|
self.machine = Machine(
|
||||||
name=self.data.flake.flake_attr,
|
name=self.data.flake.flake_attr,
|
||||||
flake=url, # type: ignore
|
flake=url, # type: ignore
|
||||||
|
@ -62,8 +62,8 @@ class JoinList:
|
|||||||
cls._instance = cls.__new__(cls)
|
cls._instance = cls.__new__(cls)
|
||||||
cls.list_store = Gio.ListStore.new(JoinValue)
|
cls.list_store = Gio.ListStore.new(JoinValue)
|
||||||
|
|
||||||
# Rerendering the join list every time an item changes in the clan_store
|
|
||||||
ClanStore.use().register_on_deep_change(cls._instance._rerender_join_list)
|
ClanStore.use().register_on_deep_change(cls._instance._rerender_join_list)
|
||||||
|
|
||||||
return cls._instance
|
return cls._instance
|
||||||
|
|
||||||
def _rerender_join_list(
|
def _rerender_join_list(
|
||||||
@ -83,7 +83,9 @@ class JoinList:
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
value = JoinValue(uri)
|
value = JoinValue(uri)
|
||||||
if value.url.get_id() in [item.url.get_id() for item in self.list_store]:
|
if value.url.machine.get_id() in [
|
||||||
|
item.url.machine.get_id() for item in self.list_store
|
||||||
|
]:
|
||||||
log.info(f"Join request already exists: {value.url}. Ignoring.")
|
log.info(f"Join request already exists: {value.url}. Ignoring.")
|
||||||
return
|
return
|
||||||
|
|
||||||
|
@ -57,7 +57,7 @@ class ClanStore:
|
|||||||
store: "GKVStore", position: int, removed: int, added: int
|
store: "GKVStore", position: int, removed: int, added: int
|
||||||
) -> None:
|
) -> None:
|
||||||
if added > 0:
|
if added > 0:
|
||||||
store.register_on_change(on_vmstore_change)
|
store.values()[position].register_on_change(on_vmstore_change)
|
||||||
callback(store, position, removed, added)
|
callback(store, position, removed, added)
|
||||||
|
|
||||||
self.clan_store.register_on_change(on_clanstore_change)
|
self.clan_store.register_on_change(on_clanstore_change)
|
||||||
@ -111,10 +111,11 @@ class ClanStore:
|
|||||||
del self.clan_store[vm.data.flake.flake_url][vm.data.flake.flake_attr]
|
del self.clan_store[vm.data.flake.flake_url][vm.data.flake.flake_attr]
|
||||||
|
|
||||||
def get_vm(self, uri: ClanURI) -> None | VMObject:
|
def get_vm(self, uri: ClanURI) -> None | VMObject:
|
||||||
clan = self.clan_store.get(uri.get_internal())
|
vm_store = self.clan_store.get(str(uri.url))
|
||||||
if clan is None:
|
if vm_store is None:
|
||||||
return None
|
return None
|
||||||
return clan.get(uri.params.flake_attr, None)
|
machine = vm_store.get(uri.machine.name, None)
|
||||||
|
return machine
|
||||||
|
|
||||||
def get_running_vms(self) -> list[VMObject]:
|
def get_running_vms(self) -> list[VMObject]:
|
||||||
return [
|
return [
|
||||||
|
@ -190,8 +190,8 @@ class ClanList(Gtk.Box):
|
|||||||
log.debug("Rendering join row for %s", join_val.url)
|
log.debug("Rendering join row for %s", join_val.url)
|
||||||
|
|
||||||
row = Adw.ActionRow()
|
row = Adw.ActionRow()
|
||||||
row.set_title(join_val.url.params.flake_attr)
|
row.set_title(join_val.url.machine.name)
|
||||||
row.set_subtitle(join_val.url.get_internal())
|
row.set_subtitle(str(join_val.url))
|
||||||
row.add_css_class("trust")
|
row.add_css_class("trust")
|
||||||
|
|
||||||
vm = ClanStore.use().get_vm(join_val.url)
|
vm = ClanStore.use().get_vm(join_val.url)
|
||||||
@ -204,7 +204,7 @@ class ClanList(Gtk.Box):
|
|||||||
)
|
)
|
||||||
|
|
||||||
avatar = Adw.Avatar()
|
avatar = Adw.Avatar()
|
||||||
avatar.set_text(str(join_val.url.params.flake_attr))
|
avatar.set_text(str(join_val.url.machine.name))
|
||||||
avatar.set_show_initials(True)
|
avatar.set_show_initials(True)
|
||||||
avatar.set_size(50)
|
avatar.set_size(50)
|
||||||
row.add_prefix(avatar)
|
row.add_prefix(avatar)
|
||||||
@ -229,7 +229,7 @@ class ClanList(Gtk.Box):
|
|||||||
|
|
||||||
def on_join_request(self, source: Any, url: str) -> None:
|
def on_join_request(self, source: Any, url: str) -> None:
|
||||||
log.debug("Join request: %s", url)
|
log.debug("Join request: %s", url)
|
||||||
clan_uri = ClanURI.from_str(url)
|
clan_uri = ClanURI(url)
|
||||||
JoinList.use().push(clan_uri, self.on_after_join)
|
JoinList.use().push(clan_uri, self.on_after_join)
|
||||||
|
|
||||||
def on_after_join(self, source: JoinValue) -> None:
|
def on_after_join(self, source: JoinValue) -> None:
|
||||||
|
Loading…
Reference in New Issue
Block a user