clan_cli: ClanURI fixed non executing tests
This commit is contained in:
@@ -4,7 +4,7 @@ import urllib.parse
|
|||||||
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 Dict
|
from typing import Dict, Self
|
||||||
|
|
||||||
from .errors import ClanError
|
from .errors import ClanError
|
||||||
|
|
||||||
@@ -41,8 +41,7 @@ class ClanScheme(Enum):
|
|||||||
# 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 ClanParameters:
|
||||||
flake_attr: str | None
|
flake_attr: str = "defaultVM"
|
||||||
machine: str | None
|
|
||||||
|
|
||||||
|
|
||||||
# Define the ClanURI class
|
# Define the ClanURI class
|
||||||
@@ -60,30 +59,25 @@ class ClanURI:
|
|||||||
self._components = urllib.parse.urlparse(self._nested_uri)
|
self._components = urllib.parse.urlparse(self._nested_uri)
|
||||||
|
|
||||||
# Parse the query string into a dictionary
|
# Parse the query string into a dictionary
|
||||||
self._query = urllib.parse.parse_qs(self._components.query)
|
query = urllib.parse.parse_qs(self._components.query)
|
||||||
|
|
||||||
params: Dict[str, str | None] = {}
|
params: Dict[str, str] = {}
|
||||||
for field in dataclasses.fields(ClanParameters):
|
for field in dataclasses.fields(ClanParameters):
|
||||||
if field.name in self._query:
|
if field.name in query:
|
||||||
# Check if the field type is a list
|
values = query[field.name]
|
||||||
if issubclass(field.type, list):
|
if len(values) > 1:
|
||||||
setattr(params, field.name, self._query[field.name])
|
raise ClanError(
|
||||||
# Check if the field type is a single value
|
"Multiple values for parameter: {}".format(field.name)
|
||||||
else:
|
)
|
||||||
values = self._query[field.name]
|
params[field.name] = values[0]
|
||||||
if len(values) > 1:
|
|
||||||
raise ClanError(
|
|
||||||
"Multiple values for parameter: {}".format(field.name)
|
|
||||||
)
|
|
||||||
setattr(params, field.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 self._query[field.name]
|
del query[field.name]
|
||||||
else:
|
|
||||||
params[field.name] = None
|
|
||||||
|
|
||||||
|
new_query = urllib.parse.urlencode(query, doseq=True)
|
||||||
|
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
|
# Use the match statement to check the scheme and create a ClanScheme member with the value
|
||||||
@@ -98,3 +92,9 @@ class ClanURI:
|
|||||||
raise ClanError(
|
raise ClanError(
|
||||||
"Unsupported scheme: {}".format(self._components.scheme)
|
"Unsupported scheme: {}".format(self._components.scheme)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_path(cls, path: Path, params: ClanParameters) -> Self: # noqa
|
||||||
|
urlparams = urllib.parse.urlencode(params.__dict__)
|
||||||
|
|
||||||
|
return cls("clan://file://{}?{}".format(path, urlparams))
|
||||||
|
|||||||
@@ -33,9 +33,9 @@ def test_is_remote() -> None:
|
|||||||
assert False
|
assert False
|
||||||
|
|
||||||
|
|
||||||
def 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?flake_attr=defaultVM")
|
uri = ClanURI("clan://https://example.com")
|
||||||
|
|
||||||
assert uri.params.flake_attr == "defaultVM"
|
assert uri.params.flake_attr == "defaultVM"
|
||||||
|
|
||||||
@@ -46,17 +46,13 @@ def remote_with_clanparams() -> None:
|
|||||||
assert False
|
assert False
|
||||||
|
|
||||||
|
|
||||||
def 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(
|
uri = ClanURI("clan://https://example.com?flake_attr=myVM&password=1234")
|
||||||
"clan://https://example.com?flake_attr=defaultVM&machine=vm1&password=1234"
|
assert uri.params.flake_attr == "myVM"
|
||||||
)
|
|
||||||
|
|
||||||
assert uri.params.flake_attr == "defaultVM"
|
|
||||||
assert uri.params.machine == "vm1"
|
|
||||||
|
|
||||||
match uri.scheme:
|
match uri.scheme:
|
||||||
case ClanScheme.HTTPS.value(url):
|
case ClanScheme.HTTPS.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
|
||||||
|
|||||||
@@ -68,9 +68,11 @@ class VMBase:
|
|||||||
vm = asyncio.run(
|
vm = asyncio.run(
|
||||||
vms.run.inspect_vm(flake_url=self._path, flake_attr="defaultVM")
|
vms.run.inspect_vm(flake_url=self._path, flake_attr="defaultVM")
|
||||||
)
|
)
|
||||||
task = vms.run.run_vm(vm)
|
vms.run.run_vm(vm)
|
||||||
for line in task.log_lines():
|
|
||||||
print(line, end="")
|
|
||||||
|
# for line in task.log_lines():
|
||||||
|
# print(line, end="")
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
|
|||||||
Reference in New Issue
Block a user