clan_cli: URI parser now only has HTTP and FILE. Also clan:///home/user or clan://~/Downloads is supported
This commit is contained in:
@@ -20,14 +20,6 @@ class ClanScheme(Enum):
|
|||||||
def __str__(self) -> str:
|
def __str__(self) -> str:
|
||||||
return f"HTTP({self.url})" # The __str__ method returns a custom string representation
|
return f"HTTP({self.url})" # The __str__ method returns a custom string representation
|
||||||
|
|
||||||
@member
|
|
||||||
@dataclass
|
|
||||||
class HTTPS:
|
|
||||||
url: str # The url field holds the HTTPS URL
|
|
||||||
|
|
||||||
def __str__(self) -> str:
|
|
||||||
return f"HTTPS({self.url})" # The __str__ method returns a custom string representation
|
|
||||||
|
|
||||||
@member
|
@member
|
||||||
@dataclass
|
@dataclass
|
||||||
class FILE:
|
class FILE:
|
||||||
@@ -78,22 +70,30 @@ class ClanURI:
|
|||||||
self._components = self._components._replace(query=new_query)
|
self._components = self._components._replace(query=new_query)
|
||||||
self.params = ClanParameters(**params)
|
self.params = ClanParameters(**params)
|
||||||
|
|
||||||
# Use the match statement to check the scheme and create a ClanScheme member with the value
|
comb = (
|
||||||
match self._components.scheme:
|
self._components.scheme,
|
||||||
case "http":
|
self._components.netloc,
|
||||||
|
self._components.path,
|
||||||
|
self._components.params,
|
||||||
|
self._components.query,
|
||||||
|
self._components.fragment,
|
||||||
|
)
|
||||||
|
|
||||||
|
match comb:
|
||||||
|
case ("http" | "https", _, _, _, _, _):
|
||||||
self.scheme = ClanScheme.HTTP.value(self._components.geturl()) # type: ignore
|
self.scheme = ClanScheme.HTTP.value(self._components.geturl()) # type: ignore
|
||||||
case "https":
|
case ("file", "", path, "", "", "") | ("", "", path, "", "", ""): # type: ignore
|
||||||
self.scheme = ClanScheme.HTTPS.value(self._components.geturl()) # type: ignore
|
self.scheme = ClanScheme.FILE.value(Path(path)) # type: ignore
|
||||||
case "file":
|
|
||||||
self.scheme = ClanScheme.FILE.value(Path(self._components.path)) # type: ignore
|
|
||||||
case _:
|
case _:
|
||||||
raise ClanError(f"Unsupported scheme: {self._components.scheme}")
|
raise ClanError(f"Unsupported uri components: {comb}")
|
||||||
|
|
||||||
|
def get_internal(self) -> str:
|
||||||
|
return self._nested_uri
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_path(cls, path: Path, params: ClanParameters) -> Self: # noqa
|
def from_path(cls, path: Path, params: ClanParameters) -> Self: # noqa
|
||||||
urlparams = urllib.parse.urlencode(params.__dict__)
|
urlparams = urllib.parse.urlencode(params.__dict__)
|
||||||
|
return cls(f"clan://{path}?{urlparams}")
|
||||||
return cls(f"clan://file://{path}?{urlparams}")
|
|
||||||
|
|
||||||
def __str__(self) -> str:
|
def __str__(self) -> str:
|
||||||
return f"ClanURI({self._components.geturl()})"
|
return f"ClanURI({self._components.geturl()})"
|
||||||
|
|||||||
@@ -82,6 +82,13 @@ def nix_eval(flags: list[str]) -> list[str]:
|
|||||||
return default_flags + flags
|
return default_flags + flags
|
||||||
|
|
||||||
|
|
||||||
|
def nix_metadata(flake: str) -> dict[str, Any]:
|
||||||
|
cmd = nix_command(["flake", "metadata", "--json", flake])
|
||||||
|
proc = subprocess.run(cmd, check=True, text=True, stdout=subprocess.PIPE)
|
||||||
|
data = json.loads(proc.stdout)
|
||||||
|
return data
|
||||||
|
|
||||||
|
|
||||||
@deal.raises(ClanError)
|
@deal.raises(ClanError)
|
||||||
def nix_shell(packages: list[str], cmd: list[str]) -> list[str]:
|
def nix_shell(packages: list[str], cmd: list[str]) -> list[str]:
|
||||||
# we cannot use nix-shell inside the nix sandbox
|
# we cannot use nix-shell inside the nix sandbox
|
||||||
|
|||||||
@@ -2,7 +2,7 @@ from pathlib import Path
|
|||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from clan_cli.clan_uri import ClanScheme, ClanURI
|
from clan_cli.clan_uri import ClanParameters, ClanScheme, ClanURI
|
||||||
from clan_cli.errors import ClanError
|
from clan_cli.errors import ClanError
|
||||||
|
|
||||||
|
|
||||||
@@ -17,7 +17,7 @@ def test_local_uri() -> None:
|
|||||||
|
|
||||||
|
|
||||||
def test_unsupported_schema() -> None:
|
def test_unsupported_schema() -> None:
|
||||||
with pytest.raises(ClanError, match="Unsupported scheme: ftp"):
|
with pytest.raises(ClanError, match="Unsupported uri components: .*"):
|
||||||
# Create a ClanURI object from an unsupported URI
|
# Create a ClanURI object from an unsupported URI
|
||||||
ClanURI("clan://ftp://ftp.example.com")
|
ClanURI("clan://ftp://ftp.example.com")
|
||||||
|
|
||||||
@@ -27,12 +27,24 @@ def test_is_remote() -> None:
|
|||||||
uri = ClanURI("clan://https://example.com")
|
uri = ClanURI("clan://https://example.com")
|
||||||
|
|
||||||
match uri.scheme:
|
match uri.scheme:
|
||||||
case ClanScheme.HTTPS.value(url):
|
case ClanScheme.HTTP.value(url):
|
||||||
assert url == "https://example.com" # type: ignore
|
assert url == "https://example.com" # type: ignore
|
||||||
case _:
|
case _:
|
||||||
assert False
|
assert False
|
||||||
|
|
||||||
|
|
||||||
|
def test_direct_local_path() -> None:
|
||||||
|
# Create a ClanURI object from a remote URI
|
||||||
|
uri = ClanURI("clan://~/Downloads")
|
||||||
|
assert uri.get_internal() == "~/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"
|
||||||
|
|
||||||
|
|
||||||
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")
|
||||||
@@ -40,19 +52,47 @@ def test_remote_with_clanparams() -> None:
|
|||||||
assert uri.params.flake_attr == "defaultVM"
|
assert uri.params.flake_attr == "defaultVM"
|
||||||
|
|
||||||
match uri.scheme:
|
match uri.scheme:
|
||||||
case ClanScheme.HTTPS.value(url):
|
case ClanScheme.HTTP.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)
|
||||||
|
assert uri.params.flake_attr == "myVM"
|
||||||
|
|
||||||
|
match uri.scheme:
|
||||||
|
case ClanScheme.FILE.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)
|
||||||
|
assert uri.params.flake_attr == "defaultVM"
|
||||||
|
|
||||||
|
match uri.scheme:
|
||||||
|
case ClanScheme.FILE.value(path):
|
||||||
|
assert path == Path("/home/user/Downloads") # type: ignore
|
||||||
|
case _:
|
||||||
|
assert False
|
||||||
|
|
||||||
|
|
||||||
def test_remote_with_all_params() -> None:
|
def test_remote_with_all_params() -> 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?flake_attr=myVM&password=1234")
|
||||||
assert uri.params.flake_attr == "myVM"
|
assert uri.params.flake_attr == "myVM"
|
||||||
|
|
||||||
match uri.scheme:
|
match uri.scheme:
|
||||||
case ClanScheme.HTTPS.value(url):
|
case ClanScheme.HTTP.value(url):
|
||||||
assert url == "https://example.com?password=1234" # type: ignore
|
assert url == "https://example.com?password=1234" # type: ignore
|
||||||
case _:
|
case _:
|
||||||
assert False
|
assert False
|
||||||
|
|||||||
@@ -89,12 +89,14 @@ class ClanList(Gtk.Box):
|
|||||||
remount_edit: Callable[[], None],
|
remount_edit: Callable[[], None],
|
||||||
set_selected: Callable[[VMBase | None], None],
|
set_selected: Callable[[VMBase | None], None],
|
||||||
selected_vm: VMBase | None,
|
selected_vm: VMBase | None,
|
||||||
|
show_toolbar: bool = True,
|
||||||
) -> None:
|
) -> None:
|
||||||
super().__init__(orientation=Gtk.Orientation.VERTICAL, expand=True)
|
super().__init__(orientation=Gtk.Orientation.VERTICAL, expand=True)
|
||||||
|
|
||||||
self.remount_edit_view = remount_edit
|
self.remount_edit_view = remount_edit
|
||||||
self.remount_list_view = remount_list
|
self.remount_list_view = remount_list
|
||||||
self.set_selected = set_selected
|
self.set_selected = set_selected
|
||||||
|
self.show_toolbar = show_toolbar
|
||||||
|
|
||||||
# TODO: We should use somekind of useState hook here.
|
# TODO: We should use somekind of useState hook here.
|
||||||
# that updates the list of VMs when the user changes something
|
# that updates the list of VMs when the user changes something
|
||||||
@@ -109,9 +111,10 @@ class ClanList(Gtk.Box):
|
|||||||
"on_stop_clicked": self.on_stop_clicked,
|
"on_stop_clicked": self.on_stop_clicked,
|
||||||
"on_edit_clicked": self.on_edit_clicked,
|
"on_edit_clicked": self.on_edit_clicked,
|
||||||
}
|
}
|
||||||
self.toolbar = ClanListToolbar(**button_hooks)
|
if show_toolbar:
|
||||||
self.toolbar.set_is_selected(self.selected_vm is not None)
|
self.toolbar = ClanListToolbar(**button_hooks)
|
||||||
self.add(self.toolbar)
|
self.toolbar.set_is_selected(self.selected_vm is not None)
|
||||||
|
self.add(self.toolbar)
|
||||||
|
|
||||||
self.list_hooks = {
|
self.list_hooks = {
|
||||||
"on_select_row": self.on_select_vm,
|
"on_select_row": self.on_select_vm,
|
||||||
@@ -134,10 +137,11 @@ class ClanList(Gtk.Box):
|
|||||||
|
|
||||||
def on_select_vm(self, vm: VMBase) -> None:
|
def on_select_vm(self, vm: VMBase) -> None:
|
||||||
print(f"on_select_vm: {vm.name}")
|
print(f"on_select_vm: {vm.name}")
|
||||||
if vm is None:
|
if self.show_toolbar:
|
||||||
self.toolbar.set_is_selected(False)
|
if vm is None:
|
||||||
else:
|
self.toolbar.set_is_selected(False)
|
||||||
self.toolbar.set_is_selected(True)
|
else:
|
||||||
|
self.toolbar.set_is_selected(True)
|
||||||
|
|
||||||
self.set_selected(vm)
|
self.set_selected(vm)
|
||||||
self.selected_vm = vm
|
self.selected_vm = vm
|
||||||
|
|||||||
Reference in New Issue
Block a user