From 93c868a3b7067f11885afab589c05e141efe6fbf Mon Sep 17 00:00:00 2001 From: Qubasa Date: Thu, 7 Mar 2024 02:24:36 +0700 Subject: [PATCH 1/2] clan_cli: Rewrite ClanURI --- pkgs/clan-cli/clan_cli/clan_uri.py | 185 ++++++++++------------- pkgs/clan-cli/clan_cli/history/update.py | 6 +- pkgs/clan-cli/tests/test_clan_uri.py | 114 +++----------- 3 files changed, 107 insertions(+), 198 deletions(-) diff --git a/pkgs/clan-cli/clan_cli/clan_uri.py b/pkgs/clan-cli/clan_cli/clan_uri.py index 184f9c9f..032a4922 100644 --- a/pkgs/clan-cli/clan_cli/clan_uri.py +++ b/pkgs/clan-cli/clan_cli/clan_uri.py @@ -5,13 +5,13 @@ import urllib.request from dataclasses import dataclass from enum import Enum, member from pathlib import Path -from typing import Any, Self +from typing import Any from .errors import ClanError # 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 @member @dataclass @@ -33,137 +33,116 @@ class ClanScheme(Enum): # Parameters defined here will be DELETED from the nested uri # so make sure there are no conflicts with other webservices @dataclass -class ClanParameters: - flake_attr: str = "defaultVM" +class MachineParams: + dummy_opt: str = "dummy" + + +@dataclass +class MachineData: + name: str = "defaultVM" + params: MachineParams = dataclasses.field(default_factory=MachineParams) # Define the ClanURI class class ClanURI: + _orig_uri: str + _nested_uri: str + _components: urllib.parse.ParseResult + url: ClanUrl + machines: list[MachineData] + # Initialize the class with a clan:// URI def __init__(self, uri: str) -> None: + self.machines = [] + # users might copy whitespace along with the uri uri = uri.strip() - self._full_uri = uri + self._orig_uri = uri # Check if the URI starts with clan:// # If it does, remove the clan:// prefix if uri.startswith("clan://"): self._nested_uri = uri[7:] else: - raise ClanError(f"Invalid scheme: expected clan://, got {uri}") + raise ClanError(f"Invalid uri: expected clan://, got {uri}") # 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) - # Parse the query string into a dictionary - query = urllib.parse.parse_qs(self._components.query) + # Replace the query string in the components with the new query string + clean_comps = self._components._replace( + query=self._components.query, fragment="" + ) - # Create a new dictionary with only the parameters we want - # example: https://example.com?flake_attr=myVM&password=1234 - # becomes: https://example.com?password=1234 - # clan_params = {"flake_attr": "myVM"} - # query = {"password": ["1234"]} - clan_params: dict[str, str] = {} - for field in dataclasses.fields(ClanParameters): - if field.name in query: - values = query[field.name] + # Parse the URL into a ClanUrl object + self.url = self._parse_url(clean_comps) + + # Parse the fragment into a list of machine queries + # Then parse every machine query into a MachineParameters object + machine_frags = list( + filter(lambda x: len(x) > 0, self._components.fragment.split("#")) + ) + 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: + self.machines.append(MachineData()) + + 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: - raise ClanError(f"Multiple values for parameter: {field.name}") - clan_params[field.name] = values[0] + raise ClanError(f"Multiple values for parameter: {dfield.name}") + machine_params[dfield.name] = values[0] # Remove the field from the query dictionary # clan uri and nested uri share one namespace for query parameters # we need to make sure there are no conflicts - del query[field.name] - # Reencode the query dictionary into a query string - real_query = urllib.parse.urlencode(query, doseq=True) + del query[dfield.name] + params = MachineParams(**machine_params) + machine = MachineData(name=machine_name, params=params) + return machine - # If the fragment contains a #, use the part after the # as the flake_attr - # on multiple #, use the first one - if self._components.fragment != "": - clan_params["flake_attr"] = self._components.fragment.split("#")[0] + def get_orig_uri(self) -> str: + return self._orig_uri - # Replace the query string in the components with the new query string - self._components = self._components._replace(query=real_query, fragment="") - - # Create a ClanParameters object from the clan_params dictionary - self.params = ClanParameters(**clan_params) - - 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): + def get_url(self) -> str: + match self.url: + case ClanUrl.LOCAL.value(path): return str(path) - case ClanScheme.REMOTE.value(url): + case ClanUrl.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 - def from_str( - cls, # noqa - url: str, - flake_attr: str | None = None, - params: dict[str, Any] | ClanParameters | None = None, - ) -> Self: - if flake_attr is not None and params is not None: - raise ClanError("flake_attr and params are mutually exclusive") - - prefix = "clan://" - if url.startswith(prefix): - url = url[len(prefix) :] - - if params is None and flake_attr is None: - return cls(f"clan://{url}") - - comp = urllib.parse.urlparse(url) - 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}") + raise ClanError(f"Unsupported uri components: {self.url}") def __str__(self) -> str: - return self.get_full_uri() + return self.get_orig_uri() def __repr__(self) -> str: - return f"ClanURI({self.get_full_uri()})" + return f"ClanURI({self})" diff --git a/pkgs/clan-cli/clan_cli/history/update.py b/pkgs/clan-cli/clan_cli/history/update.py index 9b176dea..9b13cbe3 100644 --- a/pkgs/clan-cli/clan_cli/history/update.py +++ b/pkgs/clan-cli/clan_cli/history/update.py @@ -4,7 +4,7 @@ import datetime from clan_cli.flakes.inspect import inspect_flake -from ..clan_uri import ClanParameters, ClanURI +from ..clan_uri import ClanURI, MachineParams from ..errors import ClanCmdError from ..locked_open import write_history_file from ..nix import nix_metadata @@ -28,9 +28,9 @@ def update_history() -> list[HistoryEntry]: ) uri = ClanURI.from_str( url=str(entry.flake.flake_url), - params=ClanParameters(entry.flake.flake_attr), + params=MachineParams(machine_name=entry.flake.flake_attr), ) - flake = inspect_flake(uri.get_internal(), uri.params.flake_attr) + flake = inspect_flake(uri.get_url(), uri.machines[0].name) flake.flake_url = str(flake.flake_url) entry = HistoryEntry( flake=flake, last_used=datetime.datetime.now().isoformat() diff --git a/pkgs/clan-cli/tests/test_clan_uri.py b/pkgs/clan-cli/tests/test_clan_uri.py index 64cc4632..6cfb0402 100644 --- a/pkgs/clan-cli/tests/test_clan_uri.py +++ b/pkgs/clan-cli/tests/test_clan_uri.py @@ -1,28 +1,28 @@ 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 - uri = ClanURI("clan://https://example.com?flake_attr=myVM&password=1234") - assert uri.get_internal() == "https://example.com?password=1234" + uri = ClanURI("clan://https://example.com?password=1234#myVM") + assert uri.get_url() == "https://example.com?password=1234" uri = ClanURI("clan://~/Downloads") - assert uri.get_internal().endswith("/Downloads") + assert uri.get_url().endswith("/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") - assert uri.get_internal() == "/home/user/Downloads" + assert uri.get_url() == "/home/user/Downloads" def test_local_uri() -> None: # Create a ClanURI object from a local URI uri = ClanURI("clan://file:///home/user/Downloads") - match uri.scheme: - case ClanScheme.LOCAL.value(path): + match uri.url: + case ClanUrl.LOCAL.value(path): assert path == Path("/home/user/Downloads") # type: ignore case _: assert False @@ -32,8 +32,8 @@ def test_is_remote() -> None: # Create a ClanURI object from a remote URI uri = ClanURI("clan://https://example.com") - match uri.scheme: - case ClanScheme.REMOTE.value(url): + match uri.url: + case ClanUrl.REMOTE.value(url): assert url == "https://example.com" # type: ignore case _: assert False @@ -42,104 +42,34 @@ def test_is_remote() -> None: def test_direct_local_path() -> None: # Create a ClanURI object from a remote URI uri = ClanURI("clan://~/Downloads") - assert uri.get_internal().endswith("/Downloads") + assert uri.get_url().endswith("/Downloads") def test_direct_local_path2() -> None: # Create a ClanURI object from a remote URI 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: # Create a ClanURI object from a remote URI with parameters uri = ClanURI("clan://https://example.com") - assert uri.params.flake_attr == "defaultVM" + assert uri.machines[0].name == "defaultVM" - match uri.scheme: - case ClanScheme.REMOTE.value(url): + match uri.url: + case ClanUrl.REMOTE.value(url): assert url == "https://example.com" # type: ignore case _: 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: - # Create a ClanURI object from a remote URI with parameters - uri = ClanURI("clan://https://example.com?flake_attr=myVM&password=1234") - assert uri.params.flake_attr == "myVM" - - match uri.scheme: - case ClanScheme.REMOTE.value(url): - assert url == "https://example.com?password=1234" # type: ignore + uri = ClanURI("clan://https://example.com?password=12345#myVM#secondVM") + assert uri.machines[0].name == "myVM" + assert uri.machines[1].name == "secondVM" + match uri.url: + case ClanUrl.REMOTE.value(url): + assert url == "https://example.com?password=12345" # type: ignore case _: assert False - - -def test_with_hashtag() -> None: - uri = ClanURI("clan://https://example.com?flake_attr=thirdVM#myVM#secondVM") - assert uri.params.flake_attr == "myVM" From 442e5b45badbcee36dce0b5b8e488af43ff18f8b Mon Sep 17 00:00:00 2001 From: Qubasa Date: Thu, 7 Mar 2024 19:04:48 +0700 Subject: [PATCH 2/2] clan_cli: Simplify ClanURI --- pkgs/clan-cli/clan_cli/clan_uri.py | 54 +++++++++++----- pkgs/clan-cli/clan_cli/history/add.py | 12 ++-- pkgs/clan-cli/clan_cli/history/update.py | 6 +- pkgs/clan-cli/tests/test_clan_uri.py | 61 +++++++++++++++++-- pkgs/clan-cli/tests/test_history_cli.py | 8 +-- .../clan_vm_manager/components/vmobj.py | 6 +- .../clan_vm_manager/singletons/use_join.py | 6 +- .../clan_vm_manager/singletons/use_vms.py | 9 +-- .../clan_vm_manager/views/list.py | 8 +-- 9 files changed, 123 insertions(+), 47 deletions(-) diff --git a/pkgs/clan-cli/clan_cli/clan_uri.py b/pkgs/clan-cli/clan_cli/clan_uri.py index 032a4922..20d53e4c 100644 --- a/pkgs/clan-cli/clan_cli/clan_uri.py +++ b/pkgs/clan-cli/clan_cli/clan_uri.py @@ -19,7 +19,10 @@ class ClanUrl(Enum): url: str # The url field holds the HTTP URL 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 @dataclass @@ -27,7 +30,10 @@ class ClanUrl(Enum): path: Path # The path field holds the local path 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 @@ -39,9 +45,13 @@ class MachineParams: @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 class ClanURI: @@ -49,11 +59,11 @@ class ClanURI: _nested_uri: str _components: urllib.parse.ParseResult url: ClanUrl - machines: list[MachineData] + _machines: list[MachineData] # Initialize the class with a clan:// URI def __init__(self, uri: str) -> None: - self.machines = [] + self._machines = [] # users might copy whitespace along with the uri uri = uri.strip() @@ -85,11 +95,12 @@ class ClanURI: ) for machine_frag in machine_frags: machine = self._parse_machine_query(machine_frag) - self.machines.append(machine) + self._machines.append(machine) # If there are no machine fragments, add a default machine if len(machine_frags) == 0: - self.machines.append(MachineData()) + default_machine = MachineData(url=self.url) + self._machines.append(default_machine) def _parse_url(self, comps: urllib.parse.ParseResult) -> ClanUrl: comb = ( @@ -126,20 +137,35 @@ class ClanURI: # we need to make sure there are no conflicts del query[dfield.name] params = MachineParams(**machine_params) - machine = MachineData(name=machine_name, params=params) + machine = MachineData(url=self.url, name=machine_name, params=params) return machine + @property + def machine(self) -> MachineData: + return self._machines[0] + def get_orig_uri(self) -> str: return self._orig_uri def get_url(self) -> str: - match self.url: - case ClanUrl.LOCAL.value(path): - return str(path) - case ClanUrl.REMOTE.value(url): - return url - case _: - raise ClanError(f"Unsupported uri components: {self.url}") + return str(self.url) + + @classmethod + def from_str( + cls, # noqa + url: str, + machine_name: str | None = None, + ) -> "ClanURI": + clan_uri = "" + if not url.startswith("clan://"): + clan_uri += "clan://" + + clan_uri += url + + if machine_name: + clan_uri += f"#{machine_name}" + + return cls(clan_uri) def __str__(self) -> str: return self.get_orig_uri() diff --git a/pkgs/clan-cli/clan_cli/history/add.py b/pkgs/clan-cli/clan_cli/history/add.py index 16bb9698..3b4b21dd 100644 --- a/pkgs/clan-cli/clan_cli/history/add.py +++ b/pkgs/clan-cli/clan_cli/history/add.py @@ -79,8 +79,8 @@ def new_history_entry(url: str, machine: str) -> HistoryEntry: def add_all_to_history(uri: ClanURI) -> list[HistoryEntry]: history = list_history() new_entries: list[HistoryEntry] = [] - for machine in list_machines(uri.get_internal()): - new_entry = _add_maschine_to_history_list(uri.get_internal(), machine, history) + for machine in list_machines(uri.get_url()): + new_entry = _add_maschine_to_history_list(uri.get_url(), machine, history) new_entries.append(new_entry) write_history_file(history) return new_entries @@ -89,9 +89,7 @@ def add_all_to_history(uri: ClanURI) -> list[HistoryEntry]: def add_history(uri: ClanURI) -> HistoryEntry: user_history_file().parent.mkdir(parents=True, exist_ok=True) history = list_history() - new_entry = _add_maschine_to_history_list( - uri.get_internal(), uri.params.flake_attr, history - ) + new_entry = _add_maschine_to_history_list(uri.get_url(), uri.machine.name, history) write_history_file(history) return new_entry @@ -121,9 +119,7 @@ def add_history_command(args: argparse.Namespace) -> None: # takes a (sub)parser and configures it def register_add_parser(parser: argparse.ArgumentParser) -> None: - parser.add_argument( - "uri", type=ClanURI.from_str, help="Path to the flake", default="." - ) + parser.add_argument("uri", type=ClanURI, help="Path to the flake", default=".") parser.add_argument( "--all", help="Add all machines", default=False, action="store_true" ) diff --git a/pkgs/clan-cli/clan_cli/history/update.py b/pkgs/clan-cli/clan_cli/history/update.py index 9b13cbe3..12ecf24e 100644 --- a/pkgs/clan-cli/clan_cli/history/update.py +++ b/pkgs/clan-cli/clan_cli/history/update.py @@ -4,7 +4,7 @@ import datetime from clan_cli.flakes.inspect import inspect_flake -from ..clan_uri import ClanURI, MachineParams +from ..clan_uri import ClanURI from ..errors import ClanCmdError from ..locked_open import write_history_file from ..nix import nix_metadata @@ -28,9 +28,9 @@ def update_history() -> list[HistoryEntry]: ) uri = ClanURI.from_str( url=str(entry.flake.flake_url), - params=MachineParams(machine_name=entry.flake.flake_attr), + machine_name=entry.flake.flake_attr, ) - flake = inspect_flake(uri.get_url(), uri.machines[0].name) + flake = inspect_flake(uri.get_url(), uri.machine.name) flake.flake_url = str(flake.flake_url) entry = HistoryEntry( flake=flake, last_used=datetime.datetime.now().isoformat() diff --git a/pkgs/clan-cli/tests/test_clan_uri.py b/pkgs/clan-cli/tests/test_clan_uri.py index 6cfb0402..f8c09cb7 100644 --- a/pkgs/clan-cli/tests/test_clan_uri.py +++ b/pkgs/clan-cli/tests/test_clan_uri.py @@ -55,7 +55,7 @@ def test_remote_with_clanparams() -> None: # Create a ClanURI object from a remote URI with parameters uri = ClanURI("clan://https://example.com") - assert uri.machines[0].name == "defaultVM" + assert uri.machine.name == "defaultVM" match uri.url: case ClanUrl.REMOTE.value(url): @@ -65,11 +65,64 @@ def test_remote_with_clanparams() -> None: def test_remote_with_all_params() -> None: - uri = ClanURI("clan://https://example.com?password=12345#myVM#secondVM") - assert uri.machines[0].name == "myVM" - assert uri.machines[1].name == "secondVM" + uri = ClanURI("clan://https://example.com?password=12345#myVM#secondVM?dummy_opt=1") + assert uri.machine.name == "myVM" + assert uri._machines[1].name == "secondVM" + assert uri._machines[1].params.dummy_opt == "1" match uri.url: case ClanUrl.REMOTE.value(url): assert url == "https://example.com?password=12345" # type: ignore case _: assert False + + +def test_from_str_remote() -> None: + uri = ClanURI.from_str(url="https://example.com", machine_name="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 diff --git a/pkgs/clan-cli/tests/test_history_cli.py b/pkgs/clan-cli/tests/test_history_cli.py index d5eb34bf..7da808ed 100644 --- a/pkgs/clan-cli/tests/test_history_cli.py +++ b/pkgs/clan-cli/tests/test_history_cli.py @@ -6,7 +6,7 @@ from cli import Cli from fixtures_flakes import FlakeForTest 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.history.add import HistoryEntry @@ -19,8 +19,7 @@ def test_history_add( test_flake_with_core: FlakeForTest, ) -> None: cli = Cli() - params = ClanParameters(flake_attr="vm1") - uri = ClanURI.from_path(test_flake_with_core.path, params=params) + uri = ClanURI.from_str(str(test_flake_with_core.path), "vm1") cmd = [ "history", "add", @@ -40,8 +39,7 @@ def test_history_list( test_flake_with_core: FlakeForTest, ) -> None: cli = Cli() - params = ClanParameters(flake_attr="vm1") - uri = ClanURI.from_path(test_flake_with_core.path, params=params) + uri = ClanURI.from_str(str(test_flake_with_core.path), "vm1") cmd = [ "history", "list", diff --git a/pkgs/clan-vm-manager/clan_vm_manager/components/vmobj.py b/pkgs/clan-vm-manager/clan_vm_manager/components/vmobj.py index a713d643..8b350d17 100644 --- a/pkgs/clan-vm-manager/clan_vm_manager/components/vmobj.py +++ b/pkgs/clan-vm-manager/clan_vm_manager/components/vmobj.py @@ -13,7 +13,7 @@ from typing import IO, ClassVar import gi 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.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 ) match uri.scheme: - case ClanScheme.LOCAL.value(path): + case ClanUrl.LOCAL.value(path): self.machine = Machine( name=self.data.flake.flake_attr, flake=path, # type: ignore ) - case ClanScheme.REMOTE.value(url): + case ClanUrl.REMOTE.value(url): self.machine = Machine( name=self.data.flake.flake_attr, flake=url, # type: ignore diff --git a/pkgs/clan-vm-manager/clan_vm_manager/singletons/use_join.py b/pkgs/clan-vm-manager/clan_vm_manager/singletons/use_join.py index b8d9963d..3794f6a4 100644 --- a/pkgs/clan-vm-manager/clan_vm_manager/singletons/use_join.py +++ b/pkgs/clan-vm-manager/clan_vm_manager/singletons/use_join.py @@ -62,8 +62,8 @@ class JoinList: cls._instance = cls.__new__(cls) 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) + return cls._instance def _rerender_join_list( @@ -83,7 +83,9 @@ class JoinList: """ 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.") return diff --git a/pkgs/clan-vm-manager/clan_vm_manager/singletons/use_vms.py b/pkgs/clan-vm-manager/clan_vm_manager/singletons/use_vms.py index 8ac60d60..50072c59 100644 --- a/pkgs/clan-vm-manager/clan_vm_manager/singletons/use_vms.py +++ b/pkgs/clan-vm-manager/clan_vm_manager/singletons/use_vms.py @@ -57,7 +57,7 @@ class ClanStore: store: "GKVStore", position: int, removed: int, added: int ) -> None: if added > 0: - store.register_on_change(on_vmstore_change) + store.values()[position].register_on_change(on_vmstore_change) callback(store, position, removed, added) 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] def get_vm(self, uri: ClanURI) -> None | VMObject: - clan = self.clan_store.get(uri.get_internal()) - if clan is None: + vm_store = self.clan_store.get(str(uri.url)) + if vm_store is 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]: return [ diff --git a/pkgs/clan-vm-manager/clan_vm_manager/views/list.py b/pkgs/clan-vm-manager/clan_vm_manager/views/list.py index f5c4e535..0775f40b 100644 --- a/pkgs/clan-vm-manager/clan_vm_manager/views/list.py +++ b/pkgs/clan-vm-manager/clan_vm_manager/views/list.py @@ -190,8 +190,8 @@ class ClanList(Gtk.Box): log.debug("Rendering join row for %s", join_val.url) row = Adw.ActionRow() - row.set_title(join_val.url.params.flake_attr) - row.set_subtitle(join_val.url.get_internal()) + row.set_title(join_val.url.machine.name) + row.set_subtitle(str(join_val.url)) row.add_css_class("trust") vm = ClanStore.use().get_vm(join_val.url) @@ -204,7 +204,7 @@ class ClanList(Gtk.Box): ) 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_size(50) row.add_prefix(avatar) @@ -229,7 +229,7 @@ class ClanList(Gtk.Box): def on_join_request(self, source: Any, url: str) -> None: 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) def on_after_join(self, source: JoinValue) -> None: