Merge pull request 'clan_cli: Remodeled ClanURI parser' (#610) from Qubasa-main into main

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)
+                )

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