diff --git a/pkgs/clan-cli/clan_cli/clan_uri.py b/pkgs/clan-cli/clan_cli/clan_uri.py index f140474f..f0a8121f 100644 --- a/pkgs/clan-cli/clan_cli/clan_uri.py +++ b/pkgs/clan-cli/clan_cli/clan_uri.py @@ -1,50 +1,100 @@ -# Import the urllib.parse module +# Import the urllib.parse, enum and dataclasses modules +import dataclasses import urllib.parse -from enum import Enum +from dataclasses import dataclass +from enum import Enum, member +from pathlib import Path +from typing import Dict from .errors import ClanError +# Define an enum with different members that have different values class ClanScheme(Enum): - HTTP = "http" - HTTPS = "https" - FILE = "file" + # Use the dataclass decorator to add fields and methods to the members + @member + @dataclass + class HTTP: + url: str # The url field holds the HTTP URL + + def __str__(self) -> str: + 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 + @dataclass + class FILE: + path: Path # The path field holds the local path + + def __str__(self) -> str: + return f"FILE({self.path})" # The __str__ method returns a custom string representation + + +# 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 | None + machine: str | None # Define the ClanURI class class ClanURI: # Initialize the class with a clan:// URI def __init__(self, uri: str) -> None: + # Check if the URI starts with clan:// if uri.startswith("clan://"): - uri = uri[7:] + self._nested_uri = uri[7:] else: raise ClanError("Invalid scheme: expected clan://, got {}".format(uri)) # Parse the URI into components - self.components = urllib.parse.urlparse(uri) + # scheme://netloc/path;parameters?query#fragment + self._components = urllib.parse.urlparse(self._nested_uri) - try: - self.scheme = ClanScheme(self.components.scheme) - except ValueError: - raise ClanError("Unsupported scheme: {}".format(self.components.scheme)) + # Parse the query string into a dictionary + self._query = urllib.parse.parse_qs(self._components.query) - # Define a method to get the path of the URI - @property - def path(self) -> str: - return self.components.path + params: Dict[str, str | None] = {} + for field in dataclasses.fields(ClanParameters): + if field.name in self._query: + # Check if the field type is a list + if issubclass(field.type, list): + setattr(params, field.name, self._query[field.name]) + # Check if the field type is a single value + else: + values = self._query[field.name] + if len(values) > 1: + raise ClanError( + "Multiple values for parameter: {}".format(field.name) + ) + setattr(params, field.name, values[0]) - @property - def url(self) -> str: - return self.components.geturl() + # 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 self._query[field.name] + else: + params[field.name] = None - # Define a method to check if the URI is a remote HTTP URL - def is_remote(self) -> bool: - match self.scheme: - case ClanScheme.HTTP | ClanScheme.HTTPS: - return True - case ClanScheme.FILE: - return False + self.params = ClanParameters(**params) - # Define a method to check if the URI is a local path - def is_local(self) -> bool: - return not self.is_remote() + # Use the match statement to check the scheme and create a ClanScheme member with the value + match self._components.scheme: + case "http": + self.scheme = ClanScheme.HTTP.value(self._components.geturl()) # type: ignore + case "https": + self.scheme = ClanScheme.HTTPS.value(self._components.geturl()) # type: ignore + case "file": + self.scheme = ClanScheme.FILE.value(Path(self._components.path)) # type: ignore + case _: + raise ClanError( + "Unsupported scheme: {}".format(self._components.scheme) + ) diff --git a/pkgs/clan-cli/tests/test_clan_uri.py b/pkgs/clan-cli/tests/test_clan_uri.py index 72588b4e..3a4aac59 100644 --- a/pkgs/clan-cli/tests/test_clan_uri.py +++ b/pkgs/clan-cli/tests/test_clan_uri.py @@ -1,18 +1,19 @@ +from pathlib import Path + import pytest -from clan_cli.clan_uri import ClanURI +from clan_cli.clan_uri import ClanScheme, ClanURI from clan_cli.errors import ClanError def test_local_uri() -> None: # Create a ClanURI object from a local URI uri = ClanURI("clan://file:///home/user/Downloads") - # Check that the URI is local - assert uri.is_local() - # Check that the URI is not remote - assert not uri.is_remote() - # Check that the URI path is correct - assert uri.path == "/home/user/Downloads" + match uri.scheme: + case ClanScheme.FILE.value(path): + assert path == Path("/home/user/Downloads") # type: ignore + case _: + assert False def test_unsupported_schema() -> None: @@ -24,11 +25,38 @@ def test_unsupported_schema() -> None: def test_is_remote() -> None: # Create a ClanURI object from a remote URI uri = ClanURI("clan://https://example.com") - # Check that the URI is remote - assert uri.is_remote() - # Check that the URI is not local - assert not uri.is_local() - # Check that the URI path is correct - assert uri.path == "" - assert uri.url == "https://example.com" + match uri.scheme: + case ClanScheme.HTTPS.value(url): + assert url == "https://example.com" # type: ignore + case _: + assert False + + +def remote_with_clanparams() -> None: + # Create a ClanURI object from a remote URI with parameters + uri = ClanURI("clan://https://example.com?flake_attr=defaultVM") + + assert uri.params.flake_attr == "defaultVM" + + match uri.scheme: + case ClanScheme.HTTPS.value(url): + assert url == "https://example.com" # type: ignore + case _: + assert False + + +def remote_with_all_params() -> None: + # Create a ClanURI object from a remote URI with parameters + uri = ClanURI( + "clan://https://example.com?flake_attr=defaultVM&machine=vm1&password=1234" + ) + + assert uri.params.flake_attr == "defaultVM" + assert uri.params.machine == "vm1" + + match uri.scheme: + case ClanScheme.HTTPS.value(url): + assert url == "https://example.com&password=1234" # type: ignore + case _: + assert False